MCP Client

MCP 客户端

使用服务器发送事件 (SSE) 与 Model Control Protocol (MCP) 服务交互的 Python 客户端。

架构

服务器发送事件 (SSE)

客户端使用服务器发送事件 (SSE) 与 MCP 服务建立持久连接，启用实时的服务器到客户端通信。这种架构允许服务器将更新推送到客户端，无需持续轮询。

关键 SSE 事件：

• endpoint：提供会话特定的消息传递端点
• tools：传递可用工具和功能
• error：传达服务器端错误
• result：返回工具执行结果

连接流程

1. 客户端向 /sse 端点建立 SSE 连接
2. 服务器通过 endpoint 事件响应会话 ID
3. 服务器通过 tools 事件发送可用工具
4. 客户端维持持久连接以获取实时更新
5. 工具调用通过 POST 请求发送到 SSE 端点
6. 结果通过 SSE 连接流式返回

使用

# Install dependencies
pip install -r requirements.txt

# Run the client
python client_sse.py

环境变量

• MCP_SERVER_URL：MCP 服务的 URL
• MCP_API_KEY：用于身份验证的 API 密钥
• ANTHROPIC_API_KEY：用于 Claude 集成的 API 密钥

功能

• 与 MCP 服务的实时双向通信
• 自动会话管理
• 动态工具发现和集成
• 长时间运行操作的流式响应
• 健壮的错误处理和连接管理

服务实现指南

TODO：向 MCP 服务添加 SSE 支持

要修改现有 MCP 服务以支持 SSE，请按照以下步骤操作：

1. 添加 SSE 依赖

   import { EventEmitter } from 'events';
   import { FastifyInstance } from 'fastify';
   

2. 创建会话管理

   interface Session {
     id: string;
     emitter: EventEmitter;
     tools: Tool[];
   }
   
   const sessions = new Map<string, Session>();
   

3. 实现 SSE 端点

   server.get('/sse', async (request, reply) => {
     const sessionId = generateSessionId();
     const session = createSession(sessionId);
     
     reply.raw.setHeader('Content-Type', 'text/event-stream');
     reply.raw.setHeader('Cache-Control', 'no-cache');
     reply.raw.setHeader('Connection', 'keep-alive');
     
     // Send initial session info
     reply.raw.write(`event: endpoint\ndata: /messages/?session_id=${sessionId}\n\n`);
     
     // Send available tools
     const tools = await getTools();
     reply.raw.write(`event: tools\ndata: ${JSON.stringify(tools)}\n\n`);
     
     // Handle client disconnect
     request.raw.on('close', () => {
       sessions.delete(sessionId);
     });
   });
   

4. 修改工具处理器

   server.post('/sse', async (request, reply) => {
     const { session_id, tool, args } = request.body;
     const session = sessions.get(session_id);
     
     if (!session) {
       throw new Error('Invalid session');
     }
     
     try {
       const result = await executeToolCall(tool, args);
       reply.raw.write(`event: result\ndata: ${JSON.stringify(result)}\n\n`);
     } catch (error) {
       reply.raw.write(`event: error\ndata: ${JSON.stringify(error)}\n\n`);
     }
   });
   

5. 更新服务配置

   const server = new Server({
     name: "mcp-sse-server",
     version: "1.0.0"
   }, {
     capabilities: {
       resources: {},
       tools: {},
       sse: true  // Enable SSE capability
     }
   });
   

6. 错误处理

   server.setErrorHandler(async (error, request, reply) => {
     const sessionId = request.body?.session_id;
     const session = sessions.get(sessionId);
     
     if (session) {
       session.emitter.emit('error', error);
     }
     reply.code(500).send(error);
   });
   

7. 会话清理

   function cleanupSessions() {
     const now = Date.now();
     for (const [id, session] of sessions.entries()) {
       if (now - session.lastActive > SESSION_TIMEOUT) {
         sessions.delete(id);
       }
     }
   }
   
   setInterval(cleanupSessions, CLEANUP_INTERVAL);
   

关键注意事项

1. 连接管理

   • 实现心跳机制
   • 优雅地处理重新连接
   • 清理不活跃的会话

2. 安全性

   • 验证会话 ID
   • 实现速率限制
   • 添加身份验证中间件

3. 性能

   • 监控活跃会话的内存使用情况
   • 实现连接池
   • 为重型操作添加请求队列

4. 测试

   • 为 SSE 端点添加单元测试
   • 测试连接边界情况
   • 验证通过 SSE 的工具执行

错误处理

客户端实现了多种错误处理机制：

• 连接超时检测
• 自动重新连接尝试
• 优雅的错误报告
• 会话状态验证

开发

客户端使用 Python 的 aiohttp 库进行异步 HTTP 和 SSE 处理。