MCP Client

MCP クライアント

Server-Sent Events (SSE) を使用して Model Control Protocol (MCP) サーバーと通信するための Python クライアントです。

アーキテクチャ

Server-Sent Events (SSE)

クライアントは Server-Sent Events (SSE) を使用して MCP サーバーとの永続的な接続を確立し、リアルタイムのサーバー間通信を実現します。このアーキテクチャにより、サーバーは継続的なポーリングを必要とせずにクライアントに更新をプッシュできます。

主要な SSE イベント：

• endpoint: セッション固有のメッセージングエンドポイントを提供します
• tools: 利用可能なツールと機能を配信します
• error: サーバー側のエラーを通信します
• result: ツール実行結果を返します

接続フロー

1. クライアントが /sse エンドポイントへの SSE 接続を開始します
2. サーバーが endpoint イベント経由でセッション ID に応答します
3. サーバーが tools イベント経由で利用可能なツールを送信します
4. クライアントがリアルタイム更新のための永続的な接続を維持します
5. ツール呼び出しが SSE エンドポイントへの POST リクエスト経由で行われます
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 経由のツール実行を検証します

エラーハンドリング

クライアントはいくつかのエラーハンドリングメカニズムを実装しています：

• 接続タイムアウト検出
• 自動再接続の試行
• 適切なエラー報告
• セッション状態の検証

開発

クライアントは非同期 HTTP と SSE 処理のために Python の aiohttp ライブラリを使用しています。