Cloudbase

When to use this skill

Use this skill for calling AI models in Node.js backend or CloudBase cloud functions using @cloudbase/node-sdk.

Use it when you need to:

• Integrate AI text generation in backend services
• Generate images with Hunyuan Image model
• Call AI models from CloudBase cloud functions
• Server-side AI processing

Do NOT use for:

• Browser/Web apps → use ai-model-web skill
• WeChat Mini Program → use ai-model-wechat skill
• HTTP API integration → use http-api skill

---

Available Providers and Models

CloudBase provides these built-in providers and models:

---

Installation

npm install @cloudbase/node-sdk

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

---

Initialization

In Cloud Functions

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 language model generation), set a longer timeout as these operations can be slow.

Using 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 the timeout parameter in the func object:

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

Recommended timeout values:

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

In 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

const model = ai.createModel("hunyuan-exp");

const result = await model.generateText({
  model: "hunyuan-2.0-instruct-20251111",  // Recommended model
  messages: [{ role: "user", content: "你好，请你介绍一下李白" }],
});

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

---

streamText() - Streaming

const model = ai.createModel("hunyuan-exp");

const res = await model.streamText({
  model: "hunyuan-2.0-instruct-20251111",  // Recommended model
  messages: [{ role: "user", content: "你好，请你介绍一下李白" }],
});

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

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

// Option 3: Get 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 Node SDK, not in JS SDK (Web) or WeChat Mini Program.

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

const res = await imageModel.generateImage({
  model: "hunyuan-image",
  prompt: "一只可爱的猫咪在草地上玩耍",
  size: "1024x1024",
  version: "v1.9",
});

console.log(res.data[0].url);           // Image URL (valid 24 hours)
console.log(res.data[0].revised_prompt);// Revised prompt if 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;
}

Authentication Activation Checklist

Use this checklist before generating any CloudBase authentication flow.

When this checklist applies

• Web login or registration
• SMS, email, anonymous, 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

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 service listens on port 9000. 4. Confirm the function root path points to the parent directory, not the function directory itself. 5. 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.
• 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.

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=8000)

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=8000)

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", "8000")),
        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=8000

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

docker build -t coze-agent .
docker run -p 8000:8000 \
  -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": 8000,
    "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=8000)

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]

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

Authentication and User Context

This guide explains how to implement authentication and manage user context in CloudBase Agent Python SDK using the framework's reserved fields pattern.

Overview

CloudBase Agent uses a standardized approach for passing user context through the request lifecycle:

HTTP Request (JWT in header)
  ↓ (middleware extracts)
state["__request_context__"]["user"]["id"]
state["__request_context__"]["user"]["jwt"]
  ↓ (available to)
Agent / Adapter / Tools

Framework Reserved Fields

CloudBase Agent reserves specific fields in state for user authentication:

⚠️ Security Warning: These fields are set by authentication middleware and should be treated as read-only. Modifying them in your agent logic may lead to security vulnerabilities.

Implementation Pattern

1. Authentication Middleware (Write)

Middleware extracts user information from the request and injects it into state:

import jwt
from fastapi import Request
from cloudbase_agent.server.send_message.models import RunAgentInput
from typing import Generator

def auth_middleware(
    input_data: RunAgentInput, 
    request: Request
) -> Generator[None, None, None]:
    """
    Extract user from JWT and inject into state.
    
    This middleware:
    1. Extracts JWT from Authorization header
    2. Verifies the token
    3. Injects user info into framework reserved fields
    """
    # Extract token
    auth_header = request.headers.get("Authorization", "")
    token = auth_header.replace("Bearer ", "")
    
    if token:
        try:
            # Verify JWT (use your own secret and algorithm)
            jwt_payload = jwt.decode(
                token, 
                "your-secret-key", 
                algorithms=["HS256"]
            )
            
            # Initialize state if needed
            if input_data.state is None:
                input_data.state = {}
            
            # ✅ Inject into framework reserved fields
            input_data.state["__request_context__"] = {
                "user": {
                    "id": jwt_payload["sub"],  # User ID from JWT sub claim
                    "jwt": jwt_payload         # Full JWT payload
                }
            }
            
        except jwt.InvalidTokenError as e:
            # Handle invalid token (log but don't block in this example)
            print(f"Invalid JWT token: {e}")
            # You could raise an exception here to block the request
            # raise InvalidRequestError(message="Invalid authentication token")
    
    yield  # Continue to next middleware or agent

2. Register Middleware

from cloudbase_agent.server import AgentServiceApp

app = AgentServiceApp()
app.use(auth_middleware)  # Register before run()
app.run(create_agent, port=8000)

3. Adapter/Agent (Read)

In your adapter or agent, read the user information:

def get_user_id_from_state(state: dict) -> str:
    """
    Safely extract user ID from framework reserved field.
    
    :param state: Agent state dictionary
    :return: User ID string
    :raises ValueError: If user ID not found
    """
    request_context = state.get("__request_context__", {})
    user = request_context.get("user", {})
    user_id = user.get("id")
    
    if not user_id:
        raise ValueError(
            "user_id is required but not found in "
            "state.__request_context__.user.id. "
            "Please ensure auth middleware is registered."
        )
    
    return user_id

# Usage in agent
def my_agent_function(state: dict):
    user_id = get_user_id_from_state(state)
    jwt_payload = state.get("__request_context__", {}).get("user", {}).get("jwt", {})
    
    # Use user_id and jwt_payload for your logic
    user_data = fetch_user_data(user_id)
    # ...

Example: Complete Authentication Flow

Step 1: Define Authentication Middleware

# auth.py
import jwt
from fastapi import Request, HTTPException
from cloudbase_agent.server.send_message.models import RunAgentInput
from cloudbase_agent.server.errors.exceptions import InvalidRequestError
from typing import Generator

# Your JWT configuration
JWT_SECRET = "your-secret-key"
JWT_ALGORITHM = "HS256"

def jwt_auth_middleware(
    input_data: RunAgentInput,
    request: Request
) -> Generator[None, None, None]:
    """
    JWT authentication middleware.
    
    Extracts and verifies JWT, then injects user info into state.
    Raises error if token is invalid or missing for protected routes.
    """
    # Extract Authorization header
    auth_header = request.headers.get("Authorization", "")
    
    if not auth_header:
        raise InvalidRequestError(
            message="Missing Authorization header",
            details={"header": "Authorization"}
        )
    
    # Parse Bearer token
    if not auth_header.startswith("Bearer "):
        raise InvalidRequestError(
            message="Invalid Authorization header format. Expected 'Bearer <token>'",
            details={"format": "Bearer <token>"}
        )
    
    token = auth_header[7:]  # Remove "Bearer " prefix
    
    try:
        # Verify and decode JWT
        jwt_payload = jwt.decode(
            token,
            JWT_SECRET,
            algorithms=[JWT_ALGORITHM]
        )
        
        # Validate required claims
        if "sub" not in jwt_payload:
            raise InvalidRequestError(
                message="JWT missing 'sub' claim",
                details={"claim": "sub"}
            )
        
        # Initialize state if needed
        if input_data.state is None:
            input_data.state = {}
        
        # Inject user info into framework reserved fields
        input_data.state["__request_context__"] = {
            "user": {
                "id": jwt_payload["sub"],
                "jwt": jwt_payload
            }
        }
        
        # Log successful authentication (optional)
        print(f"Authenticated user: {jwt_payload['sub']}")
        
    except jwt.ExpiredSignatureError:
        raise InvalidRequestError(
            message="JWT token has expired",
            details={"error": "expired"}
        )
    except jwt.InvalidTokenError as e:
        raise InvalidRequestError(
            message=f"Invalid JWT token: {str(e)}",
            details={"error": "invalid_token"}
        )
    
    yield  # Continue to agent execution

Step 2: Use in Coze Adapter

# agent.py
from cloudbase_agent.coze import CozeAgentAdapter
from cloudbase_agent.server import AgentServiceApp
from auth import jwt_auth_middleware

def create_agent():
    """
    Create Coze agent.
    User ID will be automatically extracted from state by the adapter.
    """
    return CozeAgentAdapter(
        bot_id="your-bot-id",
        api_key="your-api-key"
    )

# Start server with auth middleware
app = AgentServiceApp()
app.use(jwt_auth_middleware)  # Register auth middleware
app.run(create_agent, port=8000)

Step 3: Coze Adapter Internal Logic

The Coze adapter reads user ID automatically:

# Inside cloudbase_agent.coze.agent.py (framework code)
class CozeAgentAdapter:
    def _get_user_id(self, run_input: RunAgentInput) -> str:
        """Get user_id from state.__request_context__.user.id."""
        state = run_input.state or {}
        
        # Read from framework reserved field
        request_context = state.get("__request_context__", {})
        user_info = request_context.get("user", {})
        user_id = user_info.get("id")
        
        if not user_id:
            raise ValueError(
                "user_id is required but not found in "
                "state.__request_context__.user.id. "
                "Please ensure auth middleware is registered."
            )
        
        return user_id.strip()

Custom User Context Fields

You can add custom fields alongside framework reserved fields:

def auth_middleware(input_data: RunAgentInput, request: Request):
    """Auth middleware with custom fields."""
    # Verify JWT
    jwt_payload = verify_jwt(extract_token(request))
    
    # Initialize state
    if input_data.state is None:
        input_data.state = {}
    
    # Set framework reserved fields + custom fields
    input_data.state["__request_context__"] = {
        "user": {
            "id": jwt_payload["sub"],       # ← Framework reserved
            "jwt": jwt_payload,             # ← Framework reserved
        },
        # ✅ Custom fields (allowed)
        "tenant_id": jwt_payload.get("tenant_id"),
        "permissions": jwt_payload.get("permissions", []),
        "session_id": request.headers.get("X-Session-ID"),
    }
    
    yield

# Usage in agent
def my_agent(state: dict):
    # Read framework reserved fields
    user_id = state["__request_context__"]["user"]["id"]
    
    # Read custom fields
    tenant_id = state["__request_context__"].get("tenant_id")
    permissions = state["__request_context__"].get("permissions", [])
    
    if "admin" not in permissions:
        raise PermissionError("Admin permission required")

Security Best Practices

1. Use Strong Secrets

import os

JWT_SECRET = os.environ.get("JWT_SECRET_KEY")
if not JWT_SECRET or len(JWT_SECRET) < 32:
    raise ValueError("JWT_SECRET_KEY must be at least 32 characters")

2. Validate All Claims

def validate_jwt_payload(payload: dict) -> None:
    """Validate JWT payload structure."""
    required_claims = ["sub", "exp", "iat"]
    
    for claim in required_claims:
        if claim not in payload:
            raise ValueError(f"Missing required claim: {claim}")
    
    # Validate expiration (PyJWT does this automatically, but double-check)
    import time
    if payload["exp"] < time.time():
        raise ValueError("Token expired")

3. Implement Token Refresh

def refresh_token_middleware(input_data, request):
    """Check token expiration and handle refresh."""
    jwt_payload = input_data.state.get("__request_context__", {}).get("user", {}).get("jwt", {})
    
    # Check if token expires soon (e.g., within 5 minutes)
    if jwt_payload.get("exp", 0) - time.time() < 300:
        # Add header to response suggesting refresh
        request.state.should_refresh_token = True
    
    yield

4. Rate Limit by User

from collections import defaultdict
from time import time

user_request_counts = defaultdict(list)

def rate_limit_by_user_middleware(input_data, request):
    """Rate limit per user ID."""
    user_id = input_data.state.get("__request_context__", {}).get("user", {}).get("id")
    
    if user_id:
        now = time()
        # Clean old requests
        user_request_counts[user_id] = [
            t for t in user_request_counts[user_id]
            if now - t < 60  # 1-minute window
        ]
        
        if len(user_request_counts[user_id]) >= 10:
            raise Exception(f"Rate limit exceeded for user {user_id}")
        
        user_request_counts[user_id].append(now)
    
    yield

Testing Authentication

Unit Test

import pytest
from fastapi import Request
from cloudbase_agent.server.send_message.models import RunAgentInput
from auth import jwt_auth_middleware

def test_auth_middleware_with_valid_token():
    """Test middleware with valid JWT."""
    # Create mock request with valid token
    token = create_test_jwt({"sub": "user123"})
    request = Request(scope={
        "type": "http",
        "headers": [(b"authorization", f"Bearer {token}".encode())]
    })
    
    input_data = RunAgentInput(
        messages=[],
        runId="test-run",
        threadId="test-thread"
    )
    
    # Execute middleware
    gen = jwt_auth_middleware(input_data, request)
    next(gen)
    
    # Verify user info was injected
    assert input_data.state["__request_context__"]["user"]["id"] == "user123"

def test_auth_middleware_with_missing_token():
    """Test middleware rejects missing token."""
    request = Request(scope={"type": "http", "headers": []})
    input_data = RunAgentInput(messages=[], runId="test", threadId="test")
    
    with pytest.raises(InvalidRequestError):
        gen = jwt_auth_middleware(input_data, request)
        next(gen)

Integration Test

from fastapi.testclient import TestClient

def test_authenticated_request():
    """Test full request with authentication."""
    client = TestClient(app)
    
    token = create_test_jwt({"sub": "user123"})
    
    response = client.post(
        "/send-message",
        json={"messages": [{"role": "user", "content": "Hello"}]},
        headers={"Authorization": f"Bearer {token}"}
    )
    
    assert response.status_code == 200

Migration from forwarded_props

If you're migrating from the old forwarded_props pattern:

Before (Old Pattern)

# ❌ Old: forwarded_props
def create_jwt_preprocessor():
    def jwt_preprocessor(request: RunAgentInput, http_context: Request):
        user_id = extract_user_id_from_request(http_context)
        if not request.forwarded_props:
            request.forwarded_props = {}
        request.forwarded_props["user_id"] = user_id
    return jwt_preprocessor

After (New Pattern)

# ✅ New: state.__request_context__
def auth_middleware(input_data: RunAgentInput, request: Request):
    user_id = extract_user_id_from_request(request)
    
    if input_data.state is None:
        input_data.state = {}
    
    input_data.state["__request_context__"] = {
        "user": {"id": user_id}
    }
    yield

Summary

Next Steps

• Learn about server deployment
• Understand middleware patterns
• Integrate with Coze adapter
• Build UI clients

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

CloudBase Agent Recipes

Common patterns and complete examples for building agents with CloudBase Agent.

Recipe 1: Basic Chat Agent

Complete example of a simple conversational agent.

from cloudbase_agent.server import create_server, tool
from cloudbase_agent.langgraph import create_react_agent
from langchain_openai import ChatOpenAI

# Define tools
@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"Weather in {city}: Sunny, 22°C"

# Create model and agent
model = ChatOpenAI(model="gpt-4")
agent = create_react_agent(
    model=model,
    tools=[get_weather]
)

# Create server
server = create_server()
server.add_agent("/chat", agent)

if __name__ == "__main__":
    server.run(host="0.0.0.0", port=8000)

Recipe 2: Multi-Agent System

Orchestrate multiple specialized agents.

from cloudbase_agent.langgraph import create_react_agent, create_router_agent

# Specialized agents
research_agent = create_react_agent(
    model=model,
    tools=[search_web, read_document],
    system_message="You are a research assistant."
)

writing_agent = create_react_agent(
    model=model,
    tools=[grammar_check, format_text],
    system_message="You are a writing assistant."
)

# Router agent
router = create_router_agent(
    agents={
        "research": research_agent,
        "writing": writing_agent
    },
    model=model
)

server.add_agent("/assistant", router)

Recipe 3: Persistent Conversations

Maintain conversation history across sessions.

from cloudbase_agent.server.storage import RedisStorage, ConversationStorage
from cloudbase_agent.langgraph import create_checkpointer

# Setup storage
storage = RedisStorage(url="redis://localhost:6379")
conv_storage = ConversationStorage(storage)
checkpointer = create_checkpointer(storage)

# Create agent with persistence
agent = create_react_agent(
    model=model,
    tools=tools,
    checkpointer=checkpointer
)

# Handle request with conversation ID
@server.post("/chat")
async def chat(request):
    conversation_id = request.conversation_id
    
    # Load conversation
    messages = await conv_storage.load_conversation(conversation_id)
    
    # Invoke agent
    result = await agent.ainvoke(
        {"messages": messages + [request.message]},
        config={"configurable": {"thread_id": conversation_id}}
    )
    
    # Save conversation
    await conv_storage.save_conversation(
        conversation_id,
        messages + [request.message, result["messages"][-1]]
    )
    
    return result

Recipe 4: Streaming Responses

Stream agent responses in real-time.

from cloudbase_agent.server import StreamingResponse

@server.post("/chat/stream")
async def chat_stream(request):
    async def generate():
        async for chunk in agent.astream(request.data):
            # Yield SSE format
            yield f"data: {json.dumps(chunk)}\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

Recipe 5: Human-in-the-Loop

Implement approval workflows.

from cloudbase_agent.langgraph import interrupt
from cloudbase_agent.server.approval import ApprovalManager

approval_manager = ApprovalManager(storage)

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email (requires approval)."""
    # Request approval
    approval_id = interrupt(
        "email_approval",
        data={"to": to, "subject": subject, "body": body}
    )
    
    # Wait for approval
    approved = approval_manager.wait_for_approval(approval_id)
    
    if approved:
        # Actually send email
        email_service.send(to, subject, body)
        return "Email sent successfully"
    else:
        return "Email cancelled by user"

# Approval endpoint
@server.post("/approve/{approval_id}")
async def approve(approval_id: str, approved: bool):
    await approval_manager.set_approval(approval_id, approved)
    return {"status": "ok"}

Recipe 6: Tool-Based Generative UI

Generate UI components based on tool results.

from cloudbase_agent.server import tool, ui_component

@tool
@ui_component("chart")
def analyze_data(dataset: str) -> dict:
    """Analyze dataset and return chart data."""
    data = load_dataset(dataset)
    analysis = perform_analysis(data)
    
    return {
        "type": "line_chart",
        "data": analysis["timeseries"],
        "config": {
            "xAxis": "date",
            "yAxis": "value",
            "title": f"Analysis of {dataset}"
        }
    }

# Frontend receives:
# {
#   "tool": "analyze_data",
#   "result": {...},
#   "ui": {
#     "component": "chart",
#     "props": {...}
#   }
# }

Recipe 7: Rate Limiting

Implement request rate limiting.

from cloudbase_agent.server.middleware import RateLimiter

rate_limiter = RateLimiter(
    storage=storage,
    requests_per_minute=60,
    burst=10
)

@server.post("/chat")
@rate_limiter.limit(key=lambda req: req.user_id)
async def chat(request):
    return await agent.ainvoke(request.data)

Recipe 8: Authentication & Authorization

Secure agent endpoints.

from cloudbase_agent.server.auth import APIKeyAuth, JWTAuth

# API Key auth
api_key_auth = APIKeyAuth(
    storage=storage,
    header="X-API-Key"
)

# JWT auth
jwt_auth = JWTAuth(
    secret="your-secret-key",
    algorithm="HS256"
)

@server.post("/chat")
@api_key_auth.require()
async def chat(request):
    user_id = request.auth.user_id
    return await agent.ainvoke(request.data)

@server.post("/admin/chat")
@jwt_auth.require(roles=["admin"])
async def admin_chat(request):
    return await admin_agent.ainvoke(request.data)

Recipe 9: Error Recovery

Implement robust error handling.

from cloudbase_agent.server.errors import AgentError, ToolError
from tenacity import retry, stop_after_attempt, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type(ToolError)
)
async def invoke_with_retry(agent, input_data):
    try:
        return await agent.ainvoke(input_data)
    except ToolError as e:
        logger.warning(f"Tool error, retrying: {e}")
        raise
    except AgentError as e:
        logger.error(f"Agent error: {e}")
        return {"error": str(e), "fallback": "default_response"}

Recipe 10: Monitoring & Alerting

Complete observability setup.

from cloudbase_agent.server.observability import setup_observability
from cloudbase_agent.server.metrics import Counter, Histogram

# Setup
setup_observability(
    service_name="my-agent",
    enable_tracing=True,
    enable_metrics=True,
    enable_logging=True
)

# Custom metrics
agent_errors = Counter(
    "agent_errors_total",
    "Total agent errors",
    ["error_type"]
)

# Middleware
@server.middleware("http")
async def observability_middleware(request, call_next):
    with track_request(request):
        try:
            response = await call_next(request)
            return response
        except Exception as e:
            agent_errors.labels(error_type=type(e).__name__).inc()
            raise

# Alerts (example with Prometheus Alertmanager)
# rules.yml:
# - alert: HighErrorRate
#   expr: rate(agent_errors_total[5m]) > 0.1
#   annotations:
#     summary: "High error rate detected"

Recipe 11: Background Tasks

Process long-running tasks asynchronously.

from cloudbase_agent.server.tasks import TaskQueue

task_queue = TaskQueue(storage=storage)

@task_queue.task(name="process_document")
async def process_document(doc_id: str):
    """Process a document in the background."""
    document = load_document(doc_id)
    result = await agent.ainvoke({
        "task": "analyze",
        "document": document
    })
    save_result(doc_id, result)
    return result

@server.post("/documents/process")
async def submit_document(request):
    task_id = await process_document.delay(request.doc_id)
    return {"task_id": task_id, "status": "processing"}

@server.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    status = await task_queue.get_status(task_id)
    return status

Recipe 12: Multi-Modal Agent

Handle text, images, and other media.

from cloudbase_agent.server import tool
from langchain_openai import ChatOpenAI

@tool
def analyze_image(image_url: str) -> str:
    """Analyze an image and describe its contents."""
    # Vision model
    vision_model = ChatOpenAI(model="gpt-4-vision-preview")
    result = vision_model.invoke([
        {"type": "image_url", "image_url": image_url},
        {"type": "text", "text": "What's in this image?"}
    ])
    return result.content

# Multi-modal agent
agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4-vision-preview"),
    tools=[analyze_image, search_web]
)

See Also

• Server Reference - Server API details
• LangGraph Reference - Agent patterns
• Tools Reference - Tool system
• Storage Reference - Data persistence
• Observability Reference - Monitoring

Server Reference (cloudbase_agent.server)

FastAPI-based HTTP server with dual-protocol support (AG-UI + OpenAI).

Exports

Three Deployment Methods

Method 1: One-line (simplest)

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

Method 2: Build + customize (recommended for multi-agent)

app = AgentServiceApp()
fastapi_app = app.build(
    create_agent,
    base_path="/api",
    enable_openai_endpoint=True,
    enable_healthz=True,
)
# Add custom routes to fastapi_app...
uvicorn.run(fastapi_app, host="0.0.0.0", port=9000)

Method 3: Core adapters (maximum flexibility)

from fastapi import FastAPI
from cloudbase_agent.server import create_send_message_adapter, create_openai_adapter

app = FastAPI()

@app.post("/my-agent/send-message")
async def send_message(request: RunAgentInput):
    return await create_send_message_adapter(create_my_agent, request)

@app.post("/my-agent/chat/completions")
async def chat(request: OpenAIChatCompletionRequest):
    return await create_openai_adapter(create_my_agent, request)

AgentServiceApp Constructor

AgentServiceApp(
    observability=None,  # Optional[ObservabilityConfig | List[ObservabilityConfig]]
)

AgentServiceApp Methods

Middleware Pattern

Middlewares use Python's generator pattern with yield — code before yield runs pre-processing, code after yield runs post-processing (onion model).

def my_middleware(input_data: RunAgentInput, request: Request):
    # Pre-processing (runs before agent)
    auth = request.headers.get("Authorization")
    if auth and auth.startswith("Bearer "):
        if not input_data.forwarded_props:
            input_data.forwarded_props = {}
        input_data.forwarded_props["user_id"] = decode_jwt(auth[7:])
    
    yield  # Control passes to agent
    
    # Post-processing (runs after agent, optional)
    print("Request completed")

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

Agent Creator Pattern

Factory function called per-request. Supports optional cleanup callback:

def create_agent() -> AgentCreatorResult:
    db = connect_database()
    agent = LangGraphAgent(graph=workflow, name="my-agent")
    
    def cleanup():
        db.close()  # Guaranteed to run after stream completes
    
    return {"agent": agent, "cleanup": cleanup}

Multi-Agent Server

Option A: Manual routes

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from cloudbase_agent.server import create_send_message_adapter, create_openai_adapter, RunAgentInput, OpenAIChatCompletionRequest
from cloudbase_agent.server.errors import install_exception_handlers

app = FastAPI(title="Multi-Agent Server")
install_exception_handlers(app)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

@app.post("/agentic_chat/send-message")
async def chat_send(request: RunAgentInput):
    return await create_send_message_adapter(create_chat_agent, request)

@app.post("/agentic_chat/chat/completions")
async def chat_openai(request: OpenAIChatCompletionRequest):
    return await create_openai_adapter(create_chat_agent, request)

@app.post("/human_in_the_loop/send-message")
async def hitl_send(request: RunAgentInput):
    return await create_send_message_adapter(create_hitl_agent, request)

Option B: Mount sub-apps

main_app = FastAPI()
chat_app = AgentServiceApp().build(create_chat_agent, enable_openai_endpoint=True)
hitl_app = AgentServiceApp().build(create_hitl_agent, enable_openai_endpoint=True)
main_app.mount("/agentic_chat", chat_app)
main_app.mount("/human_in_the_loop", hitl_app)

CloudBase Agent Storage System Reference

Overview

CloudBase Agent provides a unified storage interface for managing agent state, conversation history, and persistent data.

Storage Interface

Basic Operations

from cloudbase_agent.server.storage import Storage

# Initialize storage
storage = Storage(backend="redis", url="redis://localhost:6379")

# Store data
await storage.set("key", {"data": "value"})

# Retrieve data
data = await storage.get("key")

# Delete data
await storage.delete("key")

# Check existence
exists = await storage.exists("key")

Storage Backends

Redis Backend

from cloudbase_agent.server.storage import RedisStorage

storage = RedisStorage(
    url="redis://localhost:6379",
    db=0,
    decode_responses=True,
    max_connections=10
)

Memory Backend (Development)

from cloudbase_agent.server.storage import MemoryStorage

storage = MemoryStorage()  # In-memory, no persistence

PostgreSQL Backend

from cloudbase_agent.server.storage import PostgresStorage

storage = PostgresStorage(
    connection_string="postgresql://user:pass@localhost/cloudbase_agent_db",
    table_name="agent_storage"
)

Conversation Storage

Store Conversation State

from cloudbase_agent.server.storage import ConversationStorage

conv_storage = ConversationStorage(storage)

# Save conversation
await conv_storage.save_conversation(
    conversation_id="conv_123",
    messages=[
        {"role": "user", "content": "Hello"},
        {"role": "assistant", "content": "Hi there!"}
    ],
    metadata={"user_id": "user_456"}
)

# Load conversation
conversation = await conv_storage.load_conversation("conv_123")

Conversation History

# Get all conversations for a user
conversations = await conv_storage.list_conversations(
    user_id="user_456",
    limit=10,
    offset=0
)

# Search conversations
results = await conv_storage.search_conversations(
    query="agent features",
    user_id="user_456"
)

Checkpointing

LangGraph Checkpointer

from cloudbase_agent.langgraph import create_checkpointer

# Create checkpointer from storage
checkpointer = create_checkpointer(storage)

# Use with LangGraph
graph = create_react_agent(
    model=model,
    tools=tools,
    checkpointer=checkpointer
)

Manual Checkpointing

# Save checkpoint
await checkpointer.put(
    config={"configurable": {"thread_id": "thread_123"}},
    checkpoint={
        "values": graph_state,
        "next": ["tool_node"],
        "metadata": {"step": 5}
    }
)

# Load checkpoint
checkpoint = await checkpointer.get(
    config={"configurable": {"thread_id": "thread_123"}}
)

TTL and Expiration

# Set data with TTL
await storage.set("temp_key", {"data": "value"}, ttl=3600)  # 1 hour

# Update TTL
await storage.expire("temp_key", ttl=7200)  # 2 hours

# Get TTL
remaining = await storage.ttl("temp_key")

Batch Operations

# Batch set
await storage.mset({
    "key1": {"value": 1},
    "key2": {"value": 2},
    "key3": {"value": 3}
})

# Batch get
values = await storage.mget(["key1", "key2", "key3"])

# Batch delete
await storage.delete_many(["key1", "key2", "key3"])

Namespacing

# Create namespaced storage
user_storage = storage.namespace("user:123")

# Operations are automatically prefixed
await user_storage.set("preferences", {"theme": "dark"})
# Actually stores at "user:123:preferences"

# Nested namespaces
session_storage = user_storage.namespace("session:456")
# Keys stored at "user:123:session:456:*"

Transactions

Redis Transactions

async with storage.transaction() as txn:
    await txn.set("counter", 0)
    value = await txn.get("counter")
    await txn.set("counter", value + 1)
    # Auto-commits on successful exit

PostgreSQL Transactions

async with storage.transaction() as txn:
    await txn.execute("UPDATE users SET balance = balance - 100 WHERE id = 1")
    await txn.execute("UPDATE users SET balance = balance + 100 WHERE id = 2")
    # Auto-commits or rolls back

Serialization

Custom Serializers

from cloudbase_agent.server.storage import Storage, JSONSerializer, PickleSerializer

# JSON serializer (default)
storage = Storage(backend="redis", serializer=JSONSerializer())

# Pickle serializer (Python objects)
storage = Storage(backend="redis", serializer=PickleSerializer())

# Custom serializer
class CustomSerializer:
    def serialize(self, obj):
        return msgpack.packb(obj)
    
    def deserialize(self, data):
        return msgpack.unpackb(data)

storage = Storage(backend="redis", serializer=CustomSerializer())

Monitoring

Storage Metrics

# Get storage stats
stats = await storage.stats()
# Returns: {
#     "keys_count": 1234,
#     "memory_usage": 5242880,  # bytes
#     "hit_rate": 0.95
# }

# Health check
is_healthy = await storage.health_check()

Migration

Data Migration

from cloudbase_agent.server.storage import migrate_storage

# Migrate from Redis to PostgreSQL
await migrate_storage(
    source=redis_storage,
    destination=postgres_storage,
    batch_size=100,
    transform_fn=lambda k, v: (k, transform(v))
)

Best Practices

1. Use Namespacing: Organize keys with namespaces to avoid collisions 2. Set Appropriate TTLs: Use TTL for temporary data to prevent memory bloat 3. Batch Operations: Use batch operations for multiple keys to reduce latency 4. Connection Pooling: Configure connection pools for production workloads 5. Error Handling: Always handle storage errors gracefully 6. Monitoring: Track storage metrics and set up alerts

Common Patterns

Session Management

class SessionManager:
    def __init__(self, storage: Storage):
        self.storage = storage.namespace("sessions")
    
    async def create_session(self, user_id: str) -> str:
        session_id = generate_session_id()
        await self.storage.set(
            session_id,
            {"user_id": user_id, "created_at": datetime.now()},
            ttl=3600  # 1 hour
        )
        return session_id
    
    async def get_session(self, session_id: str) -> dict | None:
        return await self.storage.get(session_id)

Rate Limiting

async def rate_limit(user_id: str, limit: int = 100, window: int = 3600):
    key = f"rate_limit:{user_id}"
    count = await storage.get(key) or 0
    
    if count >= limit:
        raise RateLimitError("Too many requests")
    
    await storage.set(key, count + 1, ttl=window)

See Also

• Server Reference - Server configuration
• LangGraph Reference - Checkpointing integration
• Recipes - Storage use cases

CloudBase Agent Tools System Reference

Overview

CloudBase Agent provides a flexible tool system for integrating external capabilities into agents.

Core Concepts

Tool Definition

from cloudbase_agent.server import tool

@tool
def search_database(query: str, limit: int = 10) -> list[dict]:
    """Search the database for matching records.
    
    Args:
        query: Search query string
        limit: Maximum number of results to return
        
    Returns:
        List of matching records
    """
    # Implementation
    return results

Tool Registry

from cloudbase_agent.server import ToolRegistry

# Create registry
registry = ToolRegistry()

# Register tools
registry.register(search_database)
registry.register(update_record)

# Get all tools
tools = registry.get_tools()

Built-in Tool Types

HTTP Tools

from cloudbase_agent.server.tools import HttpTool

http_tool = HttpTool(
    name="fetch_data",
    method="GET",
    url="https://api.example.com/data",
    headers={"Authorization": "Bearer TOKEN"}
)

Database Tools

from cloudbase_agent.server.tools import DatabaseTool

db_tool = DatabaseTool(
    name="query_users",
    connection_string="postgresql://localhost/mydb",
    query="SELECT * FROM users WHERE active = true"
)

Tool Execution

Synchronous Execution

result = await tool.execute({"query": "search term", "limit": 5})

Async Tool Support

@tool
async def async_search(query: str) -> dict:
    """Async tool example."""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/search?q={query}")
        return response.json()

Tool Validation

Input Validation

from pydantic import BaseModel, Field

class SearchParams(BaseModel):
    query: str = Field(..., min_length=1, max_length=100)
    limit: int = Field(default=10, ge=1, le=100)

@tool
def validated_search(params: SearchParams) -> list[dict]:
    """Tool with Pydantic validation."""
    return search(params.query, params.limit)

Error Handling

from cloudbase_agent.server.tools import ToolError

@tool
def safe_operation(data: dict) -> dict:
    """Tool with error handling."""
    try:
        result = risky_operation(data)
        return {"success": True, "data": result}
    except ValueError as e:
        raise ToolError(f"Invalid input: {e}")
    except Exception as e:
        raise ToolError(f"Operation failed: {e}")

Tool Metadata

@tool(
    name="custom_name",
    description="Detailed description",
    tags=["search", "database"],
    version="1.0.0"
)
def advanced_tool(param: str) -> dict:
    """Advanced tool with metadata."""
    return {}

Tool Composition

Chaining Tools

@tool
def fetch_and_process(query: str) -> dict:
    """Chain multiple operations."""
    # Fetch data
    raw_data = fetch_data(query)
    
    # Process data
    processed = process_data(raw_data)
    
    # Store results
    store_results(processed)
    
    return {"status": "complete", "count": len(processed)}

Integration with Agents

LangGraph Integration

from cloudbase_agent.langgraph import create_react_agent

agent = create_react_agent(
    model=model,
    tools=[search_database, update_record, async_search]
)

Custom Tool Nodes

from langgraph.prebuilt import ToolNode

tool_node = ToolNode([search_database, update_record])

# Add to graph
graph.add_node("tools", tool_node)

Best Practices

1. Clear Descriptions: Write detailed docstrings for AI to understand tool purpose 2. Type Hints: Always use type hints for parameters and return values 3. Error Handling: Catch and wrap errors with meaningful messages 4. Validation: Use Pydantic models for complex input validation 5. Async Support: Use async tools for I/O-bound operations 6. Idempotency: Make tools safe to retry when possible

Common Patterns

Retry Logic

from tenacity import retry, stop_after_attempt, wait_exponential

@tool
@retry(stop=stop_after_attempt(3), wait=wait_exponential())
async def resilient_api_call(endpoint: str) -> dict:
    """API call with automatic retries."""
    async with httpx.AsyncClient() as client:
        response = await client.get(endpoint)
        response.raise_for_status()
        return response.json()

Caching Results

from functools import lru_cache

@tool
@lru_cache(maxsize=100)
def cached_lookup(key: str) -> dict:
    """Cached database lookup."""
    return db.query(key)

See Also

• Server Reference - Server configuration
• LangGraph Reference - Agent integration
• Recipes - Common use cases

Server Quickstart Guide

This guide shows you how to create and deploy CloudBase Agent Python agents as HTTP services using FastAPI.

---

Three Deployment Methods

CloudBase Agent Python SDK provides three flexible deployment methods, each suited for different use cases:

Method 1: Core Adapters (Maximum Flexibility)

Use create_send_message_adapter() and create_openai_adapter() directly for complete control over routes.

from fastapi import FastAPI
from cloudbase_agent.server import (
    create_send_message_adapter,
    create_openai_adapter,
    RunAgentInput,
    OpenAIChatCompletionRequest
)
from cloudbase_agent.server.errors import install_exception_handlers

app = FastAPI()

# Required: Install exception handlers for AG-UI protocol compatibility
install_exception_handlers(app)

# Define routes manually
@app.post("/my-agent/send-message")
async def send_message(request: RunAgentInput):
    return await create_send_message_adapter(create_agent, request)

@app.post("/my-agent/chat/completions")
async def openai_endpoint(request: OpenAIChatCompletionRequest):
    return await create_openai_adapter(create_agent, request)

Advantages:

• Full control over route paths
• Easy to add custom middleware per route
• Clear separation of concerns
• Ideal for complex multi-agent systems

Use when:

• You need custom route structures
• You want fine-grained control
• You're building complex applications

Method 2: AgentServiceApp with Custom Routes (Recommended)

Use AgentServiceApp.build() for automatic features with flexibility:

from fastapi import FastAPI
from cloudbase_agent.server import AgentServiceApp, HealthzConfig

# Create main app
main_app = FastAPI(title="My Service")

# Build agent app with automatic features
agent_app = AgentServiceApp()
agent_fastapi = agent_app.build(
    create_agent,
    base_path="",
    enable_cors=False,  # Handle in main app
    enable_openai_endpoint=True,
    enable_healthz=True,
    healthz_config=HealthzConfig(
        service_name="my-agent",
        version="1.0.0"
    )
)

# Mount to main app
main_app.mount("/my-agent", agent_fastapi)

Advantages:

• Automatic health check endpoints
• Built-in OpenAI compatibility
• Less boilerplate code
• Modular agent deployment

Use when:

• You want automatic health checks
• You're deploying multiple agents
• You need both CloudBase Agent native and OpenAI endpoints

Method 3: One-Line Deployment (Simplest)

For single-agent deployments, use the one-line approach:

from cloudbase_agent.server import AgentServiceApp, HealthzConfig

AgentServiceApp().run(
    create_agent,
    port=9000,
    enable_openai_endpoint=True,
    healthz_config=HealthzConfig(
        service_name="my-agent",
        version="1.0.0"
    )
)

Advantages:

• Extremely simple (one line!)
• Perfect for prototyping
• Automatic CORS and health checks
• No boilerplate needed

Use when:

• You have a single agent
• You want the fastest way to start
• You don't need custom routes

---

Agent Creator Pattern

All three methods use an "agent creator" function that returns an AgentCreatorResult:

from cloudbase_agent.langgraph import LangGraphAgent
from cloudbase_agent.server import AgentCreatorResult

def create_agent() -> AgentCreatorResult:
    """Create agent with optional cleanup."""
    
    agent = LangGraphAgent(
        name="my-agent",
        description="A helpful assistant",
        graph=build_workflow(),
        use_callbacks=True
    )
    
    # Optional: Add callbacks
    agent.add_callback(ConsoleLogger())
    
    # Optional: Define cleanup function
    def cleanup():
        # Close connections, release resources, etc.
        print("Cleanup completed")
    
    return {"agent": agent, "cleanup": cleanup}

Why use creator functions?

• Agent instance is created fresh per request
• Ensures proper isolation
• Automatic cleanup after request completes
• Supports resource management (DB connections, file handles, etc.)

---

Complete Example: Multi-Agent Server

Here's a production-ready example with multiple agents:

#!/usr/bin/env python3
import logging
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from cloudbase_agent.langgraph import LangGraphAgent
from cloudbase_agent.server import (
    AgentCreatorResult,
    AgentServiceApp,
    HealthzConfig,
    OpenAIChatCompletionRequest,
    RunAgentInput,
    create_openai_adapter,
    create_send_message_adapter,
)
from cloudbase_agent.server.errors import install_exception_handlers

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)

# Import your agent workflows
from agents.chat.agent import build_chat_workflow
from agents.assistant.agent import build_assistant_workflow

# Initialize workflows
chat_workflow = build_chat_workflow()
assistant_workflow = build_assistant_workflow()


# Agent creator for chat bot
def create_chat_agent() -> AgentCreatorResult:
    agent = LangGraphAgent(
        name="chatbot",
        description="A conversational assistant",
        graph=chat_workflow,
        use_callbacks=True
    )
    return {"agent": agent}


# Agent creator for assistant
def create_assistant_agent() -> AgentCreatorResult:
    agent = LangGraphAgent(
        name="assistant",
        description="A helpful AI assistant",
        graph=assistant_workflow,
        use_callbacks=True
    )
    
    def cleanup():
        print(f"Cleanup for {agent.name}")
    
    return {"agent": agent, "cleanup": cleanup}


def main():
    # Method 1: Using core adapters
    app = FastAPI(
        title="CloudBase Agent Multi-Agent Server",
        version="1.0.0"
    )
    
    # Install exception handlers (required for Method 1)
    install_exception_handlers(app)
    
    # Add CORS
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    
    # Chat bot endpoints
    @app.post("/chatbot/send-message")
    async def chatbot_send_message(request: RunAgentInput):
        return await create_send_message_adapter(create_chat_agent, request)
    
    @app.post("/chatbot/chat/completions")
    async def chatbot_openai(request: OpenAIChatCompletionRequest):
        return await create_openai_adapter(create_chat_agent, request)
    
    # Assistant endpoints
    @app.post("/assistant/send-message")
    async def assistant_send_message(request: RunAgentInput):
        return await create_send_message_adapter(create_assistant_agent, request)
    
    @app.post("/assistant/chat/completions")
    async def assistant_openai(request: OpenAIChatCompletionRequest):
        return await create_openai_adapter(create_assistant_agent, request)
    
    # Health check
    @app.get("/healthz")
    def healthz():
        from datetime import datetime
        import platform
        
        return {
            "status": "healthy",
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "version": "1.0.0",
            "python_version": platform.python_version(),
            "agents": [
                {"name": "chatbot", "endpoints": ["/chatbot/send-message", "/chatbot/chat/completions"]},
                {"name": "assistant", "endpoints": ["/assistant/send-message", "/assistant/chat/completions"]},
            ]
        }
    
    uvicorn.run(app, host="0.0.0.0", port=9000)


if __name__ == "__main__":
    main()

---

Testing Your Server

1. CloudBase Agent Native Endpoint

curl -X POST http://localhost:9000/chatbot/send-message \
  -H "Content-Type: application/json" \
  -d '{
    "conversationId": "conv_123",
    "messages": [
      {"role": "user", "content": "Hello!"}
    ]
  }'

2. OpenAI-Compatible Endpoint

curl -X POST http://localhost:9000/chatbot/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "chatbot",
    "messages": [
      {"role": "user", "content": "Hello!"}
    ],
    "stream": true
  }'

3. Using OpenAI Python Client

from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:9000",
    api_key="dummy"  # Not required for local
)

response = client.chat.completions.create(
    model="chatbot",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True
)

for chunk in response:
    print(chunk.choices[0].delta.content, end="")

4. Health Check

curl http://localhost:9000/healthz

---

Callbacks for Monitoring

Add callbacks to your agent for real-time monitoring:

class ConsoleLogger:
    """Log agent events to console."""
    
    async def on_text_message_content(self, event, buffer):
        print(f"[AI] {buffer}", end="", flush=True)
    
    async def on_tool_call_args(self, event, buffer, partial_args):
        tool_name = getattr(event, "tool_name", "unknown")
        if partial_args:
            print(f"\n[TOOL] {tool_name}: {partial_args}")
    
    async def on_run_started(self, event):
        print(f"\n{'=' * 60}")
        print(f"Run Started: {event.run_id}")
        print(f"{'=' * 60}")
    
    async def on_run_finished(self, event):
        print(f"\n{'=' * 60}")
        print(f"Run Finished: {event.run_id}")
        print(f"{'=' * 60}\n")
    
    async def on_run_error(self, event):
        print(f"\nERROR: {getattr(event, 'message', 'Unknown')}\n")


def create_agent() -> AgentCreatorResult:
    agent = LangGraphAgent(
        name="my-agent",
        graph=build_workflow(),
        use_callbacks=True  # Enable callbacks
    )
    
    # Add console logger
    agent.add_callback(ConsoleLogger())
    
    return {"agent": agent}

---

Production Considerations

1. Use Environment Variables

# .env
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o-mini
OPENAI_TEMPERATURE=0.7
OPENAI_BASE_URL=https://api.openai.com/v1  # Optional

# Load in code
from dotenv import load_dotenv
load_dotenv()

2. Enable Observability

export AUTO_TRACES_STDOUT=true
python server.py

Or programmatically:

from cloudbase_agent.observability import ConsoleTraceConfig, enable_tracing

enable_tracing(ConsoleTraceConfig())

3. Add Authentication Middleware

See authentication.md for details on implementing JWT-based authentication.

4. Configure CORS Properly

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-frontend.com"],  # Specific origins in production
    allow_credentials=True,
    allow_methods=["POST", "GET"],
    allow_headers=["*"],
)

5. Use Production Server

# Install gunicorn
pip install gunicorn uvicorn[standard]

# Run with multiple workers
gunicorn server:app.app \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:9000

---

Next Steps

• LangGraph Integration: See adapter-langgraph.md for detailed LangGraph usage
• Coze Integration: See adapter-coze.md for Coze platform integration
• Authentication: See authentication.md for auth patterns
• Custom Adapters: See adapter-development.md for creating custom framework adapters

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.

@cloudbase/agent-adapter-langchain

Adapter that wraps LangChain's createAgent() as an AG-UI compatible agent. Provides LangchainAgent wrapper class and clientTools() middleware for client tools support.

Basic Usage

import { createAgent as createLangchainAgent } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { LangchainAgent, clientTools } from "@cloudbase/agent-adapter-langchain";

const model = new ChatOpenAI({ model: "gpt-4o" });
const checkpointer = new MemorySaver();

const lcAgent = createLangchainAgent({
  model,
  checkpointer,
  middleware: [clientTools()],
});

const agent = new LangchainAgent({ agent: lcAgent });

Checkpointer (Required)

LangchainAgent requires the agent to be created with a checkpointer.

MemorySaver (Development)

import { MemorySaver } from "@langchain/langgraph";

const lcAgent = createLangchainAgent({
  model,
  checkpointer: new MemorySaver(),
  middleware: [clientTools()],
});

CloudBaseSaver (Production)

Persistent storage using Tencent CloudBase document database. On CloudBase cloud function/cloudrun, requests are authenticated - extract user ID from the JWT in Authorization header.

import { run } from "@cloudbase/agent-server";
import { LangchainAgent, clientTools } from "@cloudbase/agent-adapter-langchain";
import { CloudBaseSaver } from "@cloudbase/agent-adapter-langgraph";
import { createAgent as createLangchainAgent } from "langchain";
import tcb from "@cloudbase/node-sdk";

const app = tcb.init({ env: process.env.CLOUDBASE_ENV_ID });

run({
  createAgent: ({ request }) => {
    // Extract user ID from JWT (sub field)
    const token = request.headers.get("Authorization")?.slice(7);
    const payload = JSON.parse(atob(token.split(".")[1]));
    const userId = payload.sub;

    const checkpointer = new CloudBaseSaver({
      db: app.database(),
      userId,  // Multi-tenant isolation
    });

    const lcAgent = createLangchainAgent({
      model,
      checkpointer,
      middleware: [clientTools()],
    });

    return { agent: new LangchainAgent({ agent: lcAgent }) };
  },
  port: 9000,
});

With @cloudbase/agent-server

import { run } from "@cloudbase/agent-server";

run({
  createAgent: () => ({ agent }),
  port: 9000,
});

clientTools() Middleware

Enables client-defined tools in your LangChain agent:

• Injects client tools - Adds client tools to the LLM's available tool list
• Routes to END - When a client tool is called, skips ToolNode and routes to END so client can execute

UI Design Activation Checklist

Use this checklist before generating any page, component, or visual interface.

Required checks

1. Output the design specification first. 2. Choose a concrete aesthetic direction rather than generic adjectives. 3. Define a color palette and typography before writing markup or styles. 4. Confirm the target platform: Web or mini program. 5. Read the platform implementation skill after the design spec is fixed.

Common failure patterns

• Starting with JSX, WXML, or CSS before design intent is stated.
• Falling back to generic AI visual patterns.
• Missing platform-specific layout or asset constraints.

Done criteria

• The design spec is visible in the response.
• Aesthetic direction, palette, and typography are explicit.
• The next implementation skill is known before UI code starts.