别再让AI单打独斗了：用MCP协议手把手教你搭建一个能‘对话’的智能体协作系统

从零构建智能体协作系统基于MCP协议的周末旅行规划实战想象一下这样的场景周五晚上你对着手机说帮我规划一个去杭州的周末旅行30秒后一份完整的行程建议出现在屏幕上——包含根据实时天气推荐的穿搭、避开人流高峰的景点路线、符合你饮食偏好的餐厅预订甚至还有基于社交媒体评价生成的游玩小贴士。这背后不是某个超级AI的单打独斗而是一群各有所长的智能体通过MCP协议协作的结果。1. 环境准备与基础架构搭建1.1 开发环境配置首先确保你的开发环境满足以下要求Python 3.10pip 23.0虚拟环境推荐使用venv或conda安装核心依赖库pip install mcp-client mcp-server fastapi uvicorn[standard]1.2 系统架构设计我们的旅行规划系统将由以下智能体组成智能体类型职责描述通信协议行程协调器任务分解与结果聚合MCP/WebSocket天气查询器获取目的地天气预报MCP/HTTP景点推荐器基于用户偏好生成景点列表MCP/gRPC餐厅预订代理对接第三方API完成订座MCP/WebSocket交通规划器整合航班/火车/租车信息MCP/HTTP2. 实现第一个智能体天气查询服务2.1 创建基础服务用FastAPI搭建一个支持MCP协议的天气查询服务from fastapi import FastAPI from mcp.server import McpServer from pydantic import BaseModel app FastAPI() mcp_server McpServer(app) class WeatherRequest(BaseModel): location: str days: int 2 mcp_server.register_operation(weather.query) async def get_weather(params: WeatherRequest): # 这里应该是调用真实天气API的代码 return { location: params.location, forecast: [ { date: 2024-06-15, high_temp: 28, low_temp: 22, condition: partly_cloudy } ] }2.2 能力声明与注册每个智能体启动时需要向中央注册表声明自己的能力{ agent_id: weather-agent-v1, capabilities: [ { operation: weather.query, description: 提供2天内天气预报, input_schema: { type: object, properties: { location: {type: string}, days: {type: integer, default: 2} }, required: [location] } } ], endpoint: http://localhost:8000/mcp }3. 构建行程协调器系统的大脑3.1 任务分解逻辑实现协调器需要理解用户意图并分解为子任务class TripPlanner: def __init__(self): self.client McpClient(registry_urlhttp://registry:8080) async def plan_trip(self, destination: str, dates: str): # 创建主任务上下文 context { session_id: str(uuid.uuid4()), user_intent: fplan_trip_to_{destination}, preferences: {} # 从用户画像加载 } # 并行发起子任务 weather_task self.client.call_async( operationweather.query, arguments{location: destination}, contextcontext ) attractions_task self.client.call_async( operationattractions.recommend, arguments{location: destination}, contextcontext ) # 等待并聚合结果 results await asyncio.gather(weather_task, attractions_task) return self._compile_itinerary(results)3.2 上下文管理技巧有效的上下文管理是多轮对话的关键会话保持使用WebSocket保持长连接状态压缩定期对历史对话进行摘要异常恢复实现checkpoint机制def summarize_context(history: list) - str: 使用LLM生成对话摘要 prompt f请用一句话总结以下对话的核心需求\n{history} return llm.generate(prompt)4. 系统集成与实战调试4.1 端到端测试流程按照以下步骤验证系统功能启动注册中心服务依次启动各功能智能体运行协调器主程序发送测试请求curl -X POST http://localhost:8000/plan \ -H Content-Type: application/json \ -d {destination:杭州,dates:2024-06-15}4.2 常见问题排查开发过程中可能会遇到这些典型问题问题现象可能原因解决方案能力注册失败注册中心URL错误检查网络连接和端点配置上下文丢失会话ID未正确传递实现中间件自动注入任务超时子智能体响应慢设置合理的timeout参数权限拒绝JWT令牌过期实现自动刷新机制5. 进阶优化与扩展思路5.1 性能优化策略当系统需要处理高并发请求时# 使用连接池管理智能体连接 from mcp.client import ConnectionPool pool ConnectionPool( max_size20, idle_timeout300 ) async with pool.get_client() as client: response await client.call(...)5.2 扩展系统能力通过MCP协议可以轻松集成新智能体用户画像分析器基于历史行为提供个性化推荐实时交通监控动态调整行程路线紧急预案生成器处理天气突变等意外情况mcp_server.register_operation(emergency.replan) async def handle_emergency(context: dict): weather context.get(last_weather) if weather[condition] heavy_rain: return {new_plan: 建议改为室内活动...}6. 生产环境部署指南6.1 安全配置要点确保系统安全的关键措施传输加密强制使用WSS替代WS认证鉴权实现基于角色的访问控制输入验证严格校验所有传入参数# 在FastAPI中添加安全中间件 app.add_middleware( SecurityMiddleware, jwt_secretos.getenv(JWT_SECRET) )6.2 监控与日志完善的观测体系应包括性能指标请求延迟、错误率业务指标任务完成率、平均规划时间审计日志记录所有关键操作app.middleware(http) async def log_requests(request, call_next): start_time time.time() response await call_next(request) process_time (time.time() - start_time) * 1000 logger.info(f{request.method} {request.url} - {process_time:.2f}ms) return response在实际项目中我们发现最影响用户体验的不是单个智能体的能力上限而是协作系统处理边界情况的能力。比如当餐厅订满时系统能否自动寻找替代选项这需要精心设计错误处理流程和备选方案生成逻辑。