MCPプロトコル実践：Model Context ProtocolでAIエージェントツールチェーンを構築

AI与大数据

概要

  • MCP（Model Context Protocol）はAnthropicが提唱したAIエージェントツール呼び出し標準プロトコルであり、2026年にはデファクトスタンダードなエージェントツールチェーンプロトコルとなりました
  • MCPはJSON-RPC 2.0によりツールの登録・発見・呼び出しを実現し、AIエージェントの「ツール孤島」問題を解決します
  • 本記事では、MCPプロトコルの原理からサーバー開発、ツール登録からマルチエージェントオーケストレーションまで、全工程の完全なコードを解説します
  • MCPはSSEとStdioの2つの転送モードをサポートしています。SSEはリモートデプロイに適し、Stdioはローカル開発に適しています
  • おまけ：MCPサーバーの本番デプロイメント方案とツールチェーンセキュリティ監査チェックリスト付き

目次

なぜAIエージェントにMCPプロトコルが必要なのか

2026年以前、AIエージェントのツール呼び出し方式は「バラバラ」でした。OpenAIはFunction Calling、LangChainはTool Abstraction、AutoGPTはカスタムプラグインを使用していました。各フレームワークが独自のツールインターフェースを定義しており、ツールはフレームワーク間で再利用できず、エージェントはプラットフォームを越えた協力ができませんでした。

` ┌──────────────────────────────────────────────────────────────┐ │ AIエージェントツール呼び出しの「バベルの塔」 │ │ │ │ OpenAI Agent ──→ Function Calling (JSON Schema) │ │ LangChain ──→ Tool Abstraction (Python Class) │ │ AutoGPT ──→ Plugin System (YAML Config) │ │ Dify ──→ Tool Node (API Config) │ │ Coze ──→ Plugin Market (Proprietary) │ │ │ │ ❌ ツール再用不可 ❌ インターフェース不統一 ❌ セキュリティ未保証 ❌ 発見標準なし │ └──────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐ │ MCPプロトコル：AIエージェントの「USB-C」 │ │ │ │ あらゆるエージェント ←──→ MCPプロトコル ←──→ あらゆるツール │ │ │ │ ┌────────┐ MCP ┌──────────┐ MCP ┌──────────┐ │ │ │ Claude │←──────→│ 検索ツール│←──────→│ GPT-4 │ │ │ │ Agent │ │ DB │ │ Agent │ │ │ └────────┘ │ API GW │ └──────────┘ │ │ └──────────┘ │ │ │ │ ✅ ツール再利用 ✅ インターフェース統一 ✅ セキュリティサンドボックス ✅ 自動発見 │ └──────────────────────────────────────────────────────────────┘ `

MCP vs Function Calling vs LangChain Tool

項目 MCP OpenAI Function Calling LangChain Tool
プロトコル標準 オープン標準（Anthropic） ベンダー独自 フレームワーク独自
ツール発見 自動発見（capabilities） 手動登録 手動登録
転送プロトコル JSON-RPC 2.0（SSE/Stdio） HTTP API Python関数呼び出し
セキュリティサンドボックス ✅ 権限制御 ❌ ❌
クロスエージェント再利用 ✅ 任意のMCPクライアント ❌ OpenAIのみ ❌ LangChainのみ
ストリーミング出力 ✅ SSE ✅ SSE ⚠️ 部分対応
マルチツールオーケストレーション ✅ ネイティブサポート ⚠️ 手動対応が必要 ✅ Chain抽象

参考：Model Context Protocol Specification

MCPプロトコルのコアメカニズム

3つのロール

ロール 説明 例え
MCP Host 接続を開始するAIアプリケーション（Claude Desktop、IDEプラグインなど） USBホスト
MCP Client MCPサーバーとの接続を確立するプロトコルクライアント。Hostに組み込まれる USBコントローラ
MCP Server ツール、リソース、プロンプトを提供するサーバー USBデバイス

プロトコル相互作用フロー

┌──────────────────────────────────────────────────────────────┐ │ MCPプロトコル相互作用フロー │ │ │ │ MCP Host (Claude Desktop) │ │ │ │ │ │ 1. initialize │ │ ├──→ MCP Client ──→ MCP Server（検索ツール） │ │ │ │ │ │ │ 2. capabilitiesレスポンス │ │ │ │←─────────────────────────────────────────┘ │ │ │ │ │ │ 3. tools/list │ │ ├──→ MCP Client ──→ MCP Server │ │ │ │ │ │ │ 4. ツール一覧レスポンス │ │ │ │←─────────────────────────────────────────┘ │ │ │ │ │ │ 5. tools/call（「K8s GPUスケジューリング」を検索） │ │ ├──→ MCP Client ──→ MCP Server │ │ │ │ │ │ │ 6. ツール実行結果 │ │ │ │←─────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────────┘

MCP機能タイプ

機能 説明 例
Tools エージェントから呼び出し可能な関数 検索、DBクエリ、API呼び出し
Resources エージェントから読み取り可能なデータ ファイル、DBレコード、設定
Prompts エージェントが使用可能なプロンプトテンプレート コードレビューテンプレート、翻訳テンプレート
Sampling サーバーがエージェントにLLM補完を要求 マルチステップ推論、コンテンツ生成

MCPサーバー開発：5ステップでカスタムツールサービスを構築

ステップ1：プロジェクト初期化

ash mkdir mcp-search-server && cd mcp-search-server pip install mcp fastapi httpx

ステップ2：ツールスキーマの定義

`python from mcp.server import Server from mcp.types import Tool, TextContent import httpx import json

server = Server("search-tools")

@server.list_tools() async def list_tools() -> list[Tool]: return [ Tool( name="web_search", description="インターネットを検索して最新情報を取得します。リアルタイムデータ、ニュース、技術ドキュメントの検索が必要な場合に使用してください。", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "検索キーワード" }, "max_results": { "type": "integer", "description": "最大返却結果数", "default": 5 }, "search_depth": { "type": "string", "enum": ["basic", "advanced"], "description": "検索深度：basicは簡易検索、advancedは詳細検索", "default": "basic" } }, "required": ["query"] } ), Tool( name="code_search", description="GitHubでコードリポジトリとコードスニペットを検索します。オープンソース実装やAPI使用例の検索が必要な場合に使用してください。", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "コード検索キーワード" }, "language": { "type": "string", "description": "プログラミング言語フィルター", "enum": ["python", "go", "rust", "typescript", "java"] } }, "required": ["query"] } ), Tool( name="api_tester", description="HTTPリクエストを送信してAPIエンドポイントをテストします。APIの可用性確認やレスポンスデータの取得が必要な場合に使用してください。", inputSchema={ "type": "object", "properties": { "url": { "type": "string", "description": "APIエンドポイントURL" }, "method": { "type": "string", "enum": ["GET", "POST", "PUT", "DELETE"], "default": "GET" }, "headers": { "type": "object", "description": "リクエストヘッダー" }, "body": { "type": "object", "description": "リクエストボディ（JSON）" } }, "required": ["url"] } ), ] `

ステップ3：ツールロジックの実装

`python @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: if name == "web_search": return await handle_web_search(arguments) elif name == "code_search": return await handle_code_search(arguments) elif name == "api_tester": return await handle_api_tester(arguments) else: raise ValueError(f"Unknown tool: {name}")

async def handle_web_search(args: dict) -> list[TextContent]: query = args["query"] max_results = args.get("max_results", 5) search_depth = args.get("search_depth", "basic")

async with httpx.AsyncClient(timeout=30) as client:
    response = await client.post(
        "https://api.search.brave.com/res/v1/web/search",
        headers={"X-Subscription-Token": BRAVE_API_KEY},
        params={
            "q": query,
            "count": max_results,
            "search_depth": search_depth,
        }
    )
    data = response.json()

results = []
for item in data.get("web", {}).get("results", [])[:max_results]:
    results.append({
        "title": item.get("title", ""),
        "url": item.get("url", ""),
        "description": item.get("description", ""),
    })

return [TextContent(
    type="text",
    text=json.dumps(results, ensure_ascii=False, indent=2)
)]

async def handle_code_search(args: dict) -> list[TextContent]: query = args["query"] language = args.get("language", "")

async with httpx.AsyncClient(timeout=30) as client:
    response = await client.get(
        "https://api.github.com/search/code",
        headers={"Authorization": f"token {GITHUB_TOKEN}"},
        params={
            "q": f"{query} language:{language}" if language else query,
            "per_page": 5,
        }
    )
    data = response.json()

results = []
for item in data.get("items", [])[:5]:
    results.append({
        "name": item.get("name", ""),
        "path": item.get("path", ""),
        "repository": item.get("repository", {}).get("full_name", ""),
        "html_url": item.get("html_url", ""),
    })

return [TextContent(
    type="text",
    text=json.dumps(results, ensure_ascii=False, indent=2)
)]

async def handle_api_tester(args: dict) -> list[TextContent]: url = args["url"] method = args.get("method", "GET") headers = args.get("headers", {}) body = args.get("body")

async with httpx.AsyncClient(timeout=30) as client:
    response = await client.request(
        method=method,
        url=url,
        headers=headers,
        json=body,
    )

return [TextContent(
    type="text",
    text=json.dumps({
        "status_code": response.status_code,
        "headers": dict(response.headers),
        "body": response.json() if "json" in response.headers.get("content-type", "") else response.text[:2000],
    }, ensure_ascii=False, indent=2)
)]

`

ステップ4：SSE転送の設定

`python from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Route

sse = SseServerTransport("/messages")

async def handle_sse(request): async with sse.connect_sse(request) as streams: await server.run(streams[0], streams[1], server.create_initialization_options())

async def handle_messages(request): await sse.handle_post_message(request)

app = Starlette( routes=[ Route("/sse", endpoint=handle_sse), Route("/messages", endpoint=handle_messages, methods=["POST"]), ] )

if name == "main": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8080) `

ステップ5：Dockerデプロイ

`dockerfile FROM python:3.11-slim

WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . .

EXPOSE 8080 HEALTHCHECK --interval=30s --timeout=5s --retries=3
CMD curl -f http://localhost:8080/sse || exit 1

CMD ["python", "server.py"] `

MCPクライアント統合：LLMにツールを呼び出させる

Python MCPクライアント

`python from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from openai import OpenAI import json

server_params = StdioServerParameters( command="python", args=["server.py"], env={"BRAVE_API_KEY": "...", "GITHUB_TOKEN": "..."} )

async def run_agent(user_query: str): async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize()

        tools_result = await session.list_tools()
        available_tools = [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema,
                }
            }
            for tool in tools_result.tools
        ]

        client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")

        messages = [{"role": "user", "content": user_query}]

        response = client.chat.completions.create(
            model="Qwen/Qwen2.5-7B-Instruct",
            messages=messages,
            tools=available_tools,
            tool_choice="auto",
        )

        message = response.choices[0].message

        if message.tool_calls:
            for tool_call in message.tool_calls:
                result = await session.call_tool(
                    tool_call.function.name,
                    arguments=json.loads(tool_call.function.arguments),
                )

                messages.append(message)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result[0].text,
                })

            final_response = client.chat.completions.create(
                model="Qwen/Qwen2.5-7B-Instruct",
                messages=messages,
            )
            return final_response.choices[0].message.content

        return message.content

`

ツール呼び出しフロー

┌──────────────────────────────────────────────────────────────┐ │ MCPエージェントツール呼び出し完全フロー │ │ │ │ ユーザー：「K8s GPUスケジューリングの最新手法を検索し、 │ │ 関連APIをテストして」 │ │ │ │ │ ▼ │ │ LLMが意図を分析 → 2つのツールの呼び出しが必要： │ │ 1. web_search("K8s GPUスケジューリング 2026") │ │ 2. api_tester("https://kubernetes.io/docs/concepts/") │ │ │ │ │ ├──→ MCP Client ──→ web_search ──→ 検索結果を返す │ │ │ │ │ ├──→ MCP Client ──→ api_tester ──→ APIレスポンスを返す │ │ │ │ │ ▼ │ │ LLMが結果を統合 → 最終回答を生成 │ └──────────────────────────────────────────────────────────────┘

マルチエージェントオーケストレーション：MCPツールチェーンの組み合わせ

マルチサーバーオーケストレーションアーキテクチャ

`python from mcp import ClientSession from mcp.client.sse import sse_client import asyncio

class MCPToolOrchestrator: def init(self): self.servers = {} self.sessions = {}

async def register_server(self, name: str, url: str):
    self.servers[name] = url

async def connect_all(self):
    for name, url in self.servers.items():
        read, write = await sse_client(url)
        session = ClientSession(read, write)
        await session.initialize()
        self.sessions[name] = session

async def discover_all_tools(self) -> list[dict]:
    all_tools = []
    for name, session in self.sessions.items():
        tools = await session.list_tools()
        for tool in tools.tools:
            all_tools.append({
                "server": name,
                "name": tool.name,
                "description": tool.description,
                "inputSchema": tool.inputSchema,
            })
    return all_tools

async def call_tool(self, tool_name: str, arguments: dict) -> str:
    for name, session in self.sessions.items():
        tools = await session.list_tools()
        if any(t.name == tool_name for t in tools.tools):
            result = await session.call_tool(tool_name, arguments)
            return result[0].text
    raise ValueError(f"Tool {tool_name} not found in any server")

async def close_all(self):
    for session in self.sessions.values():
        await session.close()

orchestrator = MCPToolOrchestrator() await orchestrator.register_server("search", "http://search-mcp:8080/sse") await orchestrator.register_server("database", "http://db-mcp:8081/sse") await orchestrator.register_server("code", "http://code-mcp:8082/sse") await orchestrator.connect_all() `

K8sでMCPサーバークラスタをデプロイ

`yaml apiVersion: apps/v1 kind: Deployment metadata: name: mcp-search-server namespace: ai-agent spec: replicas: 2 selector: matchLabels: app: mcp-search-server template: metadata: labels: app: mcp-search-server spec: containers: - name: mcp-server image: myregistry/mcp-search-server:v1.0 ports: - containerPort: 8080 resources: requests: cpu: "1" memory: 512Mi limits: cpu: "2" memory: 1Gi env: - name: BRAVE_API_KEY valueFrom: secretKeyRef: name: mcp-secrets key: brave-api-key livenessProbe: httpGet: path: /sse port: 8080 initialDelaySeconds: 10 periodSeconds: 30

apiVersion: v1 kind: Service metadata: name: mcp-search-svc namespace: ai-agent spec: selector: app: mcp-search-server ports: - port: 8080 targetPort: 8080 `

本番デプロイとセキュリティ監査

MCPセキュリティモデル

セキュリティ層 メカニズム 説明
転送層 TLS + SSE 暗号化転送、中間者攻撃を防止
認証層 OAuth 2.0 / API Key サーバーがクライアントの身元を検証
認可層 ツール権限ホワイトリスト 呼び出し可能なツールの範囲を制限
データ層 入力バリデーション + 出力フィルタリング インジェクション攻撃とデータ漏洩を防止
監査層 呼び出しログ + 計量 フルチェーン追跡とコスト制御

セキュリティ監査チェックリスト

チェック項目 リスクレベル チェック方法
ツール入力パラメータのバリデーション 高 スキーマ検証 + 型チェック
ツール出力データのマスキング 高 機密情報の正規表現フィルタリング
APIキーのハードコード禁止 高 環境変数 / Secret管理
SSE接続タイムアウト制御 中 ハートビート + タイムアウト切断
ツール呼び出しレート制限 中 Rate Limiting
呼び出しログの完全性 中 構造化ログ + 監査トレイル
エラーメッセージの内部状態漏洩防止 低 統一エラーフォーマット

MCPサーバーパフォーマンスベンチマーク

指標 単一インスタンス 3インスタンスクラスタ
ツール呼び出しQPS 500 1400
平均レイテンシ（P50） 45ms 50ms
P99レイテンシ 180ms 200ms
SSE接続数 200 600
メモリ使用量 256MB 256MB×3

まとめと関連記事

MCPプロトコルはAIエージェントツールチェーンの「USB-C」です。1つのオープン標準により、ツール孤島、インターフェース不統一、セキュリティ未保証という3つの大きな問題を解決します。MCPサーバー開発、クライアント統合、マルチエージェントオーケストレーションを通じて、本番レベルのAIエージェントツールチェーンを迅速に構築できます。

開発のポイント振り返り：

  1. MCPはJSON-RPC 2.0によりツールの登録・発見・呼び出しを実現します
  2. SSEモードはリモートデプロイに適し、Stdioモードはローカル開発に適しています
  3. ツールスキーマ定義はMCPのコアであり、パラメータとセマンティクスを詳細に記述する必要があります
  4. マルチエージェントオーケストレーションはMCPToolOrchestratorによりクロスサーバーのツール呼び出しを実現します
  5. 本番デプロイではセキュリティ監査に注力が必要です：入力バリデーション、出力マスキング、レート制限

関連記事：

権威ある参考資料：

ブラウザローカルツールを無料で試す →

#MCP协议#AI Agent工具调用#Model Context Protocol#大模型工具使用#Agent开发框架#2026