Cloudbase

Standalone Install Note

If this environment only installed the current skill, start from the CloudBase main entry and use the published cloudbase/references/... paths for sibling skills.

• CloudBase main entry: https://cnb.cool/tencent/cloud/cloudbase/cloudbase-skills/-/git/raw/main/skills/cloudbase/SKILL.md
• Current skill raw source: https://cnb.cool/tencent/cloud/cloudbase/cloudbase-skills/-/git/raw/main/skills/cloudbase/references/ai-model-nodejs/SKILL.md

Keep local references/... paths for files that ship with the current skill directory. When this file points to a sibling skill such as auth-tool or web-development, use the standalone fallback URL shown next to that reference.

When to use this skill

Use this skill for calling AI models from Node.js backends, cloud functions, or CloudRun services via @cloudbase/node-sdk.

🧭 Runtime-plane fit. This is the right skill when the AI call truly belongs on the server: image generation (the only SDK that supports it), long-running agent jobs, orchestration across multiple tools, scheduled tasks, or flows that must keep secrets server-side. If the user is building a Web page / frontend AI chat UI, do NOT wrap this SDK behind a backend proxy — route to ai-model-web and call the model directly from the browser. For WeChat Mini Programs use ai-model-wechat. Routing is decided by runtime plane first; the concrete model (deepseek-*, glm-*, hunyuan-*, kimi-*, …) only affects the model field.

Use it when you need to:

• Integrate AI text generation into a backend service
• Generate images with the Hunyuan Image model
• Call AI models from CloudBase cloud functions or CloudRun
• Do server-side AI processing (agent orchestration, batch jobs, scheduled tasks)

Do NOT use for:

• Browser/Web apps → use the ai-model-web skill
• WeChat Mini Program → use the ai-model-wechat skill
• Runtimes without a CloudBase SDK (Python, Go, PHP, curl, etc.) → use the http-api skill (it now includes the ai_model OpenAPI spec for direct HTTP calls to the AI model endpoint; do NOT wrap this SDK behind an HTTP proxy)

---

⛔ STOP — ai.createModel(...) argument is not a vendor / model name

Read this before writing any createModel(...) line. Agents frequently hallucinate this argument. There are exactly three legal shapes. Anything else is a bug.

Image generation is a separate entry point: ai.createImageModel("hunyuan-image"). Do not mix it with createModel(...).

❌ Do NOT write any of these — they are all wrong

ai.createModel("deepseek")                 // wrong — that's a vendor, not a GroupName
ai.createModel("deepseek-v4-flash")        // wrong — model id goes in the `model` field
ai.createModel("hunyuan") / "hunyuan-2.0-instruct-20251111"  // wrong — vendor / model name
ai.createModel("glm") / "kimi" / "minimax"  // wrong — vendor names
ai.createModel("openai") / "moonshot"       // wrong — vendor names
ai.createModel("custom")                   // wrong — placeholder; use your real custom-<name>
ai.createModel(modelName)                  // wrong — do not reuse the variable that holds the model id

✅ Correct pattern — GroupName vs Model are two different fields

const model = ai.createModel("cloudbase");          // ← GroupName
await model.generateText({
  model: "deepseek-v4-flash",                       // ← concrete model id
  messages: [...]
});

Decision procedure (when the user names a specific model)

1. The user says "use DeepSeek v3.2" / "use hunyuan instruct" / "use Kimi k2.6" / "use GLM-5" / … 2. createModel("cloudbase") stays the same. 3. Put the model id into the `model` field: { model: "deepseek-v3.2" }, { model: "hunyuan-2.0-instruct-20251111" }, { model: "kimi-k2.6" }, { model: "glm-5" }, … 4. Never assume the model is already enabled. Before calling the SDK, verify it is present in DescribeAIModels({ GroupName: "cloudbase" }).Models[]. If missing, call DescribeManagedAIModelList to confirm the exact Model name the platform supports (case-sensitive — do not guess the spelling) and then enable it via UpdateAIModel with Status: 1 (remember Models is a full replacement).

If you are about to type ai.createModel( and the thing inside the parentheses is a vendor name, a model name, or a guess — stop. It is almost certainly one of the three legal values above.

---

Mandatory Two-Step Preflight (before any SDK code)

Before calling any AI API on the server, run the two-step preflight: ① eligibility, ② group readiness. Text generation and image generation draw from the same Token Credits resource pack, and both must complete the preflight before code is emitted.

Step 0: obtain the environment ID

Call the MCP tool envQuery with action=info and read EnvId from the response.

---

Preflight ① — Eligibility (Token Credits resource pack)

Call the MCP tool:

callCloudApi(service="tcb", action="DescribeEnvPostpayPackage", params={ EnvId })

Pass conditions (all required):

• envPostpayPackageInfoList contains at least one entry
• That entry's postpayPackageId starts with pkg_tcb_tokencredits_
• That entry's status is NOT in [3, 4] (3 / 4 typically mean expired / disabled; trust the live response)

• ❌ Not satisfied → stop writing code and surface this to the user (replacing {envId} with the real id):

The current environment has no active Token Credits resource pack. Please purchase one before calling any AI API:

https://buy.cloud.tencent.com/lowcode?buyType=resPack&envId={envId}&resourceType=token

>

Let me know once it's done and I'll re-check the resource pack status.

• ✅ Satisfied → proceed to preflight ②.

Parameter casing is PascalCase by contract. If the call returns InvalidParameter, fall back to camelCase (envId) and trust the live response.

---

Preflight ② — Group readiness (DescribeAIModels → UpdateAIModel if needed)

Eligibility alone is not enough. Do not write `createModel("cloudbase")` yet. First confirm that the target GroupName exists in the env with Status=1, and that the target Model is present in its Models[].

1. List groups configured in the current env:

   callCloudApi(service="tcb", action="DescribeAIModels", params={ EnvId })

Returns AIModelGroups: AIModelGroup[] with GroupName, Type (builtin / custom), Models: [{ Model, EnableMCP, Tags }], Status (1 / 2), BaseUrl, Secret, Remark. The main managed GroupName is cloudbase.

2. Never assume a model is already enabled. Inspect AIModelGroups[?].Models[].Model for the target group. If the text model you plan to use (e.g. deepseek-v4-flash, or whatever the user asked for) is missing from the cloudbase group's Models[], jump to step 4 and enable it — do not call createModel("cloudbase") yet. Image generation uses createImageModel("hunyuan-image") + model: "hunyuan-image"; verify it is likewise enabled before the call.

3. User asked for a model from the managed catalog (e.g. deepseek-v3.2, hunyuan-2.0-instruct-20251111): check whether that Model is already in the cloudbase group's Models[]. If not, jump to step 4. Do not guess the exact model id — confirm the canonical spelling in DescribeManagedAIModelList first.

4. Enable / add a managed model (always inspect the authoritative catalog + pricing first):

   callCloudApi(service="tcb", action="DescribeManagedAIModelList", params={ EnvId })

Returns ManagedAIModelGroup[] with GroupName, Remark, and Models: [{ Model, EnableMCP, ModelSpec, ModelChargingInfo }]. This is the single source of truth for supported model names and pricing — do not infer them from memory. Use the exact `Model` string from here when calling `UpdateAIModel`. ModelChargingInfo includes input / output prices and billing unit. Surface the prices to the user before enabling.

Then enable (note: Models is a full replacement — always resend the already-enabled models together with the new one):

   callCloudApi(service="tcb", action="UpdateAIModel", params={
     EnvId,
     GroupName: "cloudbase",
     Models: [
       // resend every model that DescribeAIModels already showed as enabled
       { Model: "<already-enabled model>" },
       // append the newly-requested one, using the exact spelling from DescribeManagedAIModelList
       { Model: "<target model>" }
     ],
     Status: 1
   })

5. The requested model is not in the managed catalog (not found by DescribeManagedAIModelList) → jump to the next section, Custom onboarding (models outside the managed catalog).

All Actions use service=tcb, Version=2018-06-08. Parameters are PascalCase; fall back to camelCase only on InvalidParameter.

---

Available Providers and Models

ai.createModel(<GroupName>) accepts exactly three kinds of legal values; ai.createImageModel("hunyuan-image") is the dedicated image-generation entry point.

1. "cloudbase" — the main managed group (recommended)

• GroupName: "cloudbase", Type: "builtin", Remark: "腾讯云开发" (Tencent CloudBase)
• Backed by Tencent Cloud TokenHub, a unified managed pool covering multiple vendors — Hunyuan (HY 2.0 Instruct, HY 2.0 Think, Hunyuan-role, Hy3 preview, …), DeepSeek (DeepSeek-V4-Pro, DeepSeek-V4-Flash, Deepseek-v3.2, Deepseek-v3.1, Deepseek-r1-0528, Deepseek-v3-0324, …), Zhipu GLM (GLM-5, GLM-5-Turbo, GLM-5.1, GLM-5V-Turbo), Kimi (K2.5, K2.6), MiniMax (M2.5, M2.7), and more. The roster evolves — do not hard-code specific SKUs; discover at runtime
• No model is enabled by default. Always call DescribeAIModels first to see what the env has actually enabled; if your target model is missing, call DescribeManagedAIModelList for the authoritative catalog + pricing and then UpdateAIModel (Status: 1, Models full-replacement) to enable it before making the SDK call.
• Authoritative catalog + pricing: DescribeManagedAIModelList
• Env-enabled set: DescribeAIModels

2. "hunyuan-exp" — legacy builtin group (kept for compatibility)

• Default model: hunyuan-2.0-instruct-20251111; additional hunyuan SKUs must be discovered at runtime via DescribeAIModels({ GroupName: "hunyuan-exp" }).Models[] — do not hard-code other IDs
• Use it directly only if DescribeAIModels actually returns this group with Status=1. New projects should prefer cloudbase

3. User-defined GroupName

• Onboarded via CreateAIModel (see the next section). The custom GroupName MUST start with `custom-` (e.g. custom-kimi, custom-moonshot, custom-openai-compat). This naming convention prevents future collisions with built-in / vendor GroupNames (like cloudbase, hunyuan-exp, deepseek, glm, kimi, minimax) that the platform may introduce over time
• Examples: createModel("custom-kimi"), createModel("custom-openai-compat")

Image generation (independent API)

• ai.createImageModel("hunyuan-image") + model: "hunyuan-image". Only supported in the Node SDK

Never write guesses like createModel("deepseek") or createModel("custom") unless DescribeAIModels explicitly returned that exact GroupName.

---

Custom onboarding (models outside the managed catalog)

When the user wants a non-managed text model (self-hosted, enterprise-internal, third-party OpenAI-compatible endpoint, …), do not block. Guide them through onboarding:

Option 1: console flow (recommended, user handles it)

https://tcb.cloud.tencent.com/dev?envId={envId}#/ai

Option 2: programmatic onboarding (CreateAIModel)

callCloudApi(service="tcb", action="CreateAIModel", params={
  EnvId: "<envId>",
  GroupName: "custom-<your-name>",  // MUST start with "custom-" (e.g. custom-kimi, custom-openai-compat); never start with "cloudbase"
  BaseUrl: "<OpenAI-compatible endpoint, e.g. https://api.moonshot.cn/v1>",
  Models: [
    { Model: "<model name, e.g. kimi-k2.5>", EnableMCP: true }
  ],
  Remark: "<optional remark>",
  Status: 1,
  Secret: { ApiKey: "<vendor api key supplied by the user>" }
})

Once onboarded, confirm with DescribeAIModels that the group is ready, then call ai.createModel("<the GroupName you just registered>") from your code. Use UpdateAIModel to add/remove models, rotate keys, or change BaseUrl (remember Models is a full replacement). Use DeleteAIModel to remove a custom group (builtin groups cannot be deleted).

Custom-model billing is covered by the third-party provider and does not draw from the Token Credits resource pack. Field casing follows the live contract — fall back to camelCase on InvalidParameter.

---

Installation

npm install @cloudbase/node-sdk

⚠️ The AI feature requires version 3.16.0 or above. Check with npm list @cloudbase/node-sdk.

---

Initialization

Inside a CloudBase cloud function

const tcb = require('@cloudbase/node-sdk');
const app = tcb.init({ env: '<YOUR_ENV_ID>' });

exports.main = async (event, context) => {
  const ai = app.ai();
  // Use AI features
};

Cloud function configuration for AI models

⚠️ Important: when creating cloud functions that use AI models (especially generateImage() and large text generation), set a longer timeout — these operations can be slow.

Using the MCP tool `manageFunctions(action="createFunction")`:

Legacy compatibility: if an older prompt still says createFunction, keep the same payload shape but execute it through manageFunctions(action="createFunction").

Set timeout inside the func object:

• Parameter: func.timeout (number)
• Unit: seconds
• Range: 1 – 900
• Default: 20 seconds (usually too short for AI operations)

Recommended timeouts:

• Text generation (`generateText`): 60 – 120 s
• Streaming (`streamText`): 60 – 120 s
• Image generation (`generateImage`): 300 – 900 s (recommended: 900 s)
• Combined operations: 900 s (maximum allowed)

In a regular Node.js server

const tcb = require('@cloudbase/node-sdk');
const app = tcb.init({
  env: '<YOUR_ENV_ID>',
  secretId: '<YOUR_SECRET_ID>',
  secretKey: '<YOUR_SECRET_KEY>'
});

const ai = app.ai();

---

generateText() — non-streaming

Prerequisite: the two-step preflight (eligibility + group readiness) has passed. The example below assumes the user did not specify a model, so it uses the cloudbase managed group + deepseek-v4-flash.

const model = ai.createModel("cloudbase");

const result = await model.generateText({
  model: "deepseek-v4-flash",  // must already be enabled in this env (DescribeAIModels → UpdateAIModel)
  messages: [{ role: "user", content: "Give me a one-paragraph intro to Li Bai." }],
});

console.log(result.text);           // generated text string
console.log(result.usage);          // { prompt_tokens, completion_tokens, total_tokens }
console.log(result.messages);       // full message history
console.log(result.rawResponses);   // raw model responses

---

Error Handling Pattern

const model = ai.createModel("cloudbase");

try {
  const result = await model.generateText({
    model: "deepseek-v4-flash",
    messages: [{ role: "user", content: "Summarize today's deployment logs." }],
  });

  console.log(result.text);
} catch (error) {
  console.error("AI request failed", error);
}

---

streamText() — streaming

Prerequisite: the two-step preflight has passed.

const model = ai.createModel("cloudbase");

const res = await model.streamText({
  model: "deepseek-v4-flash",
  messages: [{ role: "user", content: "Give me a one-paragraph intro to Li Bai." }],
});

// Option 1: iterate the text stream (recommended)
for await (let text of res.textStream) {
  console.log(text);  // incremental text chunks
}

// Option 2: iterate the data stream for full response chunks
for await (let data of res.dataStream) {
  console.log(data);  // full response chunk with metadata
}

// Option 3: access final results
const messages = await res.messages;  // full message history
const usage = await res.usage;        // token usage

---

generateImage() — image generation

⚠️ Image generation is only available in the Node SDK, not in the JS SDK (Web) or WeChat Mini Program.

⚠️ Image generation also consumes the Token Credits resource pack, so the two-step preflight must pass before calling it. Per-call cost is higher than text and calls take longer (set cloud function timeout to 900 s).

const imageModel = ai.createImageModel("hunyuan-image");

const res = await imageModel.generateImage({
  model: "hunyuan-image",
  prompt: "A cute kitten playing on the grass",
  size: "1024x1024",
  version: "v1.9",
});

console.log(res.data[0].url);           // image URL (valid for 24 hours)
console.log(res.data[0].revised_prompt);// revised prompt when revise=true

Image Generation Parameters

interface HunyuanGenerateImageInput {
  model: "hunyuan-image";      // required
  prompt: string;                       // required: image description
  version?: "v1.8.1" | "v1.9";         // default: "v1.8.1"
  size?: string;                        // default: "1024x1024"
  negative_prompt?: string;             // v1.9 only
  style?: string;                       // v1.9 only
  revise?: boolean;                     // default: true
  n?: number;                           // default: 1
  footnote?: string;                    // watermark, max 16 chars
  seed?: number;                        // range: [1, 4294967295]
}

interface HunyuanGenerateImageOutput {
  id: string;
  created: number;
  data: Array<{
    url: string;                        // image URL (24h valid)
    revised_prompt?: string;
  }>;
}

---

Type Definitions

interface BaseChatModelInput {
  model: string;                        // required: model name
  messages: Array<ChatModelMessage>;    // required: message array
  temperature?: number;                 // optional: sampling temperature
  topP?: number;                        // optional: nucleus sampling
}

type ChatModelMessage =
  | { role: "user"; content: string }
  | { role: "system"; content: string }
  | { role: "assistant"; content: string };

interface GenerateTextResult {
  text: string;                         // generated text
  messages: Array<ChatModelMessage>;    // full message history
  usage: Usage;                         // token usage
  rawResponses: Array<unknown>;         // raw model responses
  error?: unknown;                      // error if any
}

interface StreamTextResult {
  textStream: AsyncIterable<string>;    // incremental text stream
  dataStream: AsyncIterable<DataChunk>; // full data stream
  messages: Promise<ChatModelMessage[]>;// final message history
  usage: Promise<Usage>;                // final token usage
  error?: unknown;                      // error if any
}

interface Usage {
  prompt_tokens: number;
  completion_tokens: number;
  total_tokens: number;
}

---

Best Practices

1. Run the two-step preflight before writing business code — ① eligibility: envQuery → callCloudApi(tcb, DescribeEnvPostpayPackage) to confirm the Token Credits resource pack (text + image share the same pack); ② group readiness: DescribeAIModels for the cloudbase group and its Models[], DescribeManagedAIModelList for the authoritative supported-model catalog, UpdateAIModel with a full-replacement Models[] + Status: 1 when the target model is missing. If the pack is missing, return the purchase link https://buy.cloud.tencent.com/lowcode?buyType=resPack&envId={envId}&resourceType=token instead of emitting SDK code and letting the user debug runtime errors. 2. Never assume any model is already enabled — not deepseek-v4-flash, not hunyuan-image, not anything. Always verify with DescribeAIModels first; if the target is missing, look up the exact Model string in DescribeManagedAIModelList (do not guess the spelling) and then UpdateAIModel to enable it. 3. `createModel` accepts exactly three kinds of values — "cloudbase" (the main managed group), "hunyuan-exp" (legacy builtin), or a user-defined GroupName registered via CreateAIModel (MUST start with `custom-`, e.g. custom-kimi, custom-openai-compat). Never guess with createModel("deepseek") / createModel("kimi") / createModel("custom") — the first two are vendor/model names, the last is a placeholder. createImageModel("hunyuan-image") is a separate image API — keep it as-is. 4. Do not invent SDK method names or parameters. This SKILL.md is the authoritative reference for @cloudbase/node-sdk's AI surface — look up the method signature here (or in the Type Definitions section) before writing code. If a method or field is not documented here, stop and ask, or check the live contract via the MCP tools. No guessing. 5. Show pricing before enabling a new managed model — DescribeManagedAIModelList returns ModelSpec (context length, max input/output tokens) + ModelChargingInfo (input / output / cache prices, billing unit). Show the prices to the user before calling UpdateAIModel. 6. Plan timeout and quota separately for image generation — generateImage costs more per call than text and takes longer. For cloud functions, set timeout to 900s. HTTP-function gateways cap at 60s, so use an async-task + polling pattern. Throttle per-user concurrency and frequency to avoid burning an entire Token pack on one failure. 7. Prefer streaming for long-form interactions — in HTTP-function or cloud-function SSE scenarios, use streamText + for await (const chunk of result.textStream) to flush chunks back to the client incrementally. Handle stream interruption in catch and close the underlying response. 8. Pin `@cloudbase/node-sdk` >= 3.16.0 on the server — image generation is only available from this version. Verify with npm ls @cloudbase/node-sdk to confirm the version actually loaded by the cloud function / cloud run runtime — local and production can drift. 9. Centralize model names in config, not scattered literals. Keep the chosen text / image model in a single constant and source from DescribeAIModels / DescribeManagedAIModelList. The managed catalog evolves; a single source of truth makes upgrades cheap. For models outside the managed catalog, follow the Custom Onboarding section — never hard-code third-party API keys in business code (let CreateAIModel.Secret.ApiKey hold them via CloudBase). 10. Distinguish "preflight failure" from "model call failure" — the former means the resource pack is not active or the target model has not been enabled via UpdateAIModel (guide the user to purchase / enable). The latter is a parameter issue or upstream error. Do not wrap both in one generic toast. 11. Do not log full prompts or generated text in production — log only usage.total_tokens and a short prefix. Prompts can leak sensitive content; token counts can leak cost signals. 12. TypeScript: do NOT use `any` to silence SDK type errors. The Node SDK ships its own types; narrow with unknown + a type guard, write a precise interface for the shape you consume, or augment types in a local .d.ts. Never : any, as any, @ts-ignore, @ts-nocheck. See the Engineering constitution in the web-development skill — it applies to backend TS too. 13. Self-verify before claiming done. tsc --noEmit + project build + actually invoke the function (local invoke / manageFunctions(action="invokeFunction") / direct HTTP hit) and confirm usage.total_tokens > 0 and the returned text is not an error envelope. "It should work" without a real round-trip is not acceptable evidence.

Authentication Activation Checklist

Use this checklist before generating any CloudBase authentication flow.

When this checklist applies

• Web login or registration
• SMS, email, anonymous (disabled by default), Google, or WeChat provider setup
• HTTP API auth flows for native apps or backend integrations

Required checks

1. Identify the client platform: Web, mini program, native app, or backend. 2. Confirm whether provider configuration must happen before code generation. 3. Check which login methods are required and enable them first. 4. For Web flows, get or confirm the publishable key before writing frontend auth code. 5. Route to the matching implementation skill after provider setup:

• Web -> auth-web
• Mini program -> auth-wechat
• Native app / raw HTTP -> http-api

6. Keep MCP tool routing explicit:

• management-side login -> auth
• application-side auth config -> queryAppAuth / manageAppAuth

Common failure patterns

• Writing a login page before enabling SMS or email login.
• Implementing Web login in cloud functions instead of CloudBase Auth.
• Using Web SDK patterns in native App code.

Done criteria

• Required providers are enabled.
• Platform-specific auth path is selected.
• The next skill to read is explicit before code generation starts.

Cloud Functions Execution Checklist

Use this checklist before creating or updating a CloudBase function.

Required checks

1. Decide whether this is an Event Function or an HTTP Function.

• Event Function: exports.main(event, context), SDK/timer driven
• HTTP Function: req / res, listens on port 9000

2. Pick the runtime before creation and state it explicitly. 3. For HTTP Functions, confirm scf_bootstrap exists and the Node.js binary path matches the runtime (e.g. Nodejs18.15 → /var/lang/node18/bin/node). 4. Confirm the function root path points to the parent directory, not the function directory itself. 5. For HTTP Functions that need public access, configure the function security rule with managePermissions(action="updateResourcePermission", resourceType="function") after creation. Default rules reject unauthenticated callers with EXCEED_AUTHORITY. Note: anonymous login is disabled by default — use rule: "true" for public endpoints. 6. If the request is really for a long-running container service, reroute to cloudrun-development.

Common failure patterns

• Choosing the wrong function type and compensating later.
• Mixing Event Function and HTTP Function handler shapes in the same implementation.
• Forgetting that runtime cannot be changed after creation.
• Mismatching the scf_bootstrap Node.js binary path with the function runtime.
• Forgetting to configure function security rules for HTTP Functions that need public access.
• Treating Cloud Functions as the default answer for Web authentication.

Done criteria

• Function type and runtime are explicit.
• Packaging constraints are checked.
• The task is confirmed to be a function workflow rather than CloudRun.

Cloud Functions Reference Map

Use this file to decide which detailed reference to read after the main skill.

Read this next when

• You already know the task belongs to Cloud Functions, but the main SKILL.md is intentionally keeping only the routing and guardrails.

Reference routing

./references/event-functions.md

Read this when the task is about:

• exports.main(event, context)
• SDK-invoked serverless functions
• timer-triggered jobs
• Event Function deployment or invocation patterns

./references/http-functions.md

Read this when the task is about:

• HTTP endpoints
• REST APIs
• SSE or WebSocket services
• scf_bootstrap
• browser/public access paths for HTTP Functions

./references/operations-and-config.md

Read this when the task is about:

• function logs
• environment-variable updates
• trigger or VPC configuration
• gateway exposure for Event Functions
• legacy tool-name translation
• callCloudApi fallback for Cloud Functions

Keep these distinctions straight

• Event Function code shape: exports.main(event, context)
• HTTP Function code shape: req / res web server on port 9000
• HTTP Access for Event Functions is a gateway configuration, not the HTTP Function runtime model
• CloudRun is the right route when the task is actually a long-lived service or broader container workload

Event Functions Reference

Use this reference when the task is clearly about an Event Function (exports.main(event, context)) rather than an HTTP Function.

Runtime and packaging facts

• Runtime is fixed at creation time and cannot be changed later.
• For new functions, prefer Nodejs18.15 unless dependency compatibility forces an older runtime.
• Event Functions auto-install dependencies from package.json during deployment, so you normally do not ship node_modules.
• The function root path must point to the parent directory that contains the function folder.

Minimal structure

cloudfunctions/
└── myFunction/
    ├── index.js
    └── package.json

exports.main = async (event, context) => {
  return {
    code: 0,
    message: "ok",
    data: { event }
  };
};

Create or update flow

Create

Use manageFunctions(action="createFunction") and make the function type explicit.

manageFunctions({
  action: "createFunction",
  func: {
    name: "myFunction",
    type: "Event",
    runtime: "Nodejs18.15",
    timeout: 30
  },
  functionRootPath: "/absolute/path/to/cloudfunctions"
});

Update code

Use manageFunctions(action="updateFunctionCode") when only code changes.

manageFunctions({
  action: "updateFunctionCode",
  functionName: "myFunction",
  functionRootPath: "/absolute/path/to/cloudfunctions"
});

Key reminders

• updateFunctionCode does not change runtime.
• If runtime must change, recreate the function.
• Prefer MCP management tools over CLI in agent flows.

Invocation patterns

Web

import cloudbase from "@cloudbase/js-sdk";

const app = cloudbase.init({ env: "your-env-id" });
const result = await app.callFunction({
  name: "myFunction",
  data: { userId: "123" }
});

Mini Program

const result = await wx.cloud.callFunction({
  name: "myFunction",
  data: { userId: "123" }
});

Node.js backend

const tcb = require("@cloudbase/node-sdk");
const app = tcb.init({ env: "your-env-id" });

const result = await app.callFunction({
  name: "myFunction",
  data: { userId: "123" }
});

Raw HTTP API

Use the CloudBase HTTP API only when the task is explicitly about raw API invocation.

https://{envId}.api.tcloudbasegateway.com/v1/functions/{functionName}

This path requires authentication and belongs with the http-api skill, not browser-facing anonymous access.

Common patterns

Error handling

exports.main = async (event, context) => {
  try {
    const result = await doWork(event);
    return {
      code: 0,
      message: "Success",
      data: result
    };
  } catch (error) {
    return {
      code: -1,
      message: error.message,
      data: null
    };
  }
};

Environment variables

exports.main = async () => {
  const apiKey = process.env.API_KEY;
  const envId = process.env.ENV_ID;
  return { apiKeyExists: Boolean(apiKey), envId };
};

When to stop and reroute

• If the user wants a long-lived HTTP service, SSE, or WebSocket server, reroute to HTTP Functions or CloudRun.
• If the user wants browser SDK auth or UI login, reroute to the relevant auth skill.
• If the user wants MySQL or document database schema design, reroute to the data skills instead of forcing it into a function tutorial.

HTTP Functions Reference

Use this reference when the task is clearly about an HTTP Function: REST API, browser-facing endpoint, SSE stream, or WebSocket service.

Core model

HTTP Functions are standard web services, not exports.main(event, context) handlers.

• Handle requests through req and res.
• Listen on port 9000.
• Ship an executable scf_bootstrap file.
• Include runtime dependencies in the package; HTTP Functions do not auto-install node_modules for you.
• For simple HTTP APIs, prefer the Node.js native http module so the function shape stays explicit and dependency-light. Only introduce Express, Koa, NestJS, or similar frameworks when the user explicitly asks for one or the service complexity justifies it.

Minimal structure

my-http-function/
├── scf_bootstrap
├── package.json
├── node_modules/
└── index.js

scf_bootstrap

#!/bin/bash
/var/lang/node18/bin/node index.js

Requirements:

• File name must be exactly scf_bootstrap.
• Use LF line endings.
• Make it executable with chmod +x scf_bootstrap.

The scf_bootstrap Node.js binary path must match the function runtime. Use this mapping:

If the user specifies "Node.js 18", use runtime Nodejs18.15 and the path /var/lang/node18/bin/node.

Minimal Node.js example

const http = require("http");
const { URL } = require("url");

// CORS headers — default to * for simple cross-origin APIs
const CORS_HEADERS = {
  "Access-Control-Allow-Origin": "*",
  "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
  "Access-Control-Allow-Headers": "Content-Type",
};

function sendJson(res, statusCode, data) {
  res.writeHead(statusCode, {
    "Content-Type": "application/json; charset=utf-8",
    ...CORS_HEADERS,
  });
  res.end(JSON.stringify(data));
}

function sendOptions(res) {
  res.writeHead(204, CORS_HEADERS);
  res.end();
}

function readJsonBody(req) {
  return new Promise((resolve, reject) => {
    let raw = "";

    req.on("data", (chunk) => {
      raw += chunk;
    });

    req.on("end", () => {
      if (!raw) {
        resolve({});
        return;
      }

      try {
        resolve(JSON.parse(raw));
      } catch (error) {
        reject(new Error("Invalid JSON body"));
      }
    });

    req.on("error", reject);
  });
}

const server = http.createServer(async (req, res) => {
  // Handle CORS preflight
  if (req.method === "OPTIONS") {
    return sendOptions(res);
  }

  const url = new URL(req.url || "/", "http://127.0.0.1");

  if (req.method === "GET" && url.pathname === "/health") {
    sendJson(res, 200, { ok: true });
    return;
  }

  if (req.method === "POST" && url.pathname === "/echo") {
    try {
      const body = await readJsonBody(req);
      sendJson(res, 200, { received: body });
    } catch (error) {
      sendJson(res, 400, { error: error.message });
    }
    return;
  }

  sendJson(res, 404, { error: "Not Found" });
});

server.listen(9000);

Code-writing rules

• Do not write HTTP Functions as exports.main = async (event, context) => {}. That is the Event Function contract.
• Start an HTTP server explicitly with http.createServer(...) or a framework app, and always bind to port 9000.
• Choose one Node.js module system and keep it consistent. For simple HTTP Functions, CommonJS is the safest default: use require(...) and leave "type": "module" out of package.json.
• If you intentionally use ES Modules, use import ... consistently and do not rely on CommonJS-only globals such as bare __dirname, require(...), or module.exports. When you need the current file path in ESM, derive it from import.meta.url.
• Treat routing, method checks, and body parsing as part of the function code. With the native http module, parse req.url yourself and read the request body from the stream before calling JSON.parse.
• Return JSON responses explicitly and set Content-Type yourself, for example application/json; charset=utf-8.
• Handle CORS headers. Browsers block cross-origin requests without proper CORS headers. Default to Access-Control-Allow-Origin: * for simple APIs, and always respond to OPTIONS preflight requests with 200 and CORS headers.
• Keep unsupported routes and methods explicit. Return 404 for unknown paths, and return 405 when the path exists but the HTTP method is not allowed.
• Keep scf_bootstrap, index.js, package.json, and any bundled dependencies in the function directory that will be uploaded.

Module system note

The minimal examples in this document use CommonJS:

• const http = require("http")
• no "type": "module" in package.json

That combination avoids the common ESM pitfall where __dirname is not defined. If you switch to ES Modules, switch the whole function to import syntax and update any file-path logic accordingly.

Request handling rules

• With Node native http, use new URL(req.url, "http://127.0.0.1") and read url.searchParams for query values.
• With Node native http, req.body does not exist. Read the body stream manually, then parse JSON yourself.
• req.headers -> incoming HTTP headers.
• Path parameters are framework-level conveniences. With the native http module, match url.pathname yourself.
• Always send a response explicitly. With Node native http, use res.writeHead(...) and res.end(...).
• Return meaningful status codes such as 400, 401, 404, 405, 500.

Example with method checks

const http = require("http");
const { URL } = require("url");

// CORS headers — default to * for simple cross-origin APIs
const CORS_HEADERS = {
  "Access-Control-Allow-Origin": "*",
  "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
  "Access-Control-Allow-Headers": "Content-Type",
};

function sendJson(res, statusCode, data) {
  res.writeHead(statusCode, {
    "Content-Type": "application/json; charset=utf-8",
    ...CORS_HEADERS,
  });
  res.end(JSON.stringify(data));
}

function sendOptions(res) {
  res.writeHead(204, CORS_HEADERS);
  res.end();
}

function readJsonBody(req) {
  return new Promise((resolve, reject) => {
    let raw = "";

    req.on("data", (chunk) => {
      raw += chunk;
    });

    req.on("end", () => {
      if (!raw) {
        resolve({});
        return;
      }

      try {
        resolve(JSON.parse(raw));
      } catch (error) {
        reject(new Error("Invalid JSON body"));
      }
    });

    req.on("error", reject);
  });
}

const server = http.createServer(async (req, res) => {
  // Handle CORS preflight
  if (req.method === "OPTIONS") {
    return sendOptions(res);
  }

  const url = new URL(req.url || "/", "http://127.0.0.1");

  if (url.pathname === "/users" && req.method === "POST") {
    try {
      const { name, email } = await readJsonBody(req);

      if (!name || !email) {
        sendJson(res, 400, { error: "name and email are required" });
        return;
      }

      sendJson(res, 201, { name, email });
    } catch (error) {
      sendJson(res, 400, { error: error.message });
    }

    return;
  }

  if (url.pathname === "/users") {
    sendJson(res, 405, { error: "Method Not Allowed" });
    return;
  }

  sendJson(res, 404, { error: "Not Found" });
});

server.listen(9000);

Express 5 catch-all note

If the user explicitly asks for Express, keep in mind that Express 5 uses path-to-regexp semantics for wildcards. Do not use bare * or /* as the catch-all route.

app.all("/{*splat}", (req, res) => {
  res.status(405).json({ error: "Method Not Allowed" });
});

Express 5 note: app.all("/{*splat}", (req, res) => { is the safe catch-all form when you also need to match the root path /, because the router is based on path-to-regexp rather than the older Express 4 wildcard behavior.

End-to-end deployment lifecycle

Follow these steps in order when creating an HTTP Function:

1. Write the function code — create the directory with index.js, scf_bootstrap, and package.json. 2. Deploy with `manageFunctions` — set type: "HTTP", protocolType: "HTTP", and runtime explicitly. 3. Configure security rules — HTTP Functions default to a restrictive security rule. If the function should be publicly accessible, call managePermissions(action="updateResourcePermission") with resourceType="function". Note: anonymous login is disabled by default for new environments; use permission: "CUSTOM" with securityRule: '{"invoke":"true"}' for truly public endpoints rather than relying on anonymous auth. 4. Verify — call the function URL and confirm it returns the expected response. If you get EXCEED_AUTHORITY, the security rule needs to be updated (step 3).

Deployment flow

Prefer manageFunctions over CLI in agent flows.

manageFunctions({
  action: "createFunction",
  func: {
    name: "myHttpFunction",
    type: "HTTP",
    protocolType: "HTTP",
    runtime: "Nodejs18.15",
    timeout: 60
  },
  functionRootPath: "/absolute/path/to/cloudfunctions"
});

Important parameters:

• type: "HTTP" — marks the function as an HTTP Function (not an Event Function).
• protocolType: "HTTP" — the wire protocol. Use "WS" for WebSocket.
• runtime — the execution runtime. Must match the scf_bootstrap binary path. Default is "Nodejs18.15" if omitted, but always set it explicitly to avoid ambiguity.
• functionRootPath — the parent directory of the function folder (e.g. /path/to/cloudfunctions if the code lives in /path/to/cloudfunctions/myHttpFunction/).

Security rule configuration

After creating an HTTP Function, it will reject unauthenticated callers with EXCEED_AUTHORITY by default. If the function should be publicly accessible:

⚠️ Note: Anonymous login is disabled by default for new environments. For public endpoints, use rule: "true" to allow all callers regardless of auth state, rather than relying on anonymous login being enabled.

managePermissions({
  action: "updateResourcePermission",
  resourceType: "function",
  resourceId: "myHttpFunction",
  permission: {
    aclTag: "CUSTOM",
    rule: "true"
  }
});

• aclTag: "CUSTOM" with rule: "true" allows all callers (public access without requiring any login).
• Do NOT use readSecurityRule / writeSecurityRule — those are removed. Use queryPermissions / managePermissions instead.
• Security rule semantics for resourceType="function" differ from NoSQL database rules. Do not reuse doc._openid or auth.openid expressions from NoSQL security rules.
• Official reference: https://docs.cloudbase.net/cloud-function/security-rules

If an external caller reports EXCEED_AUTHORITY, inspect the function permission first with queryPermissions(action="getResourcePermission", resourceType="function", resourceId="myHttpFunction") before widening access.

WebSocket

For WebSocket workloads, keep the function type as HTTP and switch protocolType:

manageFunctions({
  action: "createFunction",
  func: {
    name: "mySocketFunction",
    type: "HTTP",
    protocolType: "WS"
  },
  functionRootPath: "/absolute/path/to/cloudfunctions"
});

Invocation options

HTTP API with token

curl -L "https://{envId}.api.tcloudbasegateway.com/v1/functions/{name}?webfn=true" \
  -H "Authorization: Bearer <TOKEN>"

This is suitable for authenticated server-to-server access.

HTTP access path for browser/public access

Creating the function does not automatically create a browser-facing path. Add gateway access separately when the user actually needs it.

manageGateway({
  action: "createAccess",
  targetType: "function",
  targetName: "myHttpFunction",
  type: "HTTP",
  path: "/api/hello"
});

Before enabling public access, confirm both of these:

1. The access path exists. 2. The function security rule allows the intended caller identity (see Security rule configuration above). Note: anonymous login is disabled by default — for public endpoints, use rule: "true" instead of requiring anonymous auth.

SSE and WebSocket notes

SSE

res.setHeader("Content-Type", "text/event-stream");
res.write(`data: ${JSON.stringify({ content: "Hello" })}\n\n`);

WebSocket example

const WebSocket = require("ws");
const wss = new WebSocket.Server({ port: 9000 });

wss.on("connection", (ws) => {
  ws.on("message", (message) => ws.send(`Echo: ${message}`));
});

When to stop and reroute

• If the task is actually a timer-triggered or SDK-invoked serverless function, reroute to Event Functions.
• If the task needs long-lived containers, custom system packages, or broader service architecture, reroute to cloudrun-development.
• If the task is only about HTTP API calling patterns rather than implementation, reroute to http-api.

Cloud Functions Operations and Config Reference

Use this reference for logs, gateway exposure, environment-variable updates, triggers, and legacy tool-name translation.

Logs

Preferred path

• queryFunctions(action="listFunctionLogs") for the log list.
• queryFunctions(action="getFunctionLogDetail") for a specific request log.

Plan B: callCloudApi

Only use raw cloud API calls after reading the official docs or knowledge-base entry for the action and parameter contract. Do not guess the action name or payload shape from memory.

Log list

callCloudApi({
  service: "tcb",
  action: "GetFunctionLogs",
  params: {
    EnvId: "{envId}",
    FunctionName: "functionName",
    Offset: 0,
    Limit: 10,
    StartTime: "2024-01-01 00:00:00",
    EndTime: "2024-01-01 23:59:59"
  }
});

Log detail

callCloudApi({
  service: "tcb",
  action: "GetFunctionLogDetail",
  params: {
    StartTime: "2024-01-01 00:00:00",
    EndTime: "2024-01-01 23:59:59",
    LogRequestId: "request-id-from-log-list"
  }
});

Log query limits

• Offset + Limit cannot exceed 10000.
• StartTime to EndTime cannot span more than one day.
• For large ranges, page through day-sized windows.

Event Function HTTP access

Preferred path

Use manageGateway(action="createAccess").

Plan B: callCloudApi

Use raw cloud API only after checking the documentation for CreateCloudBaseGWAPI and confirming the gateway parameter contract.

callCloudApi({
  service: "tcb",
  action: "CreateCloudBaseGWAPI",
  params: {
    EnableUnion: true,
    Path: "/api/users",
    ServiceId: "{envId}",
    Type: 6,
    Name: "functionName",
    AuthSwitch: 2,
    PathTransmission: 2,
    EnableRegion: true,
    Domain: "*"
  }
});

Key parameters:

• Type: 6 -> function gateway type.
• AuthSwitch: 2 -> no auth. Use an authenticated mode only when the requirement says so.
• Domain: "*" -> default domain.

Environment variable updates

Do not overwrite function environment variables blindly.

Safe pattern

1. Read current config with queryFunctions(action="getFunctionDetail"). 2. Merge existing variables with the new variables. 3. Update with manageFunctions(action="updateFunctionConfig").

const current = await queryFunctions({
  action: "getFunctionDetail",
  functionName: "functionName"
});

const mergedEnvVariables = {
  ...current.EnvVariables,
  ...newEnvVariables
};

await manageFunctions({
  action: "updateFunctionConfig",
  functionName: "functionName",
  envVariables: mergedEnvVariables
});

Trigger and VPC notes

Timer triggers

Configure timer triggers through func.triggers.

• Type: timer
• Cron format: 7 fields -> second minute hour day month week year

Examples:

• 0 0 2 1 * * * -> 2:00 AM on the first day of every month
• 0 30 9 * * * * -> 9:30 AM every day

VPC access

{
  vpc: {
    vpcId: "vpc-xxxxx",
    subnetId: "subnet-xxxxx"
  }
}

Legacy tool-name translation

Prefer the converged entrances below, but translate historical names when they appear in old prompts or old docs.

CLI fallback

Use CLI only when MCP tools are unavailable AND CLI is explicitly enabled in the runtime environment.

• tcb fn deploy <name> -> Event Function
• tcb fn deploy <name> --httpFn -> HTTP Function
• tcb fn deploy <name> --httpFn --ws -> HTTP Function with WebSocket
• tcb fn deploy --all -> Deploy all functions
• tcb fn config update <name> -> Update function config (timeout, memorySize, envVariables)

Important: When the available capabilities include MCP tools but not CLI access, use MCP tools exclusively. Do not attempt CLI commands in such environments.

Batch updates via MCP: MCP does not have a --all batch parameter. To update multiple functions, call manageFunctions(action="updateFunctionConfig") individually for each function.

In non-interactive agent runs, do not default to CLI login or interactive setup flows.

Coze Adapter

This guide covers using the Coze platform integration with CloudBase Agent Python SDK.

Overview

The Coze adapter allows you to use Coze's hosted AI bots as your backend, while still exposing them through the AG-UI protocol. This is useful when:

• You want to leverage Coze's bot building capabilities
• You need to integrate Coze bots into AG-UI-compatible frontends
• You want unified authentication and middleware with other adapters

Installation

Coze adapter is included in the cloudbase-agent-coze package:

pip install cloudbase-agent-coze

Basic Usage

from cloudbase_agent.coze import CozeAgentAdapter
from cloudbase_agent.server import AgentServiceApp

def create_agent():
    return CozeAgentAdapter(
        bot_id="your-bot-id",
        api_key="your-api-key"
    )

AgentServiceApp().run(create_agent, port=9000)

Configuration

Required Parameters

Optional Parameters

Example with All Options

adapter = CozeAgentAdapter(
    bot_id="bot_1234567890",
    api_key="sk-1234567890",
    base_url="https://api.coze.com",
    debug_mode=True
)

Authentication Integration

The Coze adapter automatically extracts user ID from the request context set by authentication middleware.

Server Setup with Auth

from cloudbase_agent.server import AgentServiceApp
from cloudbase_agent.coze import CozeAgentAdapter
import jwt

def auth_middleware(input_data, request):
    """Extract user from JWT and inject into state."""
    token = request.headers.get("Authorization", "").replace("Bearer ", "")
    
    if token:
        jwt_payload = jwt.decode(token, "your-secret", algorithms=["HS256"])
        
        if input_data.state is None:
            input_data.state = {}
        
        # Inject user ID (Coze adapter reads from here)
        input_data.state["__request_context__"] = {
            "user": {
                "id": jwt_payload["sub"],
                "jwt": jwt_payload
            }
        }
    
    yield

def create_agent():
    return CozeAgentAdapter(
        bot_id="your-bot-id",
        api_key="your-api-key"
    )

app = AgentServiceApp()
app.use(auth_middleware)
app.run(create_agent, port=9000)

User ID Extraction

The Coze adapter reads user ID from:

state["__request_context__"]["user"]["id"]

This is used as the user_id parameter when calling Coze API, enabling:

• User-specific conversation history
• Multi-tenant isolation
• Personalized responses

Environment Variables

For production, use environment variables:

# .env
COZE_BOT_ID=bot_1234567890
COZE_API_KEY=sk-1234567890
COZE_BASE_URL=https://api.coze.com  # optional

import os
from cloudbase_agent.coze import CozeAgentAdapter

def create_agent():
    return CozeAgentAdapter(
        bot_id=os.getenv("COZE_BOT_ID"),
        api_key=os.getenv("COZE_API_KEY"),
        base_url=os.getenv("COZE_BASE_URL", "https://api.coze.com")
    )

Error Handling

The Coze adapter handles common errors and emits AG-UI ERROR events:

Common Errors

Custom Error Handling

from cloudbase_agent.coze import CozeAgentAdapter

def create_agent():
    adapter = CozeAgentAdapter(
        bot_id="your-bot-id",
        api_key="your-api-key",
        debug_mode=True  # Enable debug logging
    )
    return adapter

Features

Streaming Responses

Coze adapter automatically streams responses from the Coze API:

TEXT_MESSAGE_START
TEXT_MESSAGE_CONTENT (chunk 1)
TEXT_MESSAGE_CONTENT (chunk 2)
...
TEXT_MESSAGE_END

Tool Support

If your Coze bot uses tools, tool calls are automatically handled and streamed as AG-UI TOOL_CALL events.

Conversation History

Coze maintains conversation history on their platform. Pass threadId in requests to continue conversations:

{
  "messages": [...],
  "threadId": "conversation-123"
}

Complete Example

# app.py
import os
import jwt
from cloudbase_agent.server import AgentServiceApp
from cloudbase_agent.coze import CozeAgentAdapter
from cloudbase_agent.server.send_message.models import RunAgentInput
from fastapi import Request
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# JWT configuration
JWT_SECRET = os.getenv("JWT_SECRET_KEY", "dev-secret")
JWT_ALGORITHM = "HS256"

def auth_middleware(input_data: RunAgentInput, request: Request):
    """Extract user from JWT and inject into state."""
    auth_header = request.headers.get("Authorization", "")
    
    if not auth_header.startswith("Bearer "):
        logger.warning("Missing or invalid Authorization header")
        # For development, use a default user ID
        if input_data.state is None:
            input_data.state = {}
        input_data.state["__request_context__"] = {
            "user": {"id": "anonymous"}
        }
        yield
        return
    
    token = auth_header[7:]
    
    try:
        jwt_payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
        
        if input_data.state is None:
            input_data.state = {}
        
        input_data.state["__request_context__"] = {
            "user": {
                "id": jwt_payload["sub"],
                "jwt": jwt_payload
            }
        }
        
        logger.info(f"Authenticated user: {jwt_payload['sub']}")
        
    except jwt.InvalidTokenError as e:
        logger.error(f"JWT validation failed: {e}")
        raise
    
    yield

def logging_middleware(input_data, request):
    """Log request details."""
    logger.info(f"Request: {request.url.path}")
    logger.info(f"Run ID: {input_data.runId}")
    logger.info(f"Thread ID: {input_data.threadId}")
    
    yield
    
    logger.info("Request completed")

def create_agent():
    """Create Coze agent adapter."""
    return CozeAgentAdapter(
        bot_id=os.getenv("COZE_BOT_ID"),
        api_key=os.getenv("COZE_API_KEY"),
        debug_mode=os.getenv("DEBUG", "false").lower() == "true"
    )

# Create and configure app
app = AgentServiceApp()
app.set_cors_config(allow_origins=["*"])
app.use(logging_middleware)
app.use(auth_middleware)

if __name__ == "__main__":
    app.run(
        create_agent,
        port=int(os.getenv("PORT", "9000")),
        host="0.0.0.0"
    )

Deployment

Local Development

export COZE_BOT_ID=your-bot-id
export COZE_API_KEY=your-api-key
export JWT_SECRET_KEY=your-dev-secret

python app.py

Docker

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

ENV PORT=9000

CMD ["python", "app.py"]

docker build -t coze-agent .
docker run -p 9000:9000 \
  -e COZE_BOT_ID=your-bot-id \
  -e COZE_API_KEY=your-api-key \
  -e JWT_SECRET_KEY=your-secret \
  coze-agent

CloudRun (Tencent Cloud)

# cloudbaserc.json
{
  "envId": "your-env-id",
  "services": [{
    "name": "coze-agent",
    "path": "./",
    "runtime": "Python3.9",
    "port": 9000,
    "env": {
      "COZE_BOT_ID": "${COZE_BOT_ID}",
      "COZE_API_KEY": "${COZE_API_KEY}",
      "JWT_SECRET_KEY": "${JWT_SECRET_KEY}"
    }
  }]
}

Testing

import pytest
from cloudbase_agent.coze import CozeAgentAdapter
from cloudbase_agent.core import RunAgentInput

@pytest.mark.asyncio
async def test_coze_adapter():
    """Test Coze adapter basic flow."""
    adapter = CozeAgentAdapter(
        bot_id="test-bot",
        api_key="test-key"
    )
    
    run_input = RunAgentInput(
        runId="test-run",
        threadId="test-thread",
        messages=[{"role": "user", "content": "Hello"}],
        state={"__request_context__": {"user": {"id": "test-user"}}}
    )
    
    events = []
    async for event in adapter.run(run_input):
        events.append(event)
    
    # Verify event flow
    assert events[0].type == "RUN_STARTED"
    assert events[-1].type == "RUN_FINISHED"

Troubleshooting

"user_id not found" Error

Problem: Coze adapter can't find user ID in state.

Solution: Ensure auth middleware is registered and sets state.__request_context__.user.id:

app.use(auth_middleware)  # Register before run()

"Invalid API key" Error

Problem: Coze API key is invalid.

Solution: 1. Check your Coze API key 2. Verify it's correctly set in environment variables 3. Test with Coze API directly

Rate Limiting

Problem: Hitting Coze API rate limits.

Solution: Implement rate limiting middleware:

def rate_limit_middleware(input_data, request):
    # Implement rate limiting logic
    yield

Examples

See /python-sdk/examples/coze/ for complete examples.

Next Steps

• Learn about authentication
• Deploy your server: server-quickstart.md
• Build UI: ui-clients.md

Custom Adapter Development

This guide explains how to build custom AG-UI protocol adapters in Python.

Overview

An adapter bridges an Agent framework (LangGraph, LangChain, custom logic) to the AG-UI protocol. It translates framework events into standardized AG-UI events that clients can consume.

AbstractAgent Interface

All adapters must implement the AbstractAgent interface:

from typing import Any, AsyncGenerator
from cloudbase_agent.core import RunAgentInput, Event

class AbstractAgent:
    """Abstract base class for all AG-UI protocol adapters."""
    
    async def run(self, run_input: RunAgentInput) -> AsyncGenerator[Event, None]:
        """
        Execute the agent and yield AG-UI protocol events.
        
        :param run_input: Input data containing messages, state, tools, etc.
        :yields: AG-UI protocol events
        """
        raise NotImplementedError

Event Types

AG-UI protocol defines these event types:

from cloudbase_agent.core import EventType

class EventType:
    RUN_STARTED = "RUN_STARTED"
    RUN_FINISHED = "RUN_FINISHED"
    TEXT_MESSAGE_START = "TEXT_MESSAGE_START"
    TEXT_MESSAGE_CONTENT = "TEXT_MESSAGE_CONTENT"
    TEXT_MESSAGE_END = "TEXT_MESSAGE_END"
    TOOL_CALL_START = "TOOL_CALL_START"
    TOOL_CALL_ARGS_CHUNK = "TOOL_CALL_ARGS_CHUNK"
    TOOL_CALL_END = "TOOL_CALL_END"
    TOOL_RESULT = "TOOL_RESULT"
    STATE_SNAPSHOT = "STATE_SNAPSHOT"
    ERROR = "ERROR"

Minimal Adapter Example

from typing import Any, AsyncGenerator
from cloudbase_agent.core import RunAgentInput, Event, EventType
from uuid import uuid4

class SimpleEchoAgent:
    """Simplest possible adapter - echoes user messages."""
    
    async def run(self, run_input: RunAgentInput) -> AsyncGenerator[Event, None]:
        """Echo back the user's message."""
        # 1. Yield RUN_STARTED
        yield Event(type=EventType.RUN_STARTED, runId=run_input.runId)
        
        # 2. Get last user message
        last_message = run_input.messages[-1] if run_input.messages else None
        user_content = last_message.get("content", "") if last_message else ""
        
        # 3. Generate response
        message_id = str(uuid4())
        response_text = f"Echo: {user_content}"
        
        # 4. Yield TEXT_MESSAGE events
        yield Event(
            type=EventType.TEXT_MESSAGE_START,
            runId=run_input.runId,
            messageId=message_id,
            role="assistant"
        )
        
        yield Event(
            type=EventType.TEXT_MESSAGE_CONTENT,
            runId=run_input.runId,
            messageId=message_id,
            content=response_text
        )
        
        yield Event(
            type=EventType.TEXT_MESSAGE_END,
            runId=run_input.runId,
            messageId=message_id
        )
        
        # 5. Yield RUN_FINISHED
        yield Event(type=EventType.RUN_FINISHED, runId=run_input.runId)

Deploy it:

from cloudbase_agent.server import AgentServiceApp

AgentServiceApp().run(lambda: SimpleEchoAgent(), port=9000)

Streaming Response Pattern

For LLM streaming responses:

from openai import AsyncOpenAI

class StreamingLLMAgent:
    """Agent with streaming LLM responses."""
    
    def __init__(self, api_key: str):
        self.client = AsyncOpenAI(api_key=api_key)
    
    async def run(self, run_input: RunAgentInput) -> AsyncGenerator[Event, None]:
        yield Event(type=EventType.RUN_STARTED, runId=run_input.runId)
        
        # Convert messages to OpenAI format
        messages = [
            {"role": msg["role"], "content": msg["content"]}
            for msg in run_input.messages
        ]
        
        # Stream response
        message_id = str(uuid4())
        yield Event(
            type=EventType.TEXT_MESSAGE_START,
            runId=run_input.runId,
            messageId=message_id,
            role="assistant"
        )
        
        stream = await self.client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            stream=True
        )
        
        async for chunk in stream:
            content = chunk.choices[0].delta.content
            if content:
                yield Event(
                    type=EventType.TEXT_MESSAGE_CONTENT,
                    runId=run_input.runId,
                    messageId=message_id,
                    content=content
                )
        
        yield Event(
            type=EventType.TEXT_MESSAGE_END,
            runId=run_input.runId,
            messageId=message_id
        )
        
        yield Event(type=EventType.RUN_FINISHED, runId=run_input.runId)

Tool Calling Pattern

For agents that call tools:

class ToolCallingAgent:
    """Agent with tool calling support."""
    
    async def run(self, run_input: RunAgentInput) -> AsyncGenerator[Event, None]:
        yield Event(type=EventType.RUN_STARTED, runId=run_input.runId)
        
        # Decide to call a tool
        tool_call_id = str(uuid4())
        tool_name = "get_weather"
        tool_args = {"location": "San Francisco"}
        
        # 1. Yield TOOL_CALL_START
        yield Event(
            type=EventType.TOOL_CALL_START,
            runId=run_input.runId,
            toolCallId=tool_call_id,
            toolName=tool_name
        )
        
        # 2. Yield TOOL_CALL_ARGS_CHUNK (can stream args)
        import json
        args_json = json.dumps(tool_args)
        yield Event(
            type=EventType.TOOL_CALL_ARGS_CHUNK,
            runId=run_input.runId,
            toolCallId=tool_call_id,
            argsChunk=args_json
        )
        
        # 3. Yield TOOL_CALL_END
        yield Event(
            type=EventType.TOOL_CALL_END,
            runId=run_input.runId,
            toolCallId=tool_call_id
        )
        
        # 4. Execute tool (if server-side tool)
        result = await self.execute_tool(tool_name, tool_args)
        
        # 5. Yield TOOL_RESULT
        yield Event(
            type=EventType.TOOL_RESULT,
            runId=run_input.runId,
            toolCallId=tool_call_id,
            result=result
        )
        
        # 6. Continue with response using tool result
        # ... (yield TEXT_MESSAGE events)
        
        yield Event(type=EventType.RUN_FINISHED, runId=run_input.runId)

State Snapshot Pattern

For stateful agents:

class StatefulAgent:
    """Agent that maintains and shares state."""
    
    async def run(self, run_input: RunAgentInput) -> AsyncGenerator[Event, None]:
        yield Event(type=EventType.RUN_STARTED, runId=run_input.runId)
        
        # Process and update state
        current_state = run_input.state or {}
        current_state["message_count"] = current_state.get("message_count", 0) + 1
        current_state["last_message_time"] = time.time()
        
        # ... (process messages)
        
        # Yield STATE_SNAPSHOT
        yield Event(
            type=EventType.STATE_SNAPSHOT,
            runId=run_input.runId,
            snapshot=current_state
        )
        
        yield Event(type=EventType.RUN_FINISHED, runId=run_input.runId)

Error Handling Pattern

class RobustAgent:
    """Agent with proper error handling."""
    
    async def run(self, run_input: RunAgentInput) -> AsyncGenerator[Event, None]:
        try:
            yield Event(type=EventType.RUN_STARTED, runId=run_input.runId)
            
            # Your logic here
            # ...
            
            yield Event(type=EventType.RUN_FINISHED, runId=run_input.runId)
            
        except Exception as e:
            # Yield ERROR event
            yield Event(
                type=EventType.ERROR,
                runId=run_input.runId,
                error={
                    "code": "AGENT_ERROR",
                    "message": str(e),
                    "details": {"traceback": traceback.format_exc()}
                }
            )
            
            # Still yield RUN_FINISHED
            yield Event(type=EventType.RUN_FINISHED, runId=run_input.runId)

Complete Example: Custom Framework Adapter

from typing import Any, AsyncGenerator
from cloudbase_agent.core import RunAgentInput, Event, EventType
from uuid import uuid4
import logging

logger = logging.getLogger(__name__)

class MyCustomFrameworkAgent:
    """
    Adapter for a custom agent framework.
    
    This example shows how to integrate any custom agent logic
    with the AG-UI protocol.
    """
    
    def __init__(self, config: dict):
        """
        Initialize the adapter.
        
        :param config: Configuration for your custom framework
        """
        self.config = config
        # Initialize your framework here
        self.agent = self._initialize_agent()
    
    def _initialize_agent(self):
        """Initialize your custom agent framework."""
        # Your framework initialization logic
        return CustomFrameworkAgent(self.config)
    
    async def run(self, run_input: RunAgentInput) -> AsyncGenerator[Event, None]:
        """
        Execute agent and yield AG-UI protocol events.
        
        :param run_input: Input from AG-UI client
        :yields: AG-UI protocol events
        """
        try:
            # 1. Start
            yield Event(type=EventType.RUN_STARTED, runId=run_input.runId)
            logger.info(f"Run started: {run_input.runId}")
            
            # 2. Extract input data
            messages = run_input.messages
            state = run_input.state or {}
            tools = run_input.tools or []
            
            # 3. Get user context (if auth middleware is used)
            user_id = self._get_user_id(state)
            logger.info(f"User: {user_id}")
            
            # 4. Execute your custom framework
            message_id = str(uuid4())
            
            # Start message
            yield Event(
                type=EventType.TEXT_MESSAGE_START,
                runId=run_input.runId,
                messageId=message_id,
                role="assistant"
            )
            
            # Your framework's execution (can be streaming)
            async for chunk in self.agent.process(messages, state):
                # Handle different chunk types
                if chunk["type"] == "text":
                    yield Event(
                        type=EventType.TEXT_MESSAGE_CONTENT,
                        runId=run_input.runId,
                        messageId=message_id,
                        content=chunk["content"]
                    )
                
                elif chunk["type"] == "tool_call":
                    yield Event(
                        type=EventType.TOOL_CALL_START,
                        runId=run_input.runId,
                        toolCallId=chunk["id"],
                        toolName=chunk["name"]
                    )
                    
                    yield Event(
                        type=EventType.TOOL_CALL_ARGS_CHUNK,
                        runId=run_input.runId,
                        toolCallId=chunk["id"],
                        argsChunk=chunk["args"]
                    )
                    
                    yield Event(
                        type=EventType.TOOL_CALL_END,
                        runId=run_input.runId,
                        toolCallId=chunk["id"]
                    )
                
                elif chunk["type"] == "state_update":
                    yield Event(
                        type=EventType.STATE_SNAPSHOT,
                        runId=run_input.runId,
                        snapshot=chunk["state"]
                    )
            
            # End message
            yield Event(
                type=EventType.TEXT_MESSAGE_END,
                runId=run_input.runId,
                messageId=message_id
            )
            
            # 5. Finish
            yield Event(type=EventType.RUN_FINISHED, runId=run_input.runId)
            logger.info(f"Run finished: {run_input.runId}")
            
        except Exception as e:
            logger.error(f"Error in run: {e}", exc_info=True)
            
            yield Event(
                type=EventType.ERROR,
                runId=run_input.runId,
                error={
                    "code": "AGENT_ERROR",
                    "message": str(e)
                }
            )
            
            yield Event(type=EventType.RUN_FINISHED, runId=run_input.runId)
    
    def _get_user_id(self, state: dict) -> str:
        """Extract user ID from state (set by auth middleware)."""
        return state.get("__request_context__", {}).get("user", {}).get("id", "anonymous")

Testing Your Adapter

Unit Test

import pytest
from cloudbase_agent.core import RunAgentInput

@pytest.mark.asyncio
async def test_adapter_basic_flow():
    """Test basic event flow."""
    adapter = MyCustomFrameworkAgent(config={})
    
    run_input = RunAgentInput(
        runId="test-run",
        threadId="test-thread",
        messages=[{"role": "user", "content": "Hello"}]
    )
    
    events = []
    async for event in adapter.run(run_input):
        events.append(event)
    
    # Verify event sequence
    assert events[0].type == EventType.RUN_STARTED
    assert events[-1].type == EventType.RUN_FINISHED
    
    # Verify message events
    message_events = [e for e in events if "MESSAGE" in e.type]
    assert len(message_events) >= 3  # START, CONTENT, END

@pytest.mark.asyncio
async def test_adapter_error_handling():
    """Test error handling."""
    adapter = MyCustomFrameworkAgent(config={"force_error": True})
    
    run_input = RunAgentInput(
        runId="test-error",
        threadId="test-thread",
        messages=[]
    )
    
    events = []
    async for event in adapter.run(run_input):
        events.append(event)
    
    # Verify ERROR event is emitted
    error_events = [e for e in events if e.type == EventType.ERROR]
    assert len(error_events) == 1

Integration Test

from fastapi.testclient import TestClient
from cloudbase_agent.server import AgentServiceApp

def test_adapter_via_http():
    """Test adapter through HTTP server."""
    app_instance = AgentServiceApp()
    fastapi_app = app_instance.build(
        create_agent=lambda: MyCustomFrameworkAgent(config={})
    )
    
    client = TestClient(fastapi_app)
    
    response = client.post(
        "/send-message",
        json={
            "messages": [{"role": "user", "content": "Hello"}],
            "runId": "test-run",
            "threadId": "test-thread"
        },
        headers={"Accept": "text/event-stream"}
    )
    
    assert response.status_code == 200
    
    # Parse SSE events
    lines = response.text.split("\n")
    events = []
    for line in lines:
        if line.startswith("data: "):
            import json
            event_data = json.loads(line[6:])
            events.append(event_data)
    
    # Verify event flow
    assert events[0]["type"] == "RUN_STARTED"
    assert events[-1]["type"] == "RUN_FINISHED"

Best Practices

1. Always yield RUN_STARTED first - Clients expect this 2. Always yield RUN_FINISHED last - Even after errors 3. Use proper event sequence - START → CONTENT → END for messages 4. Handle errors gracefully - Yield ERROR event, don't raise exceptions 5. Stream when possible - Better UX with incremental updates 6. Log important events - Helps with debugging 7. Extract user context - Use state.__request_context__.user if available 8. Validate input - Check required fields before processing 9. Use type hints - Better IDE support and catch errors early 10. Write tests - Both unit and integration tests

Common Pitfalls

❌ Not yielding RUN_STARTED/FINISHED

async def run(self, run_input):
    # Missing RUN_STARTED
    yield Event(type=EventType.TEXT_MESSAGE_CONTENT, content="Hello")
    # Missing RUN_FINISHED

❌ Raising exceptions instead of ERROR events

async def run(self, run_input):
    if error:
        raise Exception("Error")  # ❌ Breaks SSE stream

Should be:

async def run(self, run_input):
    if error:
        yield Event(type=EventType.ERROR, error={"message": "Error"})
        yield Event(type=EventType.RUN_FINISHED)

❌ Not handling missing state

user_id = run_input.state["__request_context__"]["user"]["id"]  # ❌ May crash

Should be:

user_id = run_input.state.get("__request_context__", {}).get("user", {}).get("id")

Examples

See /python-sdk/examples/ for complete examples:

• langgraph/ - LangGraph adapter patterns
• langchain/ - LangChain adapter patterns
• coze/ - Third-party API integration

Next Steps

• Deploy your adapter: server-quickstart.md
• Understand protocol details: agui-protocol.md
• Add authentication: authentication.md
• Build UI: ui-clients.md

LangGraph Adapter Guide

Complete guide for integrating LangGraph agents with CloudBase Agent Python SDK.

---

Overview

The CloudBase Agent LangGraph adapter (cloudbase_agent.langgraph) provides seamless integration with LangGraph workflows, offering:

• Native LangGraph Support: Wrap any CompiledStateGraph as an CloudBase Agent agent
• AG-UI Compatibility: Automatic stability patches for frontend integration
• Streaming Support: Real-time message streaming to clients
• Memory Persistence: LangGraph checkpoint support for conversation history
• Callback System: Monitor agent events in real-time
• Resource Cleanup: Automatic cleanup after request completion

---

Quick Start

1. Install Dependencies

pip install cloudbase-agent-langgraph cloudbase-agent-server langgraph langchain-openai

This installs:

• cloudbase-agent-langgraph - LangGraph adapter
• langgraph - LangGraph framework
• langchain - LangChain core
• langchain-openai - OpenAI integration

2. Create Your First Agent

# agent.py
from langgraph.graph import StateGraph, MessagesState, END, START
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
from cloudbase_agent.langgraph import LangGraphAgent

# Define state
class State(MessagesState):
    pass

# Define chat node
def chat_node(state: State, config, writer):
    """Generate AI response."""
    chat_model = ChatOpenAI(model="gpt-4o-mini")
    system = SystemMessage(content="You are a helpful assistant.")
    messages = [system, *state["messages"]]
    
    chunks = []
    for chunk in chat_model.stream(messages, config):
        writer({"messages": [chunk]})  # Stream to client
        chunks.append(chunk)
    
    return {"messages": chunks}

# Build workflow
def build_workflow():
    graph = StateGraph(State)
    graph.add_node("chat", chat_node)
    graph.add_edge(START, "chat")
    graph.add_edge("chat", END)
    
    memory = MemorySaver()
    return graph.compile(checkpointer=memory)

# Wrap with CloudBase Agent
agent = LangGraphAgent(
    name="chatbot",
    description="A helpful conversational assistant",
    graph=build_workflow()
)

3. Deploy as HTTP Service

# server.py
from cloudbase_agent.server import AgentServiceApp

AgentServiceApp().run(
    lambda: {"agent": agent},
    port=9000,
    enable_openai_endpoint=True
)

---

LangGraphAgent Configuration

Basic Configuration

from cloudbase_agent.langgraph import LangGraphAgent

agent = LangGraphAgent(
    name="my-agent",  # Required: Agent identifier
    description="Agent description",  # Optional: For documentation
    graph=build_workflow(),  # Required: CompiledStateGraph
    use_callbacks=True,  # Optional: Enable callback system (default: False)
)

Advanced Configuration

agent = LangGraphAgent(
    name="advanced-agent",
    description="Advanced agent with full configuration",
    graph=compiled_graph,
    use_callbacks=True,
    
    # Add callbacks
    callbacks=[ConsoleLogger(), MetricsCollector()],
    
    # Add tool proxy for permission control
    tool_proxy=permission_checker,
)

# Add callbacks dynamically
agent.add_callback(DatabaseLogger())

---

State Management

Basic MessagesState

from langgraph.graph import MessagesState

class State(MessagesState):
    """Simplest state - just conversation history."""
    pass

Extended State with Tools

from langgraph.graph import MessagesState
from typing import List, Any

class State(MessagesState):
    """State with tool support."""
    tools: List[Any]  # Available tools

Custom State Fields

from langgraph.graph import MessagesState
from typing import Optional

class State(MessagesState):
    """State with custom fields."""
    user_id: str  # User identifier
    context: Optional[dict]  # Additional context
    preference: str  # User preferences

---

Streaming Response

StreamWriter Pattern

LangGraph nodes receive a writer parameter for streaming:

from langgraph.types import StreamWriter

def chat_node(state: State, config, writer: StreamWriter):
    """Node with streaming support."""
    
    chat_model = ChatOpenAI(model="gpt-4o-mini")
    
    chunks = []
    for chunk in chat_model.stream(messages, config):
        # Stream chunk to client immediately
        writer({"messages": [chunk]})
        
        # Collect for final state
        chunks.append(chunk)
    
    # Return collected chunks for state
    return {"messages": chunks}

Handling Missing Writer

def chat_node(state: State, config, writer: StreamWriter = None):
    """Node with fallback for missing writer."""
    
    # Provide no-op fallback
    if writer is None:
        def writer(x):
            pass
    
    # Use writer safely
    for chunk in chat_model.stream(messages):
        writer({"messages": [chunk]})

---

Memory & Checkpointing

In-Memory Checkpointer

For development and testing:

from langgraph.checkpoint.memory import MemorySaver

def build_workflow():
    graph = StateGraph(State)
    # ... add nodes and edges ...
    
    memory = MemorySaver()  # In-memory storage
    return graph.compile(checkpointer=memory)

Using Conversation ID

# Each conversation gets unique thread_id
curl -X POST http://localhost:9000/send-message \
  -H "Content-Type: application/json" \
  -d '{
    "conversationId": "user_123_conv_456",
    "messages": [{"role": "user", "content": "Hello!"}]
  }'

The conversationId is automatically mapped to LangGraph's thread_id for checkpoint retrieval.

Persistent Checkpointer

For production with PostgreSQL:

from langgraph.checkpoint.postgres import PostgresSaver

# Create PostgreSQL checkpointer
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/dbname"
)

def build_workflow():
    graph = StateGraph(State)
    # ... add nodes and edges ...
    
    return graph.compile(checkpointer=checkpointer)

---

Tool Integration

Defining Tools

from typing import List, Any
from langchain_core.utils.function_calling import convert_to_openai_function

class State(MessagesState):
    tools: List[Any]

def chat_node(state: State, config, writer):
    chat_model = ChatOpenAI(model="gpt-4o-mini")
    
    # Get and bind tools
    tools = state.get("tools", [])
    if tools:
        # Convert tool definitions to OpenAI format
        tools_list = [convert_to_openai_function(tool) for tool in tools]
        chat_model = chat_model.bind_tools(tools_list)
    
    # Use model with tools
    for chunk in chat_model.stream(messages, config):
        writer({"messages": [chunk]})

Providing Tools via API

curl -X POST http://localhost:9000/send-message \
  -H "Content-Type: application/json" \
  -d '{
    "conversationId": "conv_123",
    "messages": [{"role": "user", "content": "Search the web"}],
    "tools": [
      {
        "name": "search_web",
        "description": "Search the internet",
        "parameters": {
          "type": "object",
          "properties": {
            "query": {"type": "string"}
          },
          "required": ["query"]
        }
      }
    ]
  }'

---

Callbacks

Built-in Callback Interface

class MyCallback:
    """Custom callback for monitoring."""
    
    async def on_text_message_content(self, event, buffer):
        """Called when text message content is streaming."""
        print(f"AI: {buffer}")
    
    async def on_tool_call_args(self, event, buffer, partial_args):
        """Called when tool call arguments are parsed."""
        tool_name = getattr(event, "tool_name", "unknown")
        print(f"Tool: {tool_name}, Args: {partial_args}")
    
    async def on_run_started(self, event):
        """Called when agent run starts."""
        print(f"Started: {event.run_id}")
    
    async def on_run_finished(self, event):
        """Called when agent run finishes."""
        print(f"Finished: {event.run_id}")
    
    async def on_run_error(self, event):
        """Called when an error occurs."""
        print(f"Error: {getattr(event, 'message', 'Unknown')}")

Adding Callbacks

# Method 1: During agent creation
agent = LangGraphAgent(
    name="my-agent",
    graph=workflow,
    use_callbacks=True,
    callbacks=[MyCallback()]
)

# Method 2: After creation
agent.add_callback(MyCallback())

---

Error Handling

AG-UI Protocol Errors

CloudBase Agent automatically converts exceptions to AG-UI error events:

def chat_node(state: State, config, writer):
    try:
        # Your logic here
        result = dangerous_operation()
        return {"messages": [result]}
    except Exception as e:
        # Error is automatically formatted as AG-UI error event
        from langchain_core.messages import AIMessage
        return {"messages": [AIMessage(content=f"Error: {str(e)}")]}

Custom Error Handling

from cloudbase_agent.server.errors import install_exception_handlers
from fastapi import FastAPI

app = FastAPI()

# Install AG-UI error handlers
install_exception_handlers(app)

# Now all exceptions are converted to AG-UI error events

---

Complete Example: Human-in-the-Loop

#!/usr/bin/env python3
from typing import Optional
from langgraph.graph import StateGraph, MessagesState, END, START
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, AIMessage
from cloudbase_agent.langgraph import LangGraphAgent

class State(MessagesState):
    """State for human-in-the-loop workflow."""
    pending_approval: Optional[dict] = None

def chat_node(state: State, config, writer):
    """Generate AI response."""
    chat_model = ChatOpenAI(model="gpt-4o-mini")
    system = SystemMessage(content="You are a helpful assistant.")
    messages = [system, *state["messages"]]
    
    chunks = []
    for chunk in chat_model.stream(messages, config):
        writer({"messages": [chunk]})
        chunks.append(chunk)
    
    # Check if approval is needed
    final_message = chunks[-1] if chunks else AIMessage(content="")
    if "sensitive" in final_message.content.lower():
        return {
            "messages": chunks,
            "pending_approval": {
                "action": "send_message",
                "content": final_message.content
            }
        }
    
    return {"messages": chunks}

def approval_node(state: State, config, writer):
    """Wait for human approval."""
    if state.get("pending_approval"):
        writer({
            "messages": [AIMessage(
                content="This action requires approval. Please approve or reject."
            )]
        })
        # Workflow will interrupt here for human input
        return state
    return state

def should_wait_approval(state: State) -> str:
    """Decide if approval is needed."""
    if state.get("pending_approval"):
        return "approval"
    return END

def build_workflow():
    """Build human-in-the-loop workflow."""
    graph = StateGraph(State)
    
    graph.add_node("chat", chat_node)
    graph.add_node("approval", approval_node)
    
    graph.add_edge(START, "chat")
    graph.add_conditional_edges(
        "chat",
        should_wait_approval,
        {
            "approval": "approval",
            END: END
        }
    )
    
    memory = MemorySaver()
    return graph.compile(
        checkpointer=memory,
        interrupt_before=["approval"]  # Pause before approval
    )

# Create agent
agent = LangGraphAgent(
    name="human-in-the-loop",
    description="Agent with human approval workflow",
    graph=build_workflow(),
    use_callbacks=True
)

# Deploy
if __name__ == "__main__":
    from cloudbase_agent.server import AgentServiceApp
    AgentServiceApp().run(
        lambda: {"agent": agent},
        port=9000
    )

---

Best Practices

1. Always Use MemorySaver

# ✅ Correct: With memory
memory = MemorySaver()
workflow = graph.compile(checkpointer=memory)

# ❌ Wrong: No memory - conversations won't persist
workflow = graph.compile()

2. Stream Immediately

# ✅ Correct: Stream as you generate
for chunk in model.stream(messages):
    writer({"messages": [chunk]})  # Immediate streaming
    chunks.append(chunk)

# ❌ Wrong: Collect first, then stream - defeats streaming purpose
chunks = list(model.stream(messages))
for chunk in chunks:
    writer({"messages": [chunk]})

3. Handle Missing Writer

# ✅ Correct: Fallback for testing
def chat_node(state, config, writer=None):
    if writer is None:
        writer = lambda x: None
    
    # Use writer safely
    writer({"messages": [chunk]})

# ❌ Wrong: Assume writer always exists
def chat_node(state, config, writer):
    writer({"messages": [chunk]})  # Fails in tests

4. Use Environment Variables

# .env
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o-mini
OPENAI_TEMPERATURE=0.7

# Load in code
from dotenv import load_dotenv
load_dotenv()

# Use in node
import os
chat_model = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=float(os.getenv("OPENAI_TEMPERATURE", "0.7"))
)

---

Troubleshooting

Issue: Conversation history not persisting

Solution: Ensure you're using a checkpointer:

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
workflow = graph.compile(checkpointer=memory)

Issue: Streaming not working

Solution: Make sure you're calling writer() with proper format:

# ✅ Correct format
writer({"messages": [chunk]})

# ❌ Wrong format
writer(chunk)  # Missing dict wrapper

Issue: Tool calls not working

Solution: Ensure tools are in state and properly bound:

class State(MessagesState):
    tools: List[Any]  # Add tools field

def chat_node(state, config, writer):
    tools = state.get("tools", [])
    if tools:
        tools_list = [convert_to_openai_function(t) for t in tools]
        model = model.bind_tools(tools_list)

---

Next Steps

• Server Deployment: See server-quickstart.md for server configuration
• Authentication: See authentication.md for auth patterns
• Observability: Enable with export AUTO_TRACES_STDOUT=true
• Examples: Check python-sdk/examples/langgraph/ for complete working examples

Agent Deployment Guide

Core Principle

Always use the `manageAgent` MCP tool to deploy Agent services.

It natively supports SSE streaming, session persistence, and Python 3.10 runtime — purpose-built for Agent scenarios.

Do NOT use createFunction or manageCloudRun for Agent deployment.

Why HTTP Cloud Functions First

---

⚠️ DEPLOYMENT IS A 4-STEP BLOCKING PIPELINE

<!-- INSTRUCTION (AI MUST EXECUTE EVERY STEP IN ORDER) -->

AI: You MUST execute Steps 1→2→3→4 in order. Do NOT skip any step.

Do NOT call `manageAgent` until Step 3 (Verification) passes ALL checks.

Step 1: Ensure Python 3.10  ──▶  Step 2: Build env/  ──▶  Step 3: Verify  ──▶  Step 4: Deploy
       (BLOCKING)                   (BLOCKING)              (BLOCKING)           (final)

---

Step 1: Ensure Python 3.10

<!-- INSTRUCTION (AI MUST EXECUTE) -->

Execute one of the following to ensure Python 3.10 is available:

# Check if python3.10 already exists
python3.10 --version 2>/dev/null || python3 --version

# If NOT 3.10.x, install it:
# Option A: brew (macOS)
brew install python@3.10

# Option B: pyenv
pyenv install 3.10
pyenv local 3.10

Verification (REQUIRED before proceeding):

python3.10 --version
# MUST output: Python 3.10.x
# If it shows 3.11+ or 3.9-, STOP and fix before continuing.

<details> <summary>Why must it be Python 3.10? (background info, AI may skip)</summary>

pip's --python-version flag correctly selects wheel files for the target version, but it does NOT reliably evaluate environment markers (e.g., exceptiongroup; python_version < "3.11") — it may use the host interpreter's version instead of the target version. This causes conditional dependencies like exceptiongroup (required by anyio on Python < 3.11) to be silently skipped, leading to ModuleNotFoundError at runtime on the cloud (which runs Python 3.10). </details>

---

Step 2: Build env/ (One-Shot Install)

<!-- INSTRUCTION (AI MUST EXECUTE) -->

### ⚠️ CRITICAL: env/ is an immutable build artifact

>

The ONLY correct workflow is:

1. Delete → rm -rf ./env

2. Install → single pip install command (below)

3. Never touch again

>

NEVER run a second pip install into env/. NEVER manually copy/move/delete files inside env/.

If you need to change anything, edit requirements.txt and redo steps 1–2 from scratch.

Execute this script as-is. Do NOT break it into separate steps. Do NOT modify any line.

#!/bin/bash
set -euo pipefail

PROJECT_DIR="$(pwd)"

# ── Detect Python 3.10 ──
PYTHON_BIN=""
if command -v python3.10 &>/dev/null; then
  PYTHON_BIN="python3.10"
elif python3 --version 2>&1 | grep -q "3\.10\."; then
  PYTHON_BIN="python3"
else
  echo "❌ ERROR: Python 3.10 not found. Run Step 1 first."
  exit 1
fi
echo "✅ Using: $PYTHON_BIN ($($PYTHON_BIN --version 2>&1))"

# ── Atomic env/ rebuild ──
rm -rf ./env && mkdir ./env

# ── One-shot install ALL deps ──
$PYTHON_BIN -m pip install -r ./requirements.txt \
  --platform manylinux2014_x86_64 \
  --target ./env \
  --python-version 3.10 \
  --only-binary=:all: \
  --upgrade

echo "✅ env/ built successfully"

If pip install reports any errors, STOP and resolve the error first. Do NOT ignore errors and proceed to deploy — the resulting env/ will be incomplete.

---

Step 3: Verify env/ Integrity (MANDATORY)

<!-- INSTRUCTION (AI MUST EXECUTE) -->

Do NOT call `manageAgent` until ALL checks below pass.

If ANY check fails, the ONLY fix is: edit requirements.txt → rm -rf env/ → re-run Step 2.

3a. Verify all top-level packages are present

# List all packages from requirements.txt and verify they exist in env/
# This works for ANY framework — no hardcoded package names
python3.10 -c "
import subprocess, sys, os

# Read requirements.txt
with open('requirements.txt') as f:
    reqs = [line.strip().split('==')[0].split('>=')[0].split('<=')[0].split('~=')[0].split('[')[0].strip()
            for line in f if line.strip() and not line.startswith('#') and not line.startswith('-')]

# For each requirement, check if it's importable from env/
env_path = os.path.abspath('./env')
failed = []
for req in reqs:
    # Convert package name to import name (hyphens → underscores)
    import_name = req.replace('-', '_').lower()
    # Check if directory or .py file exists
    found = (os.path.isdir(os.path.join(env_path, import_name)) or
             os.path.isfile(os.path.join(env_path, import_name + '.py')) or
             os.path.isfile(os.path.join(env_path, import_name + '.so')))
    if not found:
        # Some packages have different import names, try dist-info
        dist_matches = [d for d in os.listdir(env_path) 
                       if d.endswith('.dist-info') and req.replace('-','_').lower() in d.lower()]
        if dist_matches:
            found = True
    if not found:
        failed.append(f'{req} (expected: {import_name})')
    else:
        print(f'  ✅ {req}')

if failed:
    print()
    for f in failed:
        print(f'  ❌ MISSING: {f}')
    print()
    print('Fix: Check requirements.txt spelling, then rm -rf env/ and re-run Step 2')
    sys.exit(1)
else:
    print()
    print('✅ All packages verified in env/')
"

3b. Verify entry point imports work

# Dynamically test that the project's main entry file can resolve imports
# Replace 'server.py' with whatever file the project uses as entry point
PYTHONPATH=./env python3.10 -c "
import sys, ast, os

# Find entry point (server.py or main.py)
entry = None
for candidate in ['server.py', 'main.py', 'app.py']:
    if os.path.isfile(candidate):
        entry = candidate
        break

if not entry:
    print('⚠️  No standard entry file found (server.py/main.py/app.py). Skipping import check.')
    sys.exit(0)

print(f'Checking imports from {entry}...')

# Parse and extract top-level imports
with open(entry) as f:
    tree = ast.parse(f.read())

modules = set()
for node in ast.walk(tree):
    if isinstance(node, ast.Import):
        for alias in node.names:
            modules.add(alias.name.split('.')[0])
    elif isinstance(node, ast.ImportFrom) and node.module:
        modules.add(node.module.split('.')[0])

# Filter to non-stdlib, non-relative modules
import importlib.util
failed = []
for mod in sorted(modules):
    if mod.startswith('_') or mod in ('os', 'sys', 'json', 'logging', 'typing', 'datetime', 'pathlib', 'asyncio', 'abc', 'enum', 'dataclasses', 'collections', 'functools', 'importlib', 'contextlib', 'inspect', 'traceback', 're', 'io', 'copy', 'math', 'time', 'uuid', 'hashlib', 'base64', 'urllib', 'http', 'socket', 'subprocess', 'platform', 'struct', 'itertools', 'operator', 'warnings', 'signal', 'threading', 'multiprocessing', 'concurrent', 'queue', 'pickle', 'shelve', 'tempfile', 'shutil', 'glob', 'fnmatch', 'string', 'textwrap', 'codecs', 'csv', 'configparser', 'argparse', 'getpass', 'secrets', 'hmac', 'ssl', 'email', 'html', 'xml', 'pprint'):
        continue
    spec = importlib.util.find_spec(mod)
    if spec:
        print(f'  ✅ {mod}')
    else:
        failed.append(mod)
        print(f'  ❌ {mod}')

if failed:
    print(f'\n❌ Import verification failed for: {failed}')
    print('Fix: Ensure these are in requirements.txt, then rm -rf env/ and re-run Step 2')
    sys.exit(1)
else:
    print('\n✅ All imports verified')
"

3c. Verify scf_bootstrap

# Check scf_bootstrap exists, is executable, and sets PYTHONPATH
test -f ./scf_bootstrap || { echo "❌ scf_bootstrap not found"; exit 1; }
test -x ./scf_bootstrap || { echo "❌ scf_bootstrap not executable. Run: chmod +x scf_bootstrap"; exit 1; }
grep -q 'PYTHONPATH.*env' ./scf_bootstrap || { echo "❌ scf_bootstrap missing PYTHONPATH=./env"; exit 1; }
echo "✅ scf_bootstrap OK"

All 3 checks passed? → Proceed to Step 4.

---

Step 4: Deploy with manageAgent

manageAgent(action="create", runtime="Python3.10", installDependency=false, targetPath="...")

IMPORTANT: Do NOT add env/ to the ignore list — it must be uploaded with the code.

---

Error Recovery Playbook

Golden Rule: ANY problem with `env/` has exactly ONE fix:

```

edit requirements.txt (if needed) → rm -rf env/ → re-run Step 2 script → re-run Step 3

```

There is NO other fix. Never deviate from this.

Error: pip install reports "no matching distribution"

• Cause: A package doesn't have a manylinux2014_x86_64 wheel for Python 3.10
• Fix: Pin a version in requirements.txt that has a compatible wheel, or check spelling
• Then: rm -rf env/ → re-run Step 2

Error: ModuleNotFoundError at runtime (ANY module)

• Cause 1: The module is missing from requirements.txt → add it
• Cause 2: env/ was built with Python 3.11+ → ensure Python 3.10, rebuild
• Cause 3: env/ was built incrementally (multiple pip installs) → rebuild atomically
• Fix: rm -rf env/ → re-run Step 2

Error: Namespace package submodule missing (e.g., cloudbase_agent.xxx)

• Cause: Multiple pip install commands into env/ caused namespace package fragmentation
• Fix: rm -rf env/ → re-run Step 2 (single command installs all packages atomically)

⛔ PROHIBITED OPERATIONS (will cause deployment failures)

• ⛔ Running a second pip install into an existing env/
• ⛔ Copying files from another project directory into env/
• ⛔ Manually creating or modifying __init__.py inside env/
• ⛔ Deleting selective directories inside env/ and reinstalling partial deps
• ⛔ Using pip install inside scf_bootstrap (wastes cold-start time)

---

Python Runtime Version

Always select Python 3.10 runtime (runtime="Python3.10"). This is the recommended version for CloudBase Agent Python SDK because:

• Full compatibility with all cloudbase-agent-* packages
• Best performance for async/await patterns used by FastAPI
• Stable and well-tested on the CloudBase platform

Do NOT use Python 3.9 or earlier — many SDK features require Python >= 3.10.

Code Adaptation Notes

Port Listening

Your server must listen on the port from environment variable SCF_RUNTIME_PORT:

import os
from cloudbase_agent.server import AgentServiceApp

port = int(os.environ.get("SCF_RUNTIME_PORT", "9000"))
AgentServiceApp().run(create_agent, port=port, host="0.0.0.0")

Startup Script

The startup script must be named scf_bootstrap (no file extension), placed in the project root, and have executable permissions:

#!/bin/bash
export PYTHONPATH="./env:$PYTHONPATH"
/var/lang/python310/bin/python3 -u server.py

Make it executable:

chmod +x scf_bootstrap

CORS Configuration

Ensure CORS is properly configured for cross-origin requests:

from cloudbase_agent.server import AgentServiceApp

app = AgentServiceApp()
app.set_cors_config(allow_origins=["*"])
app.run(create_agent, port=port)

Or if using FastAPI directly:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Complete Deployment Example

Project Structure

my-agent/
├── agents/
│   └── chat/agent.py        # Agent workflow (any framework)
├── env/                      # Pre-installed dependencies (built by Step 2)
├── server.py                 # Main entry point
├── scf_bootstrap             # CloudBase startup script
├── requirements.txt          # Dependencies
└── .env                      # Environment variables (local only)

scf_bootstrap

#!/bin/bash
export PYTHONPATH="./env:$PYTHONPATH"
/var/lang/python310/bin/python3 -u server.py

requirements.txt (example — varies by framework)

# Core (always needed)
cloudbase-agent-server
python-dotenv

# Framework adapter (pick ONE based on your choice)
cloudbase-agent-langgraph   # For LangGraph-based agents
# cloudbase-agent-crewai    # For CrewAI-based agents
# cloudbase-agent-coze      # For Coze platform agents

# LLM provider (example)
langchain-openai

When to Use CloudRun Instead

Despite HTTP Cloud Functions being preferred, use CloudRun in these cases:

• Custom Docker image required (special system-level dependencies like FFmpeg, Chromium, etc.)
• Resource requirements exceed Cloud Function limits
• Persistent local file storage needed
• Need to install native C extensions that require specific OS packages

For CloudRun deployment, use a Dockerfile with Python 3.11:

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

ENV PORT=9000
CMD ["python", "server.py"]

Summary

CloudBase Agent Observability Reference

Overview

CloudBase Agent provides comprehensive observability features including logging, metrics, and distributed tracing.

Logging

Basic Configuration

from cloudbase_agent.server import create_server
import logging

server = create_server(
    log_level="INFO",
    log_format="json",  # or "text"
    log_output="stdout"  # or file path
)

Structured Logging

from cloudbase_agent.server.logging import get_logger

logger = get_logger(__name__)

# Structured log with context
logger.info(
    "Agent request received",
    extra={
        "conversation_id": "conv_123",
        "user_id": "user_456",
        "agent_type": "react",
        "duration_ms": 150
    }
)

Log Levels

logger.debug("Detailed debugging information")
logger.info("General information")
logger.warning("Warning messages")
logger.error("Error messages", exc_info=True)
logger.critical("Critical errors")

Metrics

Prometheus Metrics

from cloudbase_agent.server.metrics import (
    Counter,
    Histogram,
    Gauge,
    Summary
)

# Define metrics
requests_total = Counter(
    "agent_requests_total",
    "Total agent requests",
    ["agent_type", "status"]
)

request_duration = Histogram(
    "agent_request_duration_seconds",
    "Request duration in seconds",
    ["agent_type"],
    buckets=[0.1, 0.5, 1.0, 2.5, 5.0, 10.0]
)

active_conversations = Gauge(
    "active_conversations",
    "Number of active conversations"
)

# Use metrics
requests_total.labels(agent_type="react", status="success").inc()
request_duration.labels(agent_type="react").observe(1.23)
active_conversations.set(42)

Metrics Endpoint

from cloudbase_agent.server import create_server

server = create_server(
    enable_metrics=True,
    metrics_path="/metrics"  # Default Prometheus endpoint
)

Custom Metrics

from cloudbase_agent.server.metrics import register_metric

# Register custom metric
tool_calls = Counter(
    "agent_tool_calls_total",
    "Total tool calls",
    ["tool_name", "status"]
)
register_metric(tool_calls)

# Use in tool
@tool
def my_tool(param: str) -> dict:
    try:
        result = do_work(param)
        tool_calls.labels(tool_name="my_tool", status="success").inc()
        return result
    except Exception as e:
        tool_calls.labels(tool_name="my_tool", status="error").inc()
        raise

Distributed Tracing

OpenTelemetry Setup

from cloudbase_agent.server.tracing import configure_tracing

configure_tracing(
    service_name="my-agent-service",
    exporter="otlp",  # or "jaeger", "zipkin"
    endpoint="http://localhost:4317",
    sample_rate=1.0  # Sample all traces (0.0 to 1.0)
)

Automatic Instrumentation

from cloudbase_agent.server import create_server

# Enable automatic tracing
server = create_server(
    enable_tracing=True,
    trace_agent_runs=True,
    trace_tool_calls=True,
    trace_llm_calls=True
)

Manual Tracing

from cloudbase_agent.server.tracing import trace, get_current_span

@trace(name="custom_operation")
async def custom_operation(param: str):
    # Current span auto-created
    span = get_current_span()
    span.set_attribute("param_length", len(param))
    
    # Nested spans
    with trace("sub_operation"):
        result = await sub_operation(param)
    
    span.set_attribute("result_size", len(result))
    return result

Trace Context Propagation

from cloudbase_agent.server.tracing import inject_trace_context, extract_trace_context

# Inject context into HTTP headers
headers = {}
inject_trace_context(headers)

# Make HTTP request with context
async with httpx.AsyncClient() as client:
    response = await client.get(url, headers=headers)

# Extract context from incoming request
context = extract_trace_context(request.headers)

Agent Run Tracking

Automatic Tracking

from cloudbase_agent.langgraph import create_react_agent

# Automatic run tracking enabled
agent = create_react_agent(
    model=model,
    tools=tools,
    enable_observability=True
)

# Each run automatically tracked with:
# - Run ID
# - Duration
# - Token usage
# - Tool calls
# - Errors

Custom Run Metadata

from cloudbase_agent.server.observability import track_run

@track_run(
    run_type="react_agent",
    metadata={"version": "1.0.0"}
)
async def invoke_agent(input_data: dict):
    result = await agent.ainvoke(input_data)
    return result

Error Tracking

Sentry Integration

from cloudbase_agent.server.errors import configure_error_tracking

configure_error_tracking(
    dsn="https://xxx@sentry.io/xxx",
    environment="production",
    release="1.0.0",
    traces_sample_rate=0.1
)

Error Context

from cloudbase_agent.server.errors import capture_exception, set_error_context

set_error_context({
    "conversation_id": "conv_123",
    "user_id": "user_456"
})

try:
    result = risky_operation()
except Exception as e:
    capture_exception(e, extra={
        "operation": "risky_operation",
        "input": input_data
    })
    raise

Health Checks

Health Check Endpoint

from cloudbase_agent.server import create_server

server = create_server(
    enable_health_check=True,
    health_check_path="/health"
)

Custom Health Checks

from cloudbase_agent.server.health import HealthCheck, HealthStatus

class DatabaseHealthCheck(HealthCheck):
    name = "database"
    
    async def check(self) -> HealthStatus:
        try:
            await db.execute("SELECT 1")
            return HealthStatus.HEALTHY
        except Exception as e:
            return HealthStatus.UNHEALTHY, str(e)

# Register
server.add_health_check(DatabaseHealthCheck())

Performance Monitoring

APM Integration

from cloudbase_agent.server.apm import configure_apm

configure_apm(
    service_name="my-agent",
    server_url="http://apm-server:8200",
    environment="production"
)

Performance Metrics

from cloudbase_agent.server.metrics import track_performance

@track_performance(metric_name="agent_processing")
async def process_request(data: dict):
    # Automatically tracks:
    # - Duration
    # - Memory usage
    # - CPU time
    return await agent.ainvoke(data)

Dashboard Integration

Grafana Dashboard

CloudBase Agent provides pre-built Grafana dashboards:

# Import dashboard
curl -X POST http://grafana:3000/api/dashboards/import \
  -H "Content-Type: application/json" \
  -d @dashboards/cloudbase-agent-overview.json

Custom Dashboards

Key metrics to monitor:

• agent_requests_total - Request volume
• agent_request_duration_seconds - Latency
• agent_errors_total - Error rate
• active_conversations - Concurrent users
• llm_tokens_total - Token usage
• tool_calls_total - Tool usage

Best Practices

1. Structured Logging: Always use structured logs with context 2. Metrics Labels: Use consistent label names across metrics 3. Trace Sampling: Adjust sample rate based on traffic volume 4. Error Context: Include relevant context when capturing errors 5. Health Checks: Implement health checks for all dependencies 6. Alerts: Set up alerts for critical metrics

Common Patterns

Request Tracking

from cloudbase_agent.server.observability import RequestTracker

async def handle_request(request):
    tracker = RequestTracker(request)
    
    try:
        result = await process_request(request.data)
        tracker.success(result)
        return result
    except Exception as e:
        tracker.error(e)
        raise
    finally:
        tracker.finalize()

Performance Profiling

from cloudbase_agent.server.profiling import profile

@profile(enabled=True)
async def expensive_operation(data):
    # Automatically profiles:
    # - Function calls
    # - Memory allocations
    # - I/O operations
    return await process(data)

Troubleshooting

High Latency

1. Check agent_request_duration_seconds histogram 2. Review trace spans to identify slow operations 3. Monitor llm_response_time metrics 4. Check tool execution times

Error Spikes

1. Check agent_errors_total counter 2. Review error logs with level=error 3. Check Sentry for error details 4. Analyze error traces

Memory Issues

1. Monitor process_memory_bytes gauge 2. Check for memory leaks in traces 3. Review conversation storage TTL settings 4. Analyze heap dumps if needed

See Also

• Server Reference - Server configuration
• Storage Reference - Storage monitoring
• Recipes - Observability patterns

Building Custom Adapters

An adapter bridges your AI framework to the AG-UI protocol. It converts AG-UI input (messages, tools, state) into your framework's format, and converts your framework's streaming output into AG-UI events.

Prerequisites: Deep understanding of both your AI framework's API and the AG-UI protocol events.

When to build your own: No existing adapter for your framework (check AG-UI ecosystem first).

Extend AbstractAgent and implement run() that returns Observable<BaseEvent>.

Structure

import { AbstractAgent, RunAgentInput, BaseEvent, EventType } from "@ag-ui/client";
import { Observable, Subscriber } from "rxjs";

export class MyAdapter extends AbstractAgent {
  run(input: RunAgentInput): Observable<BaseEvent> {
    return new Observable((subscriber) => this._run(subscriber, input));
  }

  private async _run(subscriber: Subscriber<BaseEvent>, input: RunAgentInput) {
    const { messages, runId, threadId, tools } = input;

    subscriber.next({ type: EventType.RUN_STARTED, threadId, runId });

    try {
      // 1. Convert AG-UI input to your framework's format
      // 2. Call your framework
      // 3. Convert your framework's output to AG-UI events (see Event Sequence below)

      subscriber.next({ type: EventType.RUN_FINISHED, threadId, runId });
    } catch (error) {
      subscriber.next({ type: EventType.RUN_ERROR, message: error.message });
    }
    subscriber.complete();
  }
}

Event Sequence (Brief)

Text: TEXT_MESSAGE_START → TEXT_MESSAGE_CONTENT (repeat) → TEXT_MESSAGE_END

Tool call: TOOL_CALL_START → TOOL_CALL_ARGS → TOOL_CALL_END

Tool result (server-executed tools only): TOOL_CALL_RESULT

Always emit full lifecycle. parentMessageId links tool calls to their parent message.

For complete event reference, see AG-UI Protocol.

Building UI Clients

Connect your UI to AG-UI endpoints served by @cloudbase/agent-server.

Web Applications

Use @ag-ui/client (official AG-UI SDK):

npm install @ag-ui/client@latest

import { HttpAgent } from "@ag-ui/client";

const agent = new HttpAgent({ url: "http://localhost:9000/send-message" });

for await (const event of agent.run({
  threadId: "thread-1",
  runId: "run-1",
  messages: [{ id: "m1", role: "user", content: "Hello" }]
})) {
  console.log(event.type, event);
}

See AG-UI documentation for full API: https://docs.ag-ui.com

WeChat Mini Program

Use @cloudbase/agent-ui-miniprogram (headless behavior mixin):

npm install @cloudbase/agent-ui-miniprogram@latest

import { createAGUIBehavior, CloudbaseTransport } from "@cloudbase/agent-ui-miniprogram";

Component({
  behaviors: [createAGUIBehavior({
    transport: new CloudbaseTransport({ botId: "your-bot-id" })
  })],
  methods: {
    onSend() {
      this.agui.sendMessage(this.data.inputText);
    }
  }
});
// State: this.data.agui.uiMessages, this.data.agui.isRunning

Beyond basic usage, the package offers more createAGUIBehavior options, this.agui.* namespace methods, state getters, and UIMessage format for rendering.