跳转到主要内容

​
ag-kit.server

提供FastAPI服务作为适配器，用于托管AG-Kit Agent。该包包含一个API类和一个函数：
  • 创建带有可挂载Agent路由的FastAPI服务器
  • 创建可挂载到现有FastAPI应用的路由
  • 将Agent处理器暴露为标准FastAPI路由接口
所有响应均通过服务器推送事件(SSE)流式传输。

​
安装

pip install ag_kit_py

​
功能特性

  • FastAPI辅助工具：AGKitAPIApp、create_server_adapter
  • HTTP适配器：用于流式响应的create_server_adapter
  • 健康检查端点：/healthz路由

​
快速开始

​
运行（推荐方式）

from ag_kit_py.server import AGKitAPIApp

app = AGKitAPIApp()

app.run(agent=myAgent);

​
端点

使用Express辅助工具时：
  • POST /send-message — 接收JSON格式的SendMessageInput并以SSE流式传输事件
  • GET /healthz — 返回OK

​
请求与事件

请求体结构（通过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事件负载： 
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
]
完整参考请查阅详细API和事件文档。

​
后续步骤