AgentKit SDK 概览
AgentKit SDK 是一个 Python 开发工具包,用于快速构建符合 AgentKit Platform 标准的 Agent 应用。SDK 提供了一套简洁的装饰器和客户端接口,帮助开发者专注于 Agent 业务逻辑的实现。
核心架构
AgentKit SDK 采用模块化设计,主要包含以下模块:
1. Runtime 应用框架
提供多种应用模式,将用户代码封装为符合 Platform 标准的 HTTP 服务:
AgentkitSimpleApp
最常用的应用框架,适用于标准的 Agent 应用。
核心装饰器:
@app.entrypoint- 定义 Agent 的主入口函数@app.ping- 定义健康检查函数@app.async_task- 定义异步任务(规划中)
标准路由:
/invoke- Agent 调用端点/ping- 健康检查/health、/readiness、/liveness- Kubernetes 就绪探针
AgentkitMCPApp
基于 MCP (Model Context Protocol) 协议的应用框架,用于将工具封装为 MCP 服务。
核心装饰器:
@app.tool- 将函数注册为 MCP 工具@app.agent_as_a_tool- 将 Agent 封装为 MCP 工具
AgentkitA2aApp
基于 A2A (Agent-to-Agent) 协议的应用框架,用于构建可互相通信的 Agent。
核心装饰器:
@app.agent_executor- 注册 Agent 执行器@app.task_store- 注册任务存储(可选)
2. Platform 服务客户端
提供便捷的客户端接口,用于访问 AgentKit Platform 的各项服务。
AgentkitMemory
记忆服务客户端,管理 Agent 的长期和短期记忆。
主要功能:
create_memory_collection()- 创建记忆库list_memory_collections()- 列举记忆库get_memory_collection()- 获取记忆库详情update_memory_collection()- 更新记忆库配置delete_memory_collection()- 删除记忆库get_memory_connection_info()- 获取连接信息
设计理念:控制面和数据面解耦
- 控制面(SDK 提供):管理记忆库的创建、配置和获取连接信息
- 数据面(ADK 实现):通过 MemoryBase 或 Mem0 协议进行实际的记忆读写
AgentkitKnowledge
知识库服务客户端,管理 Agent 的向量化知识库。
主要功能:
list_knowledge_bases()- 列举知识库add_knowledge_base()- 添加知识库get_knowledge_connection_info()- 获取知识库连接信息delete_knowledge_base()- 删除知识库
使用场景:
- RAG(检索增强生成)应用
- 文档问答系统
- 知识图谱集成
AgentkitMCP
MCP Gateway 客户端,用于管理和路由大量 MCP 工具。
主要功能:
- 注册和管理 MCP 服务
- 工具聚合和智能路由
- 降低工具集成复杂度
AgentkitRuntime
Runtime 管理客户端,用于管理 Agent 的运行时环境。
主要功能:
- 创建和管理 Runtime 实例
- 查询 Runtime 状态
- 配置环境变量和资源
3. 基础设施模块
BaseAgentkitClient
所有客户端的基类,提供统一的配置和 HTTP 请求能力。
配置项:
- API 端点配置
- 认证凭证管理
- 请求重试和超时
- 日志和监控
Context 管理
使用 contextvars 管理每个请求的上下文信息,包括:
- 用户身份
- 会话 ID
- 请求元数据
- 链路追踪信息
模块依赖关系
AgentKit SDK
│
├── Runtime Apps (应用框架层)
│ ├── AgentkitSimpleApp
│ ├── AgentkitMCPApp
│ ├── AgentkitA2aApp
│ └── AgentkitAgentServerApp
│
├── Platform Clients (服务客户端层)
│ ├── AgentkitMemory
│ ├── AgentkitKnowledge
│ ├── AgentkitMCP
│ └── AgentkitRuntime
│
└── Infrastructure (基础设施层)
├── BaseAgentkitClient
├── Context
└── Telemetry (可观测性)主要特性
1. 装饰器驱动
使用 Python 装饰器简化应用开发,开发者只需关注业务逻辑:
python
from agentkit.apps import AgentkitSimpleApp
app = AgentkitSimpleApp()
@app.entrypoint
async def run(payload: dict, headers: dict) -> str:
# 业务逻辑
return response2. 框架无关
SDK 不限制使用特定的 Agent 框架,支持:
- Volcengine ADK (veadk)
- Google ADK (gadk)
- LangChain
- 自定义框架
3. 自动化的可观测性
内置 OpenTelemetry 支持,自动收集:
- 链路追踪(Tracing)
- 性能指标(Metrics)
- 日志(Logging)
4. 生产就绪
提供完整的生产环境支持:
- 健康检查端点
- 优雅关闭
- 错误处理
- 请求重试
5. 类型安全
完整的 Python 类型注解,提供:
- IDE 自动补全
- 类型检查
- 更好的代码可维护性
快速开始
安装
bash
pip install agentkit-sdk-python创建一个简单的 Agent
python
from agentkit.apps import AgentkitSimpleApp
from veadk import Agent, Runner
from veadk.tools.demo_tools import get_city_weather
app = AgentkitSimpleApp()
agent = Agent(tools=[get_city_weather])
runner = Runner(agent=agent)
@app.entrypoint
async def run(payload: dict, headers: dict) -> str:
prompt = payload["prompt"]
user_id = headers["user_id"]
session_id = headers["session_id"]
response = await runner.run(
messages=prompt,
user_id=user_id,
session_id=session_id
)
return response
@app.ping
def ping() -> str:
return "pong!"
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)使用 Platform 服务
python
from agentkit.sdk.memory import AgentkitMemory
from agentkit.base_client import ApiConfig
# 初始化客户端
config = ApiConfig(
access_key="your_ak",
secret_key="your_sk"
)
memory_client = AgentkitMemory(config)
# 创建记忆库
response = memory_client.create_memory_collection(
name="my-memory",
short_term_configuration={...},
long_term_configuration={...}
)
print(f"Memory ID: {response.id}")下一步
- 查看注解使用文档了解各种装饰器的详细用法
- 参考
samples/目录下的完整示例 - 阅读 Platform 服务客户端的 API 文档
