Skip to main content

​
ag-kit.server

Provide a FastAPI service as adapters for serving AG-Kit agents. This package offers an API class and an Function: -Create a FastAPI server with mountable agent routes -Create FastAPI routes mountable to an existing FastAPI app -Expose an agent handler as a standard FastAPI route interface All responses stream over Server‑Sent Events (SSE).

​
Installation

pip install ag_kit_py

​
What you get

  • FastAPI helpers: AGKitAPIApp, create_server_adapter
  • HTTP adapter: create_server_adapter for streaming responses
  • Health endpoint: /healthz route

​
Quick start

from ag_kit_py.server import AGKitAPIApp

app = AGKitAPIApp()

app.run(agent=myAgent);

​
Endpoints

When using Express helpers:
  • POST /send-message — accepts a JSON SendMessageInput and streams events as SSE
  • GET /healthz — returns OK

​
Request and events

Request body schema (validated via Pydantic): 
from typing import Any, List, Literal, Optional, Union
from pydantic import BaseModel, Field

class ToolFunction(BaseModel):
    name: str
    arguments: str

class ToolCall(BaseModel):
    id: str
    type: Literal["function"] = "function"
    function: ToolFunction

class Tool(BaseModel):
    name: str
    description: str
    parameters: Any

class SystemMessage(BaseModel):
    role: Literal["system"] = "system"
    content: str

class UserMessage(BaseModel):
    role: Literal["user"] = "user"
    content: str

class ToolMessage(BaseModel):
    role: Literal["tool"] = "tool"
    content: str
    tool_call_id: str = Field(..., alias="toolCallId")

class AssistantMessage(BaseModel):
    id: str
    role: Literal["assistant"] = "assistant"
    content: Optional[str] = None
    tool_calls: Optional[List[ToolCall]] = Field(None, alias="toolCalls")

ClientMessage = Union[SystemMessage, UserMessage, ToolMessage, AssistantMessage]

class ResumeMessage(BaseModel):
    interruptId: str
    payload: str

class SendMessageInput(BaseModel):
    messages: Optional[List[ClientMessage]] = []
    tools: Optional[List[Tool]] = []
    resume: Optional[ResumeMessage] = None
    conversationId: str
SSE event payloads: 
class ErrorEvent(BaseModel):
    type: Literal["error"] = "error"
    error: str

class TextEvent(BaseModel):
    type: Literal["text"] = "text"
    content: str

class ToolCallStartEvent(BaseModel):
    type: Literal["tool-call-start"] = "tool-call-start"
    tool_call_id: str = Field(..., alias="toolCallId")
    tool_call_name: str = Field(..., alias="toolCallName")

class ToolCallArgsEvent(BaseModel):
    type: Literal["tool-call-args"] = "tool-call-args"
    tool_call_id: str = Field(..., alias="toolCallId")
    delta: str

class ToolCallEndEvent(BaseModel):
    type: Literal["tool-call-end"] = "tool-call-end"
    tool_call_id: str = Field(..., alias="toolCallId")

class ToolCallResultEvent(BaseModel):
    type: Literal["tool-result"] = "tool-result"
    tool_call_id: str = Field(..., alias="toolCallId")
    result: str

class InterruptEvent(BaseModel):
    type: Literal["interrupt"] = "interrupt"
    id: str
    reason: Optional[str] = None
    payload: Any

SendMessageEvent = Union[
    ErrorEvent, InterruptEvent, TextEvent, ToolCallArgsEvent, ToolCallEndEvent, ToolCallResultEvent, ToolCallStartEvent
]
See the detailed API and Events pages for full reference.

​
Next steps