プロトコルからエンジニアリングへ：MCPサーバーを真剣に扱うべき理由
Agentアプリケーションを構築する際、私たちはしばしば誤った考えに陥ります：ツール呼び出しを単純にHTTPインターフェースにラップし、大規模モデルにFunction Callingで直接呼び出させるというものです。このアプローチはプロトタイプ段階では効率的ですが、本番環境に入ると、認証の混乱、コンテキストの喪失、ツール状態の非同期、エラーハンドリングの不整合など、問題が次々と発生します。MCP（Model Context Protocol）は、これらの痛点を解決するために生まれたオープンプロトコルであり、クライアント（Claude Desktopや自社Agentなど）とサーバー（ツール提供者）間の標準化された通信方法を定義し、ツールをUSBのようにプラグアンドプレイにします。
この記事では、概念の紹介に留まらず、ゼロから本番レベルのMCPサーバーを構築し、プロトコルの詳細、エンジニアリング上のトレードオフ、実際の落とし穴を深く掘り下げます。Python公式SDKを使用し、DeepSeek APIを実際のツールバックエンドとして統合し、モデルがMCPを通じて実際の大規模モデル機能を呼び出せるようにします。この記事を読み終えると、信頼性が高く、スケーラブルで、観測可能なMCPサービスを構築する能力が身につきます。
MCPプロトコルの核心メカニズム：JSON-RPCだけではない
MCPはJSON-RPC 2.0に基づいていますが、その価値は完全なライフサイクルと能力ネゴシエーションを定義することにあります。プロトコル層は、ツールリスト（tools/list）、ツール呼び出し（tools/call）、リソースとプロンプトという3つの主要な能力を規定しています。しかし、MCPを単純なRPCと区別するのは、「能力ネゴシエーション」メカニズムです。クライアントとサーバーは起動時にinitializeハンドシェイクを通じて、ストリーミング出力のサポート、進捗通知のサポート、ユーザー確認の必要性など、各自がサポートする機能を交換します。この設計により、MCPは単純なCLIから複雑なIDEプラグインまで、さまざまなホスト環境に適応できます。
エンジニアリングの観点では、MCPのトランスポート層を理解する必要があります。現在、主流は2つあります：stdio（標準入出力）とStreamable HTTPです。stdioはローカルサブプロセスモードに適しており、デバッグが簡単ですが、ネットワークを越えることはできません。HTTPは分散デプロイに適していますが、認証、再接続、タイムアウトの処理が必要です。本番環境では、HTTPトランスポートを使用し、APIゲートウェイを統一エントリポイントとして組み合わせることを強くお勧めします。以下では、Python SDKを使用してHTTPベースのMCPサーバーフレームワークを構築し、実行可能なコードを提供します。
最初のMCPサーバーを構築する：環境とスケルトン
まず、公式Python SDKをインストールする必要があります：pip install mcp。バージョンはPython 3.10以上が必要です。DeepSeekのチャット補完APIを呼び出すためのcall_deepseekツールを公開する最小限のサーバーをセットアップしましょう。注意：MCPサーバーは本質的に非同期アプリケーションであり、パフォーマンスを向上させるためにuvloopの使用をお勧めします。
from mcp.server import Server, stdio_server
from mcp.types import Tool, TextContent, CallToolResult
import json, httpx
app = Server('deepseek-mcp')
# ツール定義をラップする非同期静的メソッド
@app.list_tools()
async def list_tools() -> list[Tool]:
return [Tool(
name='call_deepseek',
description='DeepSeekチャット補完APIを呼び出し、モデル生成テキストを返す',
inputSchema={
'type': 'object',
'properties': {
'prompt': {'type': 'string', 'description': 'ユーザー入力'},
'max_tokens': {'type': 'integer', 'default': 1024}
},
'required': ['prompt']
}
)]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
if name == 'call_deepseek':
prompt = arguments['prompt']
max_tokens = arguments.get('max_tokens', 1024)
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.post(
'https://api.deepseek.com/chat/completions',
headers={'Authorization': 'Bearer your-deepseek-api-key'},
json={
'model': 'deepseek-chat',
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens
}
)
result = resp.json()
return CallToolResult(content=[TextContent(type='text', text=result['choices'][0]['message']['content'])])
raise ValueError(f'Unknown tool: {name}')
# stdioモードで実行
if __name__ == '__main__':
import asyncio
from mcp.server.stdio import stdio_server
asyncio.run(stdio_server(app))
このコードは動作しますが、本番レベルにはまだ遠いです。エラーハンドリング、タイムアウト再試行、同時実行制限、セキュリティ検証などを考慮する必要があります。さらに、MCP公式HTTPトランスポートの設定はSDKで若干異なるため、後のセクションで完全なソリューションを提供します。
ツール呼び出しの深掘り：入力検証とエラーハンドリングの技術
ツール定義のinputSchemaは単なるドキュメントではなく、クライアントがパラメータ検証に使用し、モデルのパラメータ生成方法にも影響を与えます。スキーマが不合理だと、モデルは多くの無効な呼び出しを生成します。JSON Schema仕様に従い、型、必須項目、列挙、デフォルト値を厳密に定義することをお勧めします。しかし、より重要なのは、実行関数が防御的プログラミングの意識を持つことです：パラメータが異常な場合、例外を投げてサーバー全体をクラッシュさせるのではなく、構造化されたエラー情報を返すべきです。
実際のプロジェクトでは、すべてのツール関数をラップする統一デコレータを使用し、例外をキャッチしてMCPエラーオブジェクトに変換します。さらに、多くのツールはユーザーIDやセッションIDなどのコンテキストを運ぶ必要があります。MCPリクエストオブジェクトにはメタ情報が含まれています、ログ追跡用のトレースIDを渡すことができます。以下は改良版のエラーハンドリング例です：
from mcp.types import ErrorData, CallToolResult
import traceback
def safe_tool(handler):
async def wrapper(name, arguments, **ctx):
try:
return await handler(name, arguments, **ctx)
except Exception as e:
traceback.print_exc()
return CallToolResult(isError=True, content=[TextContent(type='text', text=f'エラー: {str(e)}')])
return wrapper
@app.call_tool()
@safe_tool
async def call_tool(name: str, arguments: dict) -> CallToolResult:
# ... 実装
ここでもう一つ落とし穴があります。DeepSeek APIのレスポンス構造はエラーによって異なる可能性があるため、HTTPステータスコードとレスポンスボディのerrorフィールドを必ず確認する必要があります。また、大規模言語モデルAPIのレイテンシは通常非常に高い（0.5〜2秒）ため、MCPサーバーは非同期並行処理をサポートする必要があります。そうしないと、遅いリクエストがすべてのツール呼び出しをブロックします。Pythonのasyncioは自然に適していますが、httpxの共有クライアント接続プール設定に注意する必要があります。
本番デプロイ：HTTPトランスポートとセキュリティ認証
stdioトランスポートはローカルテストにのみ適しています。本番環境では、MCPサーバーを独立したサービスとして公開する必要があります。公式SDKは、Starletteフレームワークに基づくStreamableHttpServerを提供しています。クライアントのリクエストヘッダーと一致するように、Bearer Tokenを検証するなどの認証ミドルウェアを設定する必要があります。以下に、uvicornで実行する完全なHTTPサーバー起動例を示します。
from mcp.server.streamable_http_server import StreamableHttpServer
from mcp.server import Server
# アプリのトランスポートをオーバーライド
server = Server('deepseek-http-mcp')
# ... ツールを登録（上記と同じ）
http_server = StreamableHttpServer(server)
# 認証ミドルウェアを追加
from starlette.middleware.base import BaseHTTPMiddleware
class AuthMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
token = request.headers.get('Authorization')
if token != 'Bearer your-mcp-server-token':
return Response(status_code=401)
return await call_next(request)
# ASGIアプリケーションにマウント
from starlette.applications import Starlette
from starlette.responses import JSONResponse
async def endpoint(request):
return await http_server.handle(request)
app = Starlette()
app.add_middleware(AuthMiddleware)
app.add_route('/mcp', endpoint, methods=['POST', 'GET'])
import uvicorn
uvicorn.run(app, host='0.0.0.0', port=8000)
セキュリティ面では、MCPツールが機密操作に関与する可能性があるため、必ずHTTPSを使用してください。さらに、MCPプロトコルはセッション管理をサポートしており、クライアントごとにセッションIDを生成し、認証情報をバインドできます。Claude Desktopのclaude_desktop_config.jsonにカスタムMCPサーバーを追加するなど、クライアント設定方法をドキュメントで明確に説明することをお勧めします。ただし、本番環境では、APIゲートウェイを介してキーとレート制限を一元管理することをより推奨します。
パフォーマンス最適化：接続再利用と並行性チューニング
httpxを使用してDeepSeekを呼び出す場合、デフォルトではリクエストごとに新しい接続が作成され、高並行性の下でTCPハンドシェイクのオーバーヘッドが発生します。共有のAsyncClientを使用し、接続プールのサイズとタイムアウトを設定する必要があります。以下はチューニングされたクライアント設定です：
import httpx
client = httpx.AsyncClient(
base_url='https://api.deepseek.com',
headers={'Authorization': 'Bearer your-deepseek-api-key'},
timeout=httpx.Timeout(connect=5, read=30, write=10, pool=10),
limits=httpx.Limits(max_connections=50, max_keepalive_connections=20)
)
async def deepseek_chat(prompt, max_tokens):
resp = await client.post('/chat/completions', json={
'model': 'deepseek-chat',
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': max_tokens
})
resp.raise_for_status()
data = resp.json()
if 'error' in data:
raise RuntimeError(f"APIエラー: {data['error']['message']}")
return data['choices'][0]['message']['content']
さらに、大規模言語モデルAPIの価格とレート制限のため、MCPレイヤーでレート制限（トークンバケットアルゴリズムなど）を実装して、クライアントの乱用を防ぐ必要があります。また、1回の呼び出しで最大トークン数を設定し、ユーザーリクエストが大きすぎてタイムアウトやコストの急増を防ぐ必要があります。ツール定義でmax_tokensの範囲を制限し、サーバー側で検証することをお勧めします。
テストと可観測性：MCPサービスを信頼性の高いものにする
MCPサーバーには厳格なテストが必要です。まず、単体テストツール関数を直接呼び出すことはできますが、APIレスポンスをモックする必要があります。pytest-asyncioとrespxを使用してhttpxをモックすることをお勧めします。次に、統合テストでは、公式のテストクライアントを使用して完全なハンドシェイクと呼び出しフローをシミュレートする必要があります。簡単なテストケースを次に示します。
import pytest, respx
from httpx import Response
@respx.mock
async def test_call_deepseek():
from mcp.server import Server
# 呼び出しを構築（低レベルAPI経由の可能性）
# ここでは単なる例示
本番環境ではログとメトリクスが必須です。私は各ツール呼び出しに構造化ログを追加しています：呼び出し時刻、ツール名、パラメータ、応答時間、エラー情報を記録します。structlog または標準の logging を使用します。また、Prometheus を介してメトリクス（呼び出し回数、エラー率、レイテンシ分布など）を公開します。DeepSeek API の呼び出しでは、トークン使用量とコストを記録してコスト監視を行うことをお勧めします。
もう一つの見落とされがちな点は、グレースフルシャットダウンとタイムアウト制御です。MCPサーバーは複数の長時間実行呼び出しを同時に処理する可能性があります。プロセスがSIGTERMを受け取ったとき、新しいリクエストの受け入れを停止し、既存のリクエストの完了を待つ（または最大待機時間を設定する）必要があります。HTTPサーバーでは、uvicornのshutdown_timeoutを使用して制御できます。
実践経験：最大の落とし穴3つと解決策
落とし穴1：スキーマとモデルの整合性問題。ツールの説明やパラメータの仕様が曖昧だと、モデルがパラメータを適当に入力します。例えば、promptパラメータを要求し、「ユーザー入力」と説明すると、モデルは「こんにちは」と入力するかもしれませんが、問題ありません。しかし、パラメータが非構造化JSONの場合、モデルが不正なJSONを生成して解析に失敗する可能性があります。解決策は、説明にテンプレート例を提供し、パラメータに対して二次的なJSON解析と検証を行うことです。
落とし穴2：ストリーミング応答の欠如。MCPプロトコルはストリーミング応答をサポートしていますが、多くのクライアントは即時フィードバックを期待します。大規模モデルAPIでは、長時間応答がないとユーザー体験が非常に悪くなります。MCPツール呼び出しでストリーミング転送を実装することをお勧めします。ここではStreamableHTTPのSSEサポートを使用できます。ただし、これにより複雑さが増します。対話型のシナリオであれば、ストリーミングの実装を強くお勧めします。
落とし穴3：コンテキスト管理の混乱。MCPツールはステートレスですが、多くのビジネスではマルチターン会話の状態が必要です。私の経験では、状態をサーバーメモリに保存せず、クライアントがconversation IDや履歴を渡すようにします。ツール内部ではデータベースやRedisでIDに基づいて状態を取得します。これにより水平スケーラビリティが保証されます。
ライブラリからプロダクトへ：拡張とエコシステム
最後に、孤立したMCPサーバーを構築しないでください。より大きなAgentシステムに組み込む方法を考えるべきです。例えば、リクエストに応じて異なるMCPサーバーに振り分ける「ツールルーター」を構築できます。DeepSeekモデル自体はMCPを気にしませんが、MCPサーバーを介してリアルタイムデータやプライベートAPIなどにアクセスさせることができます。この組み合わせが本番級アプリケーションの王道です。
もう一つの方向性は、MCPのリソースメカニズム（resources/list）を使用してナレッジベースやデータベーステーブルを公開し、モデルがコンテキストを取得できるようにすることです。これはツール呼び出しと補完的です。DeepSeekモデルを統合してRAGを実装できますが、コスト管理に注意してください。最後に、MCPサーバーがマルチクライアントの同時接続をサポートしているか、特にSDKのスレッドセーフティをテストしてください。
まとめと発展パス
この記事では、プロトコル原理から本番実装まで、ツール定義、エラー処理、HTTPデプロイ、パフォーマンス最適化、テスト、運用を網羅しました。要点：MCPは単なるRPCではなく、その価値は標準化と組み合わせ可能性にあります。能力ネゴシエーションとトランスポート層を深く理解し、実際のニーズに基づいて工学的な決定を行う必要があります。
さらに進むには、MCP公式仕様の「ベストプラクティス」セクションを読み、既存のMCPサーバー実装（GitHubの公式リファレンスなど）を研究することをお勧めします。次のステップとして、MCPサーバーをDockerイメージにパッケージ化し、Kubernetesでデプロイして自動スケーリングを実現できます。また、APIゲートウェイ（Kongなど）を統合して統一認証を行うこともできます。
開発中に具体的な問題が発生した場合は、コメント欄でお知らせください。このチュートリアルを継続的に更新し、より多くの実例を追加します。