Server API

The Server API provides FastAPI components for deploying AG-Kit agents as web services. It includes the main application wrapper, adapter functions, and data models for HTTP endpoints.

Installation

pip install ag_kit_py

FastAPI and Uvicorn are installed automatically as dependencies.

Core Components

AGKitAPIApp

Main FastAPI application wrapper for quick server setup.

from ag_kit_py.server import AGKitAPIApp

class AGKitAPIApp:
    """FastAPI Application Wrapper for AG-Kit Agents."""
    
    def __init__(self, auto_detect_env: bool = True):
        """Initialize the AG-Kit API application.
        
        Args:
            auto_detect_env: Automatically detect runtime environment
                           (e.g., Tencent Cloud SCF) and configure accordingly
        """

Methods

run()

Build and run the FastAPI application (one-line startup).

def run(
    self,
    create_agent: AgentCreator,
    base_path: str = "",
    host: str = "0.0.0.0",
    port: int = 9000,
    enable_openai_endpoint: bool = False,
    enable_healthz: bool = True,
    healthz_config: Optional[HealthzConfig] = None,
    **uvicorn_kwargs: Any
) -> None:
    """Build and run the FastAPI application.
    
    Args:
        create_agent: Function that creates and returns agent with optional cleanup
        base_path: Base path for agent routes
        host: Server host address
        port: Server port number
        enable_openai_endpoint: Whether to enable OpenAI-compatible endpoint
        enable_healthz: Whether to enable health check endpoint
        healthz_config: Optional health check configuration
        **uvicorn_kwargs: Additional keyword arguments for uvicorn.run
    """

Example:

from ag_kit_py.server import AGKitAPIApp

def create_agent():
    return {"agent": my_agent}

# Simplest usage
AGKitAPIApp().run(create_agent, port=8000)

# With configuration
AGKitAPIApp().run(
    create_agent,
    port=8000,
    reload=True,
    log_level="debug"
)

build()

Build and return a configured FastAPI application for advanced customization.

def build(
    self,
    create_agent: AgentCreator,
    base_path: str = "",
    enable_cors: bool = True,
    enable_openai_endpoint: bool = False,
    enable_healthz: bool = True,
    healthz_config: Optional[HealthzConfig] = None,
    **fastapi_kwargs: Any
) -> FastAPI:
    """Build and return a configured FastAPI application.
    
    Args:
        create_agent: Function that creates and returns agent with optional cleanup
        base_path: Base path for agent routes
        enable_cors: Whether to enable CORS middleware
        enable_openai_endpoint: Whether to enable OpenAI-compatible endpoint
        enable_healthz: Whether to enable health check endpoint
        healthz_config: Optional health check configuration
        **fastapi_kwargs: Additional keyword arguments for FastAPI constructor
        
    Returns:
        FastAPI: Configured FastAPI application
    """

Example:

from ag_kit_py.server import AGKitAPIApp

app_wrapper = AGKitAPIApp()
fastapi_app = app_wrapper.build(
    create_agent,
    base_path="/api",
    title="My Agent API",
    version="1.0.0"
)

# Add custom routes
@fastapi_app.get("/custom")
def custom_route():
    return {"custom": "data"}

# Run manually
import uvicorn
uvicorn.run(fastapi_app, host="0.0.0.0", port=8000)

set_cors_config()

Set CORS configuration for the application.

def set_cors_config(
    self,
    allow_origins: Optional[list[str]] = None,
    allow_credentials: bool = True,
    allow_methods: Optional[list[str]] = None,
    allow_headers: Optional[list[str]] = None,
    **kwargs: Any
) -> "AGKitAPIApp":
    """Set CORS configuration for the application.
    
    Args:
        allow_origins: List of allowed origins (default: ["*"])
        allow_credentials: Whether to allow credentials (default: True)
        allow_methods: List of allowed HTTP methods (default: ["*"])
        allow_headers: List of allowed headers (default: ["*"])
        **kwargs: Additional CORS configuration options
        
    Returns:
        AGKitAPIApp: Self for method chaining
    """

Example:

from ag_kit_py.server import AGKitAPIApp

AGKitAPIApp() \
    .set_cors_config(
        allow_origins=["https://example.com"],
        allow_credentials=True
    ) \
    .run(create_agent, port=8000)

Adapter Functions

create_send_message_adapter()

Create a FastAPI adapter for AG-Kit native send_message endpoint.

async def create_send_message_adapter(
    create_agent: AgentCreator,
    request: SendMessageInput
) -> StreamingResponse:
    """Create a FastAPI adapter for send_message requests.
    
    Args:
        create_agent: Function that creates and returns agent with optional cleanup
        request: The validated request input containing messages and tools
        
    Returns:
        StreamingResponse: Streaming response with SSE-formatted events
        
    Raises:
        ValidationError: When request validation fails
        Exception: When agent processing fails
    """

Example:

from fastapi import FastAPI
from ag_kit_py.server import create_send_message_adapter, SendMessageInput

app = FastAPI()

def create_agent():
    return {"agent": my_agent}

@app.post("/send-message")
async def send_message(request: SendMessageInput):
    return await create_send_message_adapter(create_agent, request)

create_openai_adapter()

Create a FastAPI adapter for OpenAI-compatible chat/completions endpoint.

async def create_openai_adapter(
    create_agent: AgentCreator,
    request: OpenAIChatCompletionRequest
) -> StreamingResponse:
    """Create a FastAPI adapter for OpenAI-compatible requests.
    
    Args:
        create_agent: Function that creates and returns agent with optional cleanup
        request: OpenAI-formatted chat completion request
        
    Returns:
        StreamingResponse: Streaming response with SSE-formatted events
    """

Example:

from fastapi import FastAPI
from ag_kit_py.server import create_openai_adapter
from ag_kit_py.server.openai.models import OpenAIChatCompletionRequest

app = FastAPI()

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

Data Models

AgentCreator

Type alias for agent creator functions.

from typing import Callable, Union, Awaitable

AgentCreator = Callable[
    [], 
    Union[AgentCreatorResult, Awaitable[AgentCreatorResult]]
]

Agent creator functions can be synchronous or asynchronous:

# Synchronous
def create_agent() -> AgentCreatorResult:
    return {"agent": MyAgent()}

# Asynchronous
async def create_agent() -> AgentCreatorResult:
    agent = await initialize_agent()
    return {"agent": agent}

AgentCreatorResult

Result type for agent creator functions.

from typing import TypedDict, Optional, Callable, Union, Awaitable, Any

class AgentCreatorResult(TypedDict, total=False):
    """Result type for agent creator functions.
    
    Attributes:
        agent: The agent instance to handle requests
        cleanup: Optional cleanup function called after stream completes
    """
    agent: Any
    cleanup: Optional[Callable[[], Union[None, Awaitable[None]]]]

Example:

def create_agent() -> AgentCreatorResult:
    db = connect_database()
    agent = MyAgent(db)
    
    def cleanup():
        db.close()
        print("Database closed")
    
    return {"agent": agent, "cleanup": cleanup}

SendMessageInput

Request model for send_message endpoint.

from typing import List, Optional
from pydantic import BaseModel

class SendMessageInput(BaseModel):
    """Send message API input model.
    
    Attributes:
        messages: List of conversation messages in chronological order
        tools: Optional list of available tools for the agent to use
        resume: Optional resume message for interrupted conversations
        conversationId: Unique identifier for the conversation thread
    """
    messages: Optional[List[ClientMessage]] = []
    tools: Optional[List[Tool]] = []
    resume: Optional[ResumeMessage] = None
    conversationId: str

Example:

from ag_kit_py.server import SendMessageInput

request = SendMessageInput(
    messages=[
        {"role": "system", "content": "You are helpful."},
        {"role": "user", "content": "Hello!"}
    ],
    tools=[
        {
            "name": "calculator",
            "description": "Perform calculations",
            "parameters": {"type": "object", "properties": {...}}
        }
    ],
    conversationId="conv_abc123"
)

OpenAIChatCompletionRequest

Request model for OpenAI-compatible endpoint.

from typing import List, Optional
from pydantic import BaseModel

class OpenAIChatCompletionRequest(BaseModel):
    """OpenAI-compatible chat completion request model.
    
    Attributes:
        model: Model identifier (ignored, uses configured agent)
        messages: List of conversation messages
        stream: Whether to stream the response
        tools: Optional list of available tools
        temperature: Sampling temperature (ignored)
        max_tokens: Maximum tokens to generate (ignored)
    """
    model: str
    messages: List[OpenAIMessage]
    stream: bool = True
    tools: Optional[List[Tool]] = None
    temperature: Optional[float] = None
    max_tokens: Optional[int] = None

Example:

from ag_kit_py.server.openai.models import OpenAIChatCompletionRequest

request = OpenAIChatCompletionRequest(
    model="gpt-4o-mini",
    messages=[
        {"role": "user", "content": "Hello!"}
    ],
    stream=True
)

HealthzConfig

Configuration for health check endpoint.

from typing import Optional, Callable, Any
from pydantic import BaseModel, Field

class HealthzConfig(BaseModel):
    """Configuration for health check endpoint.
    
    Attributes:
        service_name: Name of the service
        version: Service version
        include_agent_info: Whether to include agent information
        custom_checks: Optional function that performs additional health checks
    """
    service_name: str = Field(default="ag-kit-python-server")
    version: str = Field(default="1.0.0")
    include_agent_info: bool = Field(default=False)
    custom_checks: Optional[Callable[[], dict[str, Any]]] = Field(default=None)

Example:

from ag_kit_py.server import HealthzConfig

def check_database():
    if not db.is_connected():
        raise Exception("Database not connected")
    return {"database": "connected"}

config = HealthzConfig(
    service_name="my-agent-service",
    version="2.0.0",
    custom_checks=check_database
)

HealthzResponse

Health check response model.

from typing import Optional, Any
from pydantic import BaseModel, Field

class HealthzResponse(BaseModel):
    """Health check response model.
    
    Attributes:
        status: Health status ("healthy" or "unhealthy")
        timestamp: ISO 8601 timestamp of the health check
        service: Service name
        version: Service version
        python_version: Python runtime version
        base_path: Base path of the service
        agent_info: Optional agent information
        custom: Optional custom check results
        error: Optional error message if unhealthy
    """
    status: str
    timestamp: str
    service: str
    version: str
    python_version: str
    base_path: str
    agent_info: Optional[dict[str, Any]] = None
    custom: Optional[dict[str, Any]] = None
    error: Optional[str] = None

Usage Examples

Quick Start

from ag_kit_py.server import AGKitAPIApp
from ag_kit_py.agents import LangGraphAgent

def create_agent():
    return {"agent": my_agent}

# One-line server startup
AGKitAPIApp().run(create_agent, port=8000)

With Resource Cleanup

from ag_kit_py.server import AGKitAPIApp, AgentCreatorResult

def create_agent() -> AgentCreatorResult:
    db = connect_database()
    agent = MyAgent(db)
    
    def cleanup():
        db.close()
        print("Resources cleaned up")
    
    return {"agent": agent, "cleanup": cleanup}

AGKitAPIApp().run(create_agent, port=8000)

With CORS Configuration

from ag_kit_py.server import AGKitAPIApp

AGKitAPIApp() \
    .set_cors_config(
        allow_origins=["https://example.com", "https://app.example.com"],
        allow_credentials=True
    ) \
    .run(create_agent, port=8000)

Advanced Customization

from ag_kit_py.server import AGKitAPIApp
from fastapi import FastAPI

app_wrapper = AGKitAPIApp()
fastapi_app = app_wrapper.build(
    create_agent,
    base_path="/api",
    title="My Agent API",
    description="AI Agent Service",
    version="1.0.0"
)

# Add custom middleware
@fastapi_app.middleware("http")
async def add_custom_header(request, call_next):
    response = await call_next(request)
    response.headers["X-Custom-Header"] = "value"
    return response

# Add custom routes
@fastapi_app.get("/api/status")
def status():
    return {"status": "running"}

# Run the application
import uvicorn
uvicorn.run(fastapi_app, host="0.0.0.0", port=8000)

With Health Check Configuration

from ag_kit_py.server import AGKitAPIApp, HealthzConfig

def check_redis():
    if not redis.ping():
        raise Exception("Redis not available")
    return {"redis": "connected"}

config = HealthzConfig(
    service_name="my-agent-service",
    version="2.0.0",
    custom_checks=check_redis
)

AGKitAPIApp().run(
    create_agent,
    port=8000,
    healthz_config=config
)

OpenAI-Compatible Endpoint

from ag_kit_py.server import AGKitAPIApp

AGKitAPIApp().run(
    create_agent,
    port=8000,
    enable_openai_endpoint=True  # Enables /chat/completions endpoint
)

Development Mode

from ag_kit_py.server import AGKitAPIApp

AGKitAPIApp().run(
    create_agent,
    port=8000,
    reload=True,           # Auto-reload on code changes
    log_level="debug",     # Detailed logging
    access_log=True        # Log all requests
)

Production Mode

from ag_kit_py.server import AGKitAPIApp

AGKitAPIApp().run(
    create_agent,
    port=8000,
    workers=4,             # Multiple worker processes
    access_log=False,      # Disable access logging
    log_level="warning"    # Only warnings and errors
)

Custom Adapter Usage

from fastapi import FastAPI
from ag_kit_py.server import (
    create_send_message_adapter,
    create_openai_adapter,
    SendMessageInput,
)
from ag_kit_py.server.openai.models import OpenAIChatCompletionRequest

app = FastAPI()

def create_agent():
    return {"agent": my_agent}

@app.post("/custom/send-message")
async def send_message(request: SendMessageInput):
    return await create_send_message_adapter(create_agent, request)

@app.post("/custom/chat/completions")
async def chat_completions(request: OpenAIChatCompletionRequest):
    return await create_openai_adapter(create_agent, request)

# Run with uvicorn
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

Environment Detection

AGKitAPIApp automatically detects runtime environments and configures accordingly:

Tencent Cloud SCF

import os

# Automatically detected
os.environ["TENCENTCLOUD_RUNENV"] = "SCF"

# Base path automatically set to "/v1/aibot/bots"
AGKitAPIApp().run(create_agent, port=8000)

Custom Environment

app = AGKitAPIApp(auto_detect_env=False)  # Disable auto-detection

app.run(
    create_agent,
    base_path="/custom/path",
    port=8000
)

Endpoints

When using AGKitAPIApp, the following endpoints are automatically created:

POST /send-message

AG-Kit native endpoint for processing messages. Request:

{
  "messages": [
    {"role": "user", "content": "Hello"}
  ],
  "tools": [],
  "conversationId": "conv_123"
}

Response: Server-Sent Events stream

GET /healthz

Health check endpoint. Response:

{
  "status": "healthy",
  "timestamp": "2024-01-01T00:00:00.000Z",
  "service": "ag-kit-python-server",
  "version": "1.0.0",
  "python_version": "3.11.0",
  "base_path": "/"
}

POST /chat/completions (Optional)

OpenAI-compatible endpoint (enabled with enable_openai_endpoint=True). Request:

{
  "model": "gpt-4o-mini",
  "messages": [
    {"role": "user", "content": "Hello"}
  ],
  "stream": true
}

Response: Server-Sent Events stream

Best Practices

1. Use Resource Cleanup

def create_agent() -> AgentCreatorResult:
    # Acquire resources
    db = connect_database()
    cache = connect_redis()
    agent = MyAgent(db, cache)
    
    # Define cleanup
    def cleanup():
        db.close()
        cache.close()
    
    return {"agent": agent, "cleanup": cleanup}

2. Configure CORS Properly

# Development: Allow all origins
AGKitAPIApp() \
    .set_cors_config(allow_origins=["*"]) \
    .run(create_agent, port=8000)

# Production: Restrict origins
AGKitAPIApp() \
    .set_cors_config(
        allow_origins=["https://yourdomain.com"],
        allow_credentials=True
    ) \
    .run(create_agent, port=8000)

3. Use Health Checks

def check_dependencies():
    """Check all service dependencies."""
    checks = {}
    
    # Check database
    try:
        db.ping()
        checks["database"] = "connected"
    except:
        raise Exception("Database unavailable")
    
    # Check cache
    try:
        cache.ping()
        checks["cache"] = "connected"
    except:
        raise Exception("Cache unavailable")
    
    return checks

config = HealthzConfig(custom_checks=check_dependencies)
AGKitAPIApp().run(create_agent, healthz_config=config, port=8000)

4. Handle Errors Gracefully

def create_agent() -> AgentCreatorResult:
    try:
        agent = initialize_agent()
        return {"agent": agent}
    except Exception as e:
        logger.error(f"Failed to create agent: {e}")
        raise

5. Use Type Hints

from ag_kit_py.server import AgentCreatorResult

def create_agent() -> AgentCreatorResult:
    """Create agent with proper type hints."""
    return {"agent": my_agent}

Server Overview - Server system overview
Server Events - Event system details
Agents Overview - Agent system
LangGraph Agent - LangGraph implementation

TypeScript SDK

Python SDK

Headless UI Components

CLI Tools

​Server API

​Installation

​Core Components

​AGKitAPIApp

​Methods

run()

build()

set_cors_config()

​Adapter Functions

​create_send_message_adapter()

​create_openai_adapter()

​Data Models

​AgentCreator

​AgentCreatorResult

​SendMessageInput

​OpenAIChatCompletionRequest

​HealthzConfig

​HealthzResponse

​Usage Examples

​Quick Start

​With Resource Cleanup

​With CORS Configuration

​Advanced Customization

​With Health Check Configuration

​OpenAI-Compatible Endpoint

​Development Mode

​Production Mode

​Custom Adapter Usage

​Environment Detection

​Tencent Cloud SCF

​Custom Environment

​Endpoints

​POST /send-message

​GET /healthz

​POST /chat/completions (Optional)

​Best Practices

​1. Use Resource Cleanup

​2. Configure CORS Properly

​3. Use Health Checks

​4. Handle Errors Gracefully

​5. Use Type Hints

​Related Documentation

Server API

Installation

Core Components

AGKitAPIApp

Methods

Adapter Functions

create_send_message_adapter()

create_openai_adapter()

Data Models

AgentCreator

AgentCreatorResult

SendMessageInput

OpenAIChatCompletionRequest

HealthzConfig

HealthzResponse

Usage Examples

Quick Start

With Resource Cleanup

With CORS Configuration

Advanced Customization

With Health Check Configuration

OpenAI-Compatible Endpoint

Development Mode

Production Mode

Custom Adapter Usage

Environment Detection

Tencent Cloud SCF

Custom Environment

Endpoints

POST /send-message

GET /healthz

POST /chat/completions (Optional)

Best Practices

1. Use Resource Cleanup

2. Configure CORS Properly

3. Use Health Checks

4. Handle Errors Gracefully

5. Use Type Hints

Related Documentation