Documentation Index
Fetch the complete documentation index at: https://docs.ag-kit.dev/llms.txt
Use this file to discover all available pages before exploring further.
模型上下文协议（MCP）是一种连接人工智能应用与外部数据源和工具的开放标准。AG-Kit 提供全面的 MCP 集成支持，使您能够轻松扩展 Agent 的能力。
什么是 MCP？
MCP 采用客户端-服务器架构，允许人工智能应用安全访问外部资源：
- MCP 服务器：提供工具、数据源和提示模板
- MCP 客户端：消费这些资源的人工智能应用（如 AG-Kit Agent）
使用场景
1. 连接现有工具生态
许多开发工具和服务已提供可直接在 AG-Kit 中使用的 MCP 服务器：
2. 暴露 AG-Kit 工具
将您的 AG-Kit 工具封装为 MCP 服务器供其他人工智能应用使用：
- 内部工具共享：跨团队复用工具
- 跨平台集成：与其他 AI 框架互操作
- 服务化部署：将工具作为独立服务提供
快速入门
使用外部 MCP 工具
最常见场景是连接现有 MCP 服务器以访问其工具：
from agkit.tools import create_mcp_toolkit
import os
# 使用对象映射创建 MCP 工具包（推荐）
mcp_toolkit = await create_mcp_toolkit({
'playwright': {
'command': 'npx',
'args': ['@playwright/mcp@latest']
}
})
# 获取工具（addServer 后自动加载）
tools = mcp_toolkit.get_tools()
暴露 AG-Kit 工具
将您的 AG-Kit 工具作为 MCP 服务器暴露：
from agkit.tools.mcp import AGKitMCPServer
from agkit.tools import tool
from pydantic import BaseModel, Field
from typing import Literal
class WeatherInput(BaseModel):
city: str = Field(description="城市名称")
units: Literal['celsius', 'fahrenheit'] = Field(default='celsius')
async def weather_func(input_data: WeatherInput):
# 模拟天气 API 调用
weather_data = {
'city': input_data.city,
'temperature': 22 if input_data.units == 'celsius' else 72,
'condition': 'sunny',
'humidity': 65
}
return {
'success': True,
'data': weather_data
}
# 创建 AG-Kit 工具
weather_tool = tool(
func=weather_func,
name='get_weather',
description='获取指定城市的天气信息',
schema=WeatherInput
)
# 创建 MCP 服务器
server = AGKitMCPServer(
name='weather-service',
version='1.0.0'
)
# 注册工具
server.register_tool(weather_tool)
# 启动服务器
await server.run({'type': 'stdio'})
连接类型
标准输入/输出（Stdio）
最常见的连接类型，适用于本地工具和命令行程序：
# 使用对象映射格式（推荐）
toolkit = await create_mcp_toolkit({
'python-tools': {
'command': 'python',
'args': ['-m', 'my_mcp_server'],
'timeout': 10000
},
'node-tools': {
'command': 'npx',
'args': ['@company/mcp-server']
}
})
HTTP 连接
适用于远程服务和 Web 应用：
# 使用对象映射格式（推荐）
toolkit = await create_mcp_toolkit({
'remote-api': {
'url': 'https://api.example.com/mcp',
'timeout': 15000
}
})
内存连接
用于测试和进程内通信：
# 服务端
await server.run({
'type': 'memory',
'memory_id': 'test-server'
})
# 客户端
toolkit = await create_mcp_toolkit({
'test-server': {
'memory_id': 'test-server'
}
})
实际案例
多服务器集成
实际项目中可能需要连接多个 MCP 服务器以获得不同能力：
import os
# 使用对象映射创建综合工具包（推荐）
toolkit = await create_mcp_toolkit({
'filesystem': {
'command': 'npx',
'args': ['@modelcontextprotocol/server-filesystem', './workspace']
},
'database': {
'command': 'python',
'args': ['-m', 'database_server'],
'env': {
'DATABASE_URL': os.getenv('DATABASE_URL')
}
},
'search': {
'url': 'https://search-api.company.com/mcp'
}
}, 'enterprise-tools')
# 从特定服务器获取工具
fs_tools = toolkit.get_server_tools('filesystem')
db_tools = toolkit.get_server_tools('database')
search_tools = toolkit.get_server_tools('search')
# 或获取所有工具
all_tools = toolkit.get_tools()
错误处理与监控
生产环境需要全面的错误处理和监控：
import asyncio
# 服务器事件监控
def server_event_handler(event):
if event.type == 'connected':
print(f"客户端已连接: {event.client_name}")
elif event.type == 'tool_called':
print(f"工具被调用: {event.tool_name}", event.arguments)
elif event.type == 'error':
print(f"服务器错误: {event.error}")
# 发送至监控系统
monitoring.report_error(event.error)
server.add_event_listener(server_event_handler)
# 客户端连接管理
toolkit = MCPToolkit('monitored-toolkit')
def client_event_handler(event):
if event.type == 'connected':
print(f"已连接至服务器: {event.client_name}")
elif event.type == 'disconnected':
print(f"与服务器断开连接: {event.client_name}")
# 尝试重连
asyncio.create_task(asyncio.sleep(5)).add_done_callback(
lambda _: asyncio.create_task(toolkit.refresh())
)
elif event.type == 'error':
print(f"客户端错误: {event.error}")
toolkit.add_event_listener(client_event_handler)
后续步骤
需要高级功能？本指南涵盖了基本的MCP集成。有关连接池、重试策略、事件监控、模式转换和性能优化等生产功能，请参阅完整的MCP参考。 完整的MCP参考
高级功能、MCPClientManager和生产配置
函数工具
了解如何从TypeScript函数创建自定义工具
