跳转到主要内容
模型上下文协议（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函数创建自定义工具

MCP服务器注册表

发现社区提供的可用MCP服务器