MCPサーバー

Name: MCP Server
Author: freedanfan

プロジェクト概要

FastAPIとMCP（モデルコンテキストプロトコル）を基盤とするこのプロジェクトは、AIモデルと開発環境間の標準化されたコンテキストインタラクションを可能にします。モデルのデプロイメントを簡素化し、効率的なAPIエンドポイントを提供し、モデルの入出力の一貫性を確保することで、AIアプリケーションのスケーラビリティと保守性を向上させ、開発者によるAIタスクの統合と管理を容易にします。

MCP（モデルコンテキストプロトコル）は、AIモデルと開発環境間のコンテキスト相互作用のための統一プロトコルです。このプロジェクトは、初期化、サンプリング、セッション管理といったMCPプロトコルの基本的な機能をサポートするPythonベースのMCPサーバー実装を提供します。

Related MCP server: OpenAI MCP Server

特徴

JSON-RPC 2.0 : 標準JSON-RPC 2.0プロトコルに基づくリクエスト-レスポンス通信
SSE 接続: リアルタイム通知のための Server-Sent Events 接続のサポート
モジュラー設計: 拡張とカスタマイズが容易なモジュラーアーキテクチャ
非同期処理：FastAPIと非同期IOを使用した高性能サービス
完全なクライアント: 完全なテストクライアントの実装が含まれています

プロジェクト構造

mcp_server/ ├── mcp_server.py # MCP server main program ├── mcp_client.py # MCP client test program ├── routers/ │ ├── __init__.py # Router package initialization │ └── base_router.py # Base router implementation ├── requirements.txt # Project dependencies └── README.md # Project documentation

インストール

リポジトリをクローンします。

git clone https://github.com/freedanfan/mcp_server.git cd mcp_server

依存関係をインストールします:

pip install -r requirements.txt

使用法

サーバーの起動

python mcp_server.py

デフォルトでは、サーバーは127.0.0.1:12000で起動します。環境変数を使用してホストとポートをカスタマイズできます。

export MCP_SERVER_HOST=0.0.0.0 export MCP_SERVER_PORT=8000 python mcp_server.py

クライアントの実行

別のターミナルでクライアントを実行します。

python mcp_client.py

サーバーがデフォルトのアドレスで実行されていない場合は、環境変数を設定できます。

export MCP_SERVER_URL="http://your-server-address:port" python mcp_client.py

APIエンドポイント

サーバーは次の API エンドポイントを提供します。

ルートパス（ / ）：サーバー情報を提供します
APIエンドポイント（ /api ）：JSON-RPCリクエストを処理する
SSEエンドポイント（ /sse ）：SSE接続を処理する

MCPプロトコルの実装

初期化フロー

クライアントはSSE経由でサーバーに接続します
サーバーはAPIエンドポイントURIを返します
クライアントはプロトコルバージョンと機能を含む初期化要求を送信します
サーバーは初期化要求に応答し、サーバーの機能を返します

サンプリングリクエスト

クライアントはプロンプトを使用してサンプリングリクエストを送信できます。

{ "jsonrpc": "2.0", "id": "request-id", "method": "sample", "params": { "prompt": "Hello, please introduce yourself." } }

サーバーはサンプリング結果を返します:

{ "jsonrpc": "2.0", "id": "request-id", "result": { "content": "This is a response to the prompt...", "usage": { "prompt_tokens": 10, "completion_tokens": 50, "total_tokens": 60 } } }

セッションの終了

クライアントはシャットダウン要求を送信できます。

{ "jsonrpc": "2.0", "id": "request-id", "method": "shutdown", "params": {} }

サーバーは正常にシャットダウンします。

{ "jsonrpc": "2.0", "id": "request-id", "result": { "status": "shutting_down" } }

開発拡張

新しいメソッドの追加

新しい MCP メソッドを追加するには、 MCPServerクラスにハンドラー関数を追加し、 _register_methodsメソッドに登録します。

def handle_new_method(self, params: dict) -> dict: """Handle new method""" logger.info(f"Received new method request: {params}") # Processing logic return {"result": "success"} def _register_methods(self): # Register existing methods self.router.register_method("initialize", self.handle_initialize) self.router.register_method("sample", self.handle_sample) self.router.register_method("shutdown", self.handle_shutdown) # Register new method self.router.register_method("new_method", self.handle_new_method)

AIモデルの統合

実際の AI モデルを統合するには、 handle_sampleメソッドを変更します。

async def handle_sample(self, params: dict) -> dict: """Handle sampling request""" logger.info(f"Received sampling request: {params}") # Get prompt prompt = params.get("prompt", "") # Call AI model API # For example: using OpenAI API response = await openai.ChatCompletion.acreate( model="gpt-4", messages=[{"role": "user", "content": prompt}] ) content = response.choices[0].message.content usage = response.usage return { "content": content, "usage": { "prompt_tokens": usage.prompt_tokens, "completion_tokens": usage.completion_tokens, "total_tokens": usage.total_tokens } }

トラブルシューティング

よくある問題

接続エラー: サーバーが実行中であること、およびクライアントが正しいサーバー URL を使用していることを確認してください
405 メソッドが許可されていません: クライアントが正しい API エンドポイントにリクエストを送信していることを確認してください
SSE接続失敗: ネットワーク接続とファイアウォールの設定を確認してください

ログ記録

サーバーとクライアントの両方で詳細なログが提供されます。詳細についてはログを参照してください。

# Increase log level export PYTHONPATH=. python -m logging -v DEBUG -m mcp_server

参考文献

ライセンス

このプロジェクトはMITライセンスの下で提供されています。詳細はLICENSEファイルをご覧ください。

MCP Server