AIエージェントシステムを構築する際、エージェントが外部ツールやデータを安全かつ確実に呼び出し、かつ動的に発見できることが共通の課題となります。この課題を解決するため、大規模言語モデル（LLM）と外部データソースまたはツールの間で標準化されたインターフェースプロトコルを提供するために、Model Context Protocol (MCP)が開発されました。

簡単に言えば、MCPはLLM（クライアントとして機能）とデータ/ツールプロバイダー（MCPサーバーとして機能）間の通信ルールを定義します。これにより、以下のいくつかの主要なエンジニアリング上の課題が解決されます。

分散したツール管理: ツールの発見、記述、呼び出し方法を統一します。
困難なコンテキストと状態管理: コンテキストの注入と転送メカニズムを標準化します。
曖昧なパーミッションとセキュリティ境界: プロトコル層でツールとデータアクセスに対する明確な境界を確立します。
異種システム統合の複雑性: 標準プロトコルを通じて異なるバックエンドシステムを橋渡しします。

MCPサーバーの典型的なシナリオ:

マルチツールAIエージェント: エージェントがクエリ、書き込み、計算、API操作など、さまざまな機能を柔軟に組み合わせて呼び出す必要がある場合。
社内システム/プライベートデータへのアクセス: 社内データベース、CRM、ナレッジベースなどをAIエージェントに安全に公開する場合。
統一されたコンテキストと権限を必要とするエージェントアーキテクチャ: 複雑な多段階ワークフローで、ユーザーID、セッション状態、データアクセス権限を管理する場合。

MCPサーバーに不適切なシナリオ:

単一のAPI呼び出し: ユーザーのクエリを固定のAPIエンドポイントにマッピングするだけであれば、従来のFunction Callingの方が軽量である可能性があります。
シンプルなチャットボット/フォームベースの自動化: タスクロジックが固定されており、動的なツールの発見や組み合わせが不要なシナリオ。

主な違い:

Function Callingとの比較: Function CallingはLLMプロバイダー（例：OpenAI）による独自の機能実装であり、関数を記述・呼び出すために使用されますが、そのプロトコルおよびトランスポート層は通常オープンではありません。一方、MCPはオープンでベンダーに依存しないプロトコル標準であり、サーバー・クライアント間の通信モデル、ツールの発見（list_tools）、リソース処理のメカニズムを定義し、上位のAIエージェントアーキテクチャのための標準化された基盤を提供します。
Webhookとの比較: Webhookは通知目的でイベントによってトリガーされる一方通行のHTTPコールバックです。MCPは、クライアント（LLM）がツール実行結果やデータリソースをオンデマンドで取得するために開始する双方向の要求-応答またはストリーミング通信を伴います。

本記事は、公式の Model Context Protocol (MCP) 仕様に基づいており、実際のAIエージェントのエンジニアリング実践に焦点を当て、企業におけるエージェントシステム統合とMCPサーバープロトタイピングに関する著者の経験をまとめています。

対象読者:

テクノロジー愛好家および初心者学習者
効率改善を求める専門家および管理者
企業の意思決定者および事業部門の責任者
AIの将来のトレンドに興味を持つ一般ユーザー

1. MCPサーバー構築前の技術選定とアーキテクチャ決定

1.1 公式MCPプロトコル実装

MCPの核となるのは、JSON-RPC 2.0に基づくメッセージングプロトコルです。公式SDK（TypeScriptやPythonなど）はプロトコルの詳細をカプセル化し、以下の主要な機能を提供します。

メッセージのシリアライズ/デシリアライズ: JSON-RPCリクエストとレスポンスを自動的に処理します。
トランスポート層の抽象化: STDIO（標準入出力）とHTTPを介したSSE（Server-Sent Events）の2つの標準トランスポート方式をサポートします。
サーバーライフサイクル管理: サーバーの初期化、ツール登録、起動プロセスを簡素化します。
型安全性（特にTypeScript）: ツール定義とリソース記述のための型インターフェースを提供します。

1.2 MCPサーバーの主要アーキテクチャコンポーネント

MCPサーバーのコンポーネントを理解することは、設計と実装の基本です。

ツール (Tools): サーバーがMCPクライアントに公開する実行可能な操作です。各ツールには名前、説明、および厳密に型付けされた入力パラメータスキーマ（JSON Schema）があります。会議室予約の例では、book_conference_roomがツールです。
リソース (Resources): MCPサーバーがMCPクライアントに提供する読み取り可能なデータユニットです。各リソースにはURI（Uniform Resource Identifier）、MIMEタイプ、およびオプションのテキストコンテンツがあります。例えば、company-holiday-calendarはLLMのコンテキストにリソースとして注入できます。
プロンプト (Prompts): LLMの動作をガイドするために、MCPクライアントに事前定義されたプロンプトテンプレートを提供するために使用されます。その有効性は、クライアントがこの機能を消費するかどうか、そしてどのように消費するかに依存します。
トランスポート (Transport): MCPサーバーとMCPクライアント間の通信方法です。MCP仕様では2つを定義しています。
1. Stdio: 標準入出力を介した通信です。サーバーがクライアントのサブプロセスとして起動されるシナリオに適しており、デプロイが容易です。
2. HTTP with Server-Sent Events (SSE): HTTP Server-Sent Eventsに基づいています。サーバーがスタンドアロンのネットワークサービスとして動作するシナリオに適しており、リモート接続をサポートします。

1.3 プログラミング言語の選定（エンジニアリングの観点から）

選択は、チームのスキル、パフォーマンス要件、およびデプロイ環境に依存します。

Python:
- 利点: 活発なエコシステム（特にAI/ML分野）、迅速な開発速度、成熟した公式SDK。迅速なプロトタイピングや、様々なPythonライブラリ（データ分析、MLモデルなど）との連携に理想的です。
- 並行処理モデル: 非同期I/Oのためのasyncioに基づいており、ネットワークリクエストやデータベースクエリのようなI/O集約型操作に適しています。CPU集約型タスクでは、Global Interpreter Lock (GIL)に注意が必要です。
- 最適なユースケース: 迅速な検証、データサイエンスツール、Pythonに精通したチーム。
Node.js (TypeScript):
- 利点: 高性能な非同期ノンブロッキングI/O、成熟した公式SDK、高並行性ネットワークサービスの構築に適しています。TypeScriptは優れた型安全性を提供します。
- 並行処理モデル: イベント駆動型であり、単一スレッドで高並行接続を処理する優れた能力を持っています。
- 最適なユースケース: 大量の並行ツール呼び出しが必要なシナリオや、フロントエンドまたはNode.jsバックエンドサービスとの深い統合。
Go:
- 利点: 静的コンパイル、シンプルなデプロイ、極めて高い並行処理性能（ゴルーチン）、高いメモリ効率。
- 課題: 現在、公式SDKが不足しており（コミュニティによる実装または自社開発が必要）、AI分野のエコシステムはPython/Nodeと比較してやや劣ります。
- 最適なユースケース: 性能とリソース消費に厳格な要件がある本番環境、またはGoベースの技術スタックを持つ企業。

2. 環境準備とMCP開発ツールチェーン

ここでは「会議室予約システム」を例として、デバッグ可能でデプロイ可能、そして持続可能なMCPサーバーをゼロから開発する方法を詳しく説明します。

以下のサンプルコードは、HTTP SSE通信モードを使用して最小限のMCPサーバーの実装を示すために、ローカル環境（Python 3.13.3、mcp Python SDK v1.25.0）で完全に検証されています。

2.1 最小動作可能なMCPサーバー環境

Pythonを例にとると、以下のものが必要です。

Python 3.13.3+: バージョンの互換性を確認してください。
MCP Python SDK: pip経由でインストールします。

pip install mcp
pip install pydantic

テキストエディタまたはIDE: VS Codeなど。
ターミナル: MCPサーバーの起動とテストのため。

2.2 公式MCP SDKとエコシステムツール

SDKの機能範囲: Python SDKのmcpライブラリは、サーバーの作成、ツール/リソースの定義、リクエストの処理のためのコアクラス（Server、Toolなど）を提供します。
サポートツール:
- MCP Inspector: MCPサーバーのデバッグとテストのためのグラフィカルなクライアントツールです。これを使用してサーバーに接続し、すべてのツール/リソースをリストし、手動でツールを呼び出すことができます。開発フェーズでは不可欠なツールです。
- CLIツール: 一部のSDKやコミュニティプロジェクトは、迅速なプロジェクト初期化のためのスキャフォールディングCLIを提供することがあります。

2.3 必須の開発ツールキットチェックリスト

ロギング: サーバーコードに構造化ロギングを統合し、リクエスト、パラメータ、エラーを記録します。これはデバッグの基盤となります。
スキーマ検証: SDKの型ヒントとJSON Schema検証機能を活用して、入力と出力が期待通りであることを保証します。
HTTPサーバー: このサンプルコードはHTTP SSE通信方式に基づいており、uvicornをHTTPサーバーとして使用しています。

3. 初めてのMCPサーバー：最小限ながら拡張可能な例

ここでは、book_conference_room ツールを実装します。ロジックは以下の通りです：予約リクエストの受信 → パラメータの確認 → データベース（インメモリ辞書でシミュレート）で指定された時間の空き状況を照会 → 空きがあれば予約された部屋番号を返却；空きがなければ「部屋が利用できません」という失敗メッセージを返却します。

3.1 ツールとリソース定義の設計原則

ツールの粒度:

ツールは論理的に独立した完全な操作を完了するべきです。book_conference_roomはその良い例です：入力は時間、出力は予約結果。
「ゴッドツール」（すべてを行う一つのツール）の作成は避けてください。また、過度な分割も避けてください（例えば、check_room_availabilityとconfirm_bookingが非常にアトミックである場合、それらをまとめても構いません）。

リソースとツールの境界:

リソースは、会社のポリシー文書や製品カタログなど、LLMが読み取るための静的または変化の遅いコンテキスト情報です。
ツールは、作成、更新、削除、計算など、副作用を伴う動的な操作です。
この例では、Meeting Room User Manualはリソースになり得ますが、予約操作はツールである必要があります。

コンテキスト転送のベストプラクティス:

ユーザーID、セッショントークンなどは、MCPリクエストのContextを通じて渡されるべきです（クライアントがサポートしている場合）。
サーバーはContext内のID情報を検証し、ビジネスロジックで使用すべきです。例えば、book_conference_roomは誰が予約しているかを知る必要があります。この情報はContextから取得できます。

3.2 リクエストとレスポンスの構造設計

Python SDKのToolクラスとPydanticモデルを使用して、厳密に型付けされた入力を定義します。

「conference_room_server.py」の内容:

import asyncio
from datetime import datetime, timedelta
from typing import Any

from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent, CallToolResult
from pydantic import BaseModel, Field
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.responses import Response

# 1. 入力パラメータ用のPydanticモデルを定義します。
class BookRoomInput(BaseModel):
    start_time: datetime = Field(..., description="会議開始時刻、形式：YYYY-MM-DD HH:MM")
    duration_hours: float = Field(..., gt=0, le=8, description="会議時間（時間）、最大8時間")

class RoomBookingSystem:
    def __init__(self):
        """ 2つの会議室をシミュレートします。値は既に予約されている時間枠のリストです。"""
        self.rooms = {"A101": [], "B202": []}

    def is_room_available(self, room_id: str, start: datetime, duration_hours: float) -> bool:
        """指定された会議室が指定期間中に利用可能かどうかを確認します。"""
        end = start + timedelta(hours=duration_hours)
        for b_start, b_end in self.rooms[room_id]:
            if not (end <= b_start or start >= b_end): return False
        return True

    def book_room(self, room_id: str, start: datetime, duration_hours: float, booker: str) -> bool:
        """会議室を予約しようと試み、成功したかどうかを返します。"""
        if self.is_room_available(room_id, start, duration_hours):
            self.rooms[room_id].append((start, start + timedelta(hours=duration_hours)))
            return True
        return False

booking_system = RoomBookingSystem()

# --- 2. MCPサーバーインスタンスを作成します ---
server = Server("conference-room-booking")

3.3 MCPサーバー呼び出しフローの詳細

次に、サーバーを作成し、ツールを処理するロジックを実装します。

# ツールリスト
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    return [
        Tool(
            name="book_conference_room",
            description="特定の時刻に会議室を予約します。",
            inputSchema=BookRoomInput.model_json_schema()
        )
    ]

# 会議室予約ロジック
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> CallToolResult:
    if name != "book_conference_room" or not arguments:
        return CallToolResult(is_error=True, content=[TextContent(type="text", text="パラメータエラー")])
    try:
        input_data = BookRoomInput(**arguments)
        booked = False
        booked_room = None
        for room_id in booking_system.rooms.keys():
            if booking_system.book_room(room_id, input_data.start_time, input_data.duration_hours, "demo_user"):
                booked, booked_room = True, room_id
                break
        res_text = f"予約成功！部屋 {booked_room}" if booked else "予約失敗"
        return CallToolResult(is_error=False, content=[TextContent(type="text", text=res_text)])
    except Exception as e:
        return CallToolResult(is_error=True, content=[TextContent(type="text", text=str(e))])

# --- 3. HTTP SSEトランスポート層の設定 ---
# SSEトランスポートインスタンスを作成します
sse = SseServerTransport("/messages")

async def handle_sse(request):
    """SSE接続を確立するためのクライアントリクエストを処理します。"""
    async with sse.connect_sse(
        request.scope,
        request.receive,
        request._send
    ) as (read_stream, write_stream):
        # MCPサーバーを実行します
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )
    return Response()


# Starletteのルーティングを定義します
app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages", app=sse.handle_post_message),
    ]
)

if __name__ == "__main__":
    import uvicorn
    # サービスを開始し、ポート8000でリッスンします
    print("MCP SSEサーバーは http://127.0.0.1:8000/sse で実行中です")
    uvicorn.run(app, host="127.0.0.1", port=8000)

呼び出しフロー分析:

クライアント → サーバー: クライアント（例：Claude Desktop、MCP Inspector）はHTTP SSE経由でlist_toolsリクエストを送信します。サーバーはツールリストを返します。
クライアントの決定: LLMはユーザーのコマンド（「明日午後2時から2時間、部屋を予約して」）に基づき、book_conference_roomを呼び出すことを決定します。
ツール実行: クライアントはnameとargumentsを含むcallToolリクエストを送信します。サーバーはそれをhandle_call_tool関数にルーティングします。
内部処理:
- パラメータ検証: Pydanticモデルがstart_timeの形式とduration_hoursの範囲を自動的に検証します。
- ビジネスロジック: シミュレートされた会議室を巡回し、空き状況を確認し、予約を試みます。
- 状態変更: 予約が成功すると、booking_system.roomsの状態が変更されます。
- ロギング: 予約ログを出力します。
結果の返却: サーバーはCallToolResultをJSON-RPCレスポンスにカプセル化し、クライアントに返送します。
コンテキスト注入: このツールがユーザーIDを必要とする場合、クライアントはcallToolリクエストのcontextフィールドにIDトークンを含めるべきです。サーバーのhandle_call_tool関数は、リクエストオブジェクトからこのトークンを解析し、検証する必要があります（この例では固定ユーザーに簡略化されています）。

達成目標: 上記は、完全で実行可能なMCPサーバーのMVP（Minimum Viable Product）です。明確な入力定義、ビジネスロジック、エラー処理、ロギングを備え、さらなる拡張のための強固な基盤を提供します。

4. MCPサーバーのデバッグとテスト方法

4.1 ローカルデバッグワークフロー

起動方法: python conference_room_server.pyを使用してPythonスクリプトを直接実行します。スクリプトはHTTP SSEサービスを起動し、イベントループに入り、MCPクライアントの接続を待ちます。
MCP Inspectorでの接続: これは最も効果的なデバッグ方法です。
- 「Tools」タブでbook_conference_roomを表示します。
- 「Resources」タブで定義されたリソース（存在する場合）を表示します。
- 「Session」タブでツールを直接呼び出し、パラメータを入力し、戻り結果とサーバーサイドのログを観察します。
ログ監視ポイント: ツールハンドラの開始時、終了時、例外捕捉ポイントでログを出力します。パラメータが正しく解析されているか、ビジネスロジックが期待通りに実行されているかを観察します。
一般的な起動失敗の原因:
- Pythonパスエラー: MCP Inspectorがpythonコマンドを見つけられない。
- 依存関係の欠落: mcpまたはpydanticライブラリがインストールされていない。
- スクリプト構文エラー: Pythonインタープリタは起動前にエラーを報告します。
- ポート競合（SSEモードのみ）：指定されたHTTPポートが既に占有されている。

4.2 MCPクライアントをシミュレートした機能テスト

Inspectorの他に、クライアントの挙動をシミュレートする簡単なテストスクリプトを作成できます。

「test_client.py」の内容:

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


async def test_booking_http():
    # 1. サーバーのSSEアドレスを定義します
    server_url = "http://127.0.0.1:8000/sse"

    print(f"MCP SSEサーバーに接続中: {server_url}...")

    try:
        # 2. 標準のsse_clientを使用してトランスポート層を確立します。
        async with sse_client(server_url) as (read_stream, write_stream):
            # 3. 標準のクライアントセッションを作成します
            # セッションは自動的に初期化ハンドシェイクを完了し、notifications/initialized通知を送信します。
            async with ClientSession(read_stream, write_stream) as session:

                # 初期化ハンドシェイク
                print("[1/3] プロトコルハンドシェイクを実行中...")
                await session.initialize()
                print("ハンドシェイクが成功しました！")

                # 4. すべてのツールをリスト表示します (tools/listに対応)
                print("\n[2/3] ツールリストを取得中...")
                tools_result = await session.list_tools()
                print(f"利用可能なツール: {[tool.name for tool in tools_result.tools]}")

                # 5. 事前定義されたツールを呼び出します (tools/callに対応)
                print("\n[3/3] 会議室の予約を試行中...")
                arguments = {
                    "start_time": "2025-12-25 14:00",
                    "duration_hours": 2.0
                }

                # `#call_tool` は、応答のカプセル化を自動的に処理する標準SDKメソッドです。
                result = await session.call_tool("book_conference_room", arguments)

                # 返されたコンテンツを解析します。
                for content in result.content:
                    if content.type == "text":
                        print(f"\n[サーバー応答]: {content.text}")

    except ConnectionRefusedError:
        print("エラー: サーバーに接続できません。サーバーがポート8000で実行されていることを確認してください。")
    except Exception as e:
        print(f"実行中にエラーが発生しました: {type(e).__name__}: {e}")


if __name__ == "__main__":
    asyncio.run(test_booking_http())

スキーマ検証失敗のトラブルシューティング: 呼び出しが失敗した場合は、Inspectorまたはテストスクリプトから返されるエラーメッセージを確認してください。通常、パラメータのタイプミス、型の不一致（例：整数が期待される場所に文字列を渡す）、または必須フィールドの欠落が原因です。

コンテキスト喪失のトラブルシューティング: ツールロジックがContextに依存しているにもかかわらず取得に失敗する場合は、以下を確認してください。

クライアントがContextを送信するように設定されているか。
サーバーサイドのハンドラがリクエストオブジェクトからContextデータを正しく抽出しているか。

4.3 一般的な開発段階のエラーチェックリスト

ツールが発見できない: list_toolsメソッドが正しく登録され、ツール定義を返しているか確認してください。サーバー起動時にすべてのツールが正常にロードされていることを確認してください。
タイムアウト: ツールハンドラの実行に時間がかかりすぎている。コードロジックを最適化するか、クライアント側で適切なタイムアウトを設定してください。長時間実行されるタスクの場合、非同期通知または結果ポーリングメカニズムの実装を検討してください。
パラメータの不一致: クライアントから送信されたパラメータのJSON構造が、ツールのinputSchemaと一致しない。Pydanticのような厳密なモデル検証を使用し、エラーメッセージで明確なプロンプトを提供してください。

5. MCPサーバーの本番デプロイパターン

デプロイは、独立したプロセスとしてのMCPサーバーの運用特性に焦点を当てるべきです。エージェントとの統合は通常、エージェントのサブプロセスとして（Stdio経由で）実行するか、または独立したネットワークサービスとして（HTTP/SSE経由で）実行することで行われます。

5.1 MCPサーバーのステートレス設計とセッション状態管理

中心原則：MCPサーバーは可能な限りステートレスであるべきです。

状態管理戦略: 複数のツール呼び出しをまたいで永続化する必要がある状態（ユーザーのショッピングカートや多段階承認フローの中間結果など）は、サーバープロセスメモリに保存すべきではありません。代わりに、以下の方法を取ります。
1. MCPのContextメカニズムを使用して、クライアントがすべてのリクエストに必要な状態を保持するようにします。
2. データベース、Redis、ファイルストレージなどの外部永続システムに保存します。サーバーツールは、これらの外部システムをクエリまたは更新することで状態を操作できます。
エージェントとの協調: エージェント（クライアント）は会話状態とユーザーの意図を維持する責任があります。MCPサーバーはアトミックなツール呼び出しにのみ応答します。例えば、会議室の予約はサーバーによって保証されるアトミックな操作ですが、「来週のチームミーティングに適切な時間を見つけて予約する」のような多段階の計画は、エージェントによって調整されるべきです。

5.2 デプロイメント比較：ローカル vs. クラウド vs. コンテナ化

ローカルデプロイメント: サーバーとエージェントクライアントが同じ物理または仮想マシン上で実行されます。開発、テスト、または小規模な社内アプリケーションに適しています。低コストですが、スケーラビリティと可用性が劣ります。
コンテナ化デプロイメント（推奨）: サーバーをDockerコンテナにパッケージ化します。これにより、環境の一貫性、スケーリングの容易さ（Kubernetes経由）、依存関係管理の簡素化において大きなメリットがあります。本番環境で推奨されます。
クラウドサーバーレスデプロイメント: サーバーをクラウド関数（例：AWS Lambda、Google Cloud Functions）としてデプロイします。呼び出しが infrequent なシナリオや、バースト的なトラフィックに適しています。ただし、コールドスタートのレイテンシと実行時間の制限に注意が必要で、長時間タスクや永続的なTCP/SSE接続を必要とするシナリオには適さない場合があります。

5.3 設定管理とシークレットのセキュリティ

コード内にシークレット、パスワード、APIトークン、データベース接続文字列をハードコードしないでください。

環境変数: 環境変数を通じて設定を注入します。Dockerでは-eフラグまたはシークレット管理を使用し、KubernetesではConfigMapsとSecretsを使用します。

import os
database_url = os.getenv('DATABASE_URL')
api_key = os.getenv('EXTERNAL_API_KEY')
if not database_url:
    raise ValueError("DATABASE_URL環境変数が設定されていません")

シークレット管理: HashiCorp Vault、AWS Secrets Manager、Azure Key Vaultのようなプロフェッショナルなシークレット管理サービスを使用します。アプリケーションは起動時にこれらのサービスから動的にキーを取得する必要があります。
ホット設定更新: レートリミットの閾値など、動的な調整が必要な設定については、外部データベースまたは設定センター（etcdやZooKeeperなど）に保存し、サーバーが変更をリッスンするようにします。頻繁なサーバー再起動は避けてください。

6. パフォーマンス、安定性、および高並行性プラクティス

本章での「高並行性」とは、MCPサーバー自体が処理できる並行ツール呼び出しリクエストを指します。レート制限とサーキットブレーキングは、サーバーまたはダウンストリームシステムを保護するためにサーバーによって実装される戦略です。

6.1 MCPツール呼び出しのための並行処理モデル

Python (asyncio): コルーチン並行処理のためにasync/awaitを使用します。ツールハンドラがデータベースクエリや外部API応答を待機している間（await状態）、イベントループは別のツールリクエストの処理に切り替えることができます。これはI/O集約型操作に理想的です。
- ポイント: すべてのブロッキングI/O操作が非同期ライブラリ（例：PostgreSQL用のasyncpg、HTTPリクエスト用のaiohttp）を使用していることを確認してください。
Node.js: 同様のイベントループベースの非同期モデルで、高並行I/Oを自然にサポートします。
Go: 各接続またはリクエストは通常、独立したゴルーチンによって処理され、マルチコア機能を活用し、I/OおよびCPU集約型タスクの両方で優れたパフォーマンスを発揮します。

I/O集約型最適化:

コネクションプールを使用してデータベースと外部サービスへの接続を管理します。
外部HTTPリクエストには適切なタイムアウトとリトライポリシーを設定します。
頻繁に読み取られ、変更が遅いデータにはキャッシュ層（Redisなど）の追加を検討します。

6.2 タイムアウト、リトライ、および障害境界の設計

タイムアウトの伝播: すべてのツール呼び出しに合計タイムアウトを設定します。ツールが内部で複数のダウンストリームサービスを呼び出す場合、それらのタイムアウトはツールの合計タイムアウトよりも短くし、ダウンストリームのタイムアウト例外は適切に処理する必要があります。
リトライのリスク: リトライの決定はクライアント（エージェント）が行うべきであり、サーバーツール内で自動的に行われるべきではありません。これは以下の理由によります。
- ツールがべき等ではない可能性があります（複数回実行しても同じ結果が得られない）。例えば、book_conference_roomがネットワークタイムアウトのためにリトライされると、重複予約につながる可能性があります。
- クライアントはより完全なコンテキスト（例：ユーザー指示）を持っており、リトライすべきかどうか、どのようにリトライすべきかを決定できます。
- サーバーは、クライアントが決定を下すのに役立つ明確で実行可能な情報をエラー応答で提供すべきです。
べき等性の要件: 書き込み操作のツールの場合、クライアントがユニークなリクエストIDを提供することで、サーバーが重複処理を回避できるように、べき等性を保つように設計することを試みてください。

6.3 レート制限、サーキットブレーキング、およびフォールバック戦略

ツールレベルの保護:
- レート制限: トークンバケットまたはリーキーバケットアルゴリズムを使用して、ツールごと、またはユーザー/APIキーごとのレートを制限します。これにより、単一のツールが過度に呼び出され、サーバーやダウンストリームサービスがクラッシュするのを防ぎます。
- サーキットブレーキング: ダウンストリームサービス（データベースや外部APIなど）が閾値を超えて連続的に失敗した場合、一時的にサーキットを「トリップ」させ、一定期間直接高速に失敗させ、その後半開状態プローブを使用して回復を試みます。pybreaker（Python）などのライブラリを使用できます。
エージェントレベルのレート制限: 特定のクライアントまたは総リクエストボリュームに対して、サーバーのエントリポイントでグローバルレート制限を実装します。
フォールバック: コアサービスが利用できない場合に、機能が低下した代替手段を提供します。例えば、リアルタイム会議室クエリサービスがダウンしている場合、book_conference_roomツールはタイムアウトを待つ代わりに、「サービス一時的に利用不可、後で再度お試しください」のような静的応答を返すことにフォールバックできます。

7. セキュリティ設計とリスクコントロール

セキュリティ制御は、MCPサーバー開発者が実装する必要がある部分に焦点を当てます。MCPプロトコル自体は認証や認可のメカニズムを提供しません。すべてのセキュリティ検証ロジックは、アプリケーション層でクライアントとサーバーによって実装される必要があります。

7.1 MCPサーバーのセキュリティ脅威モデル

具体的な脅威の例:

不正な呼び出し: 「クエリ」権限のみを持つクライアントが、リクエストを構築することで「削除」ツールを正常に呼び出す。または、「部署AのデータAPI」へのアクセスが許可されたツールが、クライアントによって悪用され「部署Bのデータ」へのアクセスを試みられる。
データ漏洩: ツールがエラー応答やログで誤って機密情報（例：テーブル構造やSQLステートメントを含むデータベースエラー詳細）を返す。
コンテキスト注入リスク: クライアントからのContextデータを検証せずにデータベースクエリやコマンド実行に盲目的に信頼すると、SQLインジェクションやコマンドインジェクションにつながる可能性があります。

7.2 権限とアクセス制御の実践

「三層の権限検証」モデルを実装します。

トランスポート層: 誰がサーバーに接続できるか？
- Stdio: 通常、OSのプロセス権限によって制御されます。クライアントはサーバーサブプロセスを起動する権限が必要です。
- SSE/HTTP: 暗号化通信のためにTLS (HTTPS) を使用します。ネットワークレベルのファイアウォール、APIゲートウェイ、または有効なクライアント証明書 (mTLS) を要求して接続を制限します。
ツール呼び出し層: 現在のユーザー/IDはこのツールを呼び出せるか？
- リクエストのContextからIDトークン（JWTなど）を抽出します。
- トークンの署名と有効期限を検証します。
- トークン内のロールまたは権限クレームに基づいて、book_conference_roomツールの呼び出しが許可されているかを判断します。シンプルな「ロール-ツール」マッピングテーブルを維持できます。
データ層: このIDはターゲットリソースにアクセスできるか？
- book_conference_roomビジネスロジック内で、きめ細かな検証を実行します。
- 例えば、ユーザーが予約ツールを呼び出すことが許可されていても、そのユーザーが部屋の使用を許可された部署に属しているか、または予約時間が制限を超えていないかを確認します。
- これには、外部のユーザーディレクトリや権限システムへのクエリが必要です。

最小特権の原則: 各ツールは、その機能を完了するために必要な最小限の権限のみを持つべきです。例えば、クエリツールは読み取り専用のデータベース権限のみを持つべきです。

7.3 セキュリティインシデント対応戦略

ログフォレンジック: すべての認証、認可の決定、ツール呼び出し（パラメータを含む）、および主要なビジネス操作について、構造化されたログを確保し、セキュアなログプラットフォーム（例：ELK Stack、Splunk）に一元的に収集します。
迅速なツール無効化: ツールに重大な脆弱性が発見された場合、設定スイッチを介して、またはサーバー全体をオフラインにすることなく新しいバージョンをリリースすることで、そのツールを迅速に無効化できる機能を持つべきです。
ロールバック戦略: 新しいサーバーバージョンをリリースする際、セキュリティまたは機能上の問題が発生した場合に迅速にロールバックできるように、以前のDockerイメージまたはデプロイパッケージを準備しておく必要があります。

8. よくある質問とトラブルシューティングガイド

これらの問題は、MCPプロトコル層またはサーバー実装層のいずれかで典型的です。

8.1 MCPサーバー起動失敗チェックリスト

現象	考えられる原因	チェックリスト
プロセスがすぐに終了する	Pythonの構文エラー、依存関係の欠落	1. ターミナルで`python your_server.py`を直接実行し、エラー出力を確認します。 2. `pip list`を実行して依存関係を確認します。
MCP Inspectorへの接続失敗	不正確なトランスポート方式またはパス設定	1. Inspectorで選択したトランスポート方式（Stdio/SSE）がサーバーコードと一致していることを確認します。 2. Stdioの場合、「Command」と「Arguments」が正しいPythonインタープリタとスクリプトパスを指していることを確認します。
クライアントが`Tool not found`を報告する	ツールが正しく登録されていないか、名前の不一致	1. サーバーコードの`list_tools`メソッドの戻り値にターゲットツールが含まれているか確認します。 2. ツールの`name`文字列とクライアント呼び出しで使用されている名前が一致しているか比較します（大文字と小文字を区別します）。

8.2 パフォーマンス異常診断プロセス

遅いリクエストの特定:
- ツールハンドラ関数の開始時と終了時にタイムスタンプを記録します。
- Python cProfileやpy-spyのようなパフォーマンス分析ツールを使用して、どの関数や外部呼び出しが最も時間がかかっているかを特定します。
並行処理ボトルネック分析:
- 監視ツールを使用して、並行リクエストが増加するにつれてシステムのCPU、メモリ、I/O使用量を監視します。
- データベースがボトルネックの場合、遅いクエリログを確認し、インデックスとクエリステートメントを最適化します。
- 非同期環境でイベントループをブロックする可能性のある同期ブロッキング操作（例：非同期ではないデータベースドライバの使用）がないか確認します。

8.3 本番環境でのインシデントレビューと改善

古典的なケース：SQLインジェクション

シナリオ: search_documentsツールがユーザー入力キーワードを受け取り、それを直接SQL文字列に連結してクエリを実行します。
攻撃: ユーザーが”’; DROP TABLE documents; --”を入力します。
結果: データが破損または漏洩します。
根本原因: ツール実装が入力の検証に失敗したか、パラメータ化クエリを使用しなかった。
教訓と改善点:
1. すべてのツール入力パラメータは厳密なスキーマ検証を受ける必要があります（例：Pydanticを使用して型、長さ、範囲を制限する）。
2. データベースにアクセスする際は、常にパラメータ化クエリ（プリペアドステートメント）またはORMを使用し、SQL文字列を連結しないでください。
3. データベース構造などの詳細が漏洩するのを避けるため、エラー応答では汎用的でユーザーフレンドリーなメッセージを返してください。
4. 外部システムとのインタラクションを含むすべてのコードに焦点を当て、セキュアコーディング標準とコードレビューを必須プロセスにします。

9. まとめ：実世界のAIエージェントシステムでMCPサーバーを使用する方法

エージェントアーキテクチャにおけるMCPサーバーの位置づけ: AIエージェント（脳/コーディネーター）とその「手」（ツール）、「目」（データソース）との間の標準化されたセキュアな接続ブリッジとして機能します。エージェントはMCPプロトコルを介して、サーバーが提供する機能を動的に発見し、使用します。
MVPから本番環境への進化パス:
1. フェーズ1（MVP）: 少数のコアツール（例：book_room、query_calendar）を含む単一のMCPサーバーが、エージェント（例：Claude Desktop）と同じ開発マシン上でStdio経由で実行されます。アイデアを迅速に検証します。
2. フェーズ2（開発）: 機能ドメインに基づいて、複数の単一責任MCPサーバーに分割されます。例えば、
  - calendar-server: カレンダーとイベントを管理します。
  - conference-server: 会議室と設備を管理します。
  - user-directory-server: 従業員情報クエリを提供します。
    エージェントは複数のサーバーに同時に接続し、その機能を組み合わせて複雑なタスクを実行できます。
3. フェーズ3（プラットフォーム化）: サービス登録と発見メカニズム（例：Consul、etcd）を導入します。MCPサーバーは起動時に、提供するツールをレジストリに登録します。エージェントはレジストリから利用可能なサーバーを動的に発見し、接続を確立します。さらに、ヘルスチェック、ロードバランシング、統一された監視/アラートを追加します。
さらなる探求のための次のステップ:
- MCP Inspector のようなデバッグツールの高度な機能の利用を深掘りします。
- MCPのネイティブサポートを追加しているLangChainやLlamaIndexのような主要なエージェントフレームワークとMCPサーバーを統合する方法を探求します。
- サーバーがリソースの更新をクライアントに能動的にプッシュする（リソースの変更通知）など、より複雑なパターンを研究します。
- Official Model Context ProtocolとSDKの更新をフォローし、新しいプロトコル機能とベストプラクティスを迅速に採用します。

このガイドを通じて、堅牢で安全かつスケーラブルなMCPサーバーを構築するための核となる知識と実践的なスキルを習得しました。さあ、あなたの社内システムをAIエージェントの世界に安全に接続しましょう。

MCP記事シリーズ:
MCPサーバーの包括的分析：AIエージェント時代のコンテキストとツール通信ハブ
 MCPサーバーはどのような課題を解決するのか？ AIエージェントに不可欠な理由
 MCPサーバーアーキテクチャと動作原理：プロトコルから実行フローまで
 MCPサーバー実践ガイド：ゼロからの構築、テスト、デプロイ
 MCPサーバーのアプリケーションシナリオ評価と技術選定ガイド

著者について

このコンテンツは、NavGoodコンテンツ編集チームによってまとめられ公開されています。

NavGoodは、AIツールとAIアプリケーションエコシステムに焦点を当てたナビゲーションおよびコンテンツプラットフォームであり、AIエージェント、自動化ワークフロー、生成AIの開発と実践的な実装を追跡しています。

免責事項: 本記事は、著者の個人的な理解と実践経験を表すものです。いかなるフレームワーク、組織、または企業の公式見解を代表するものではなく、商業的、金融的、または投資に関する助言を構成するものでもありません。すべての情報は公開された情報源と著者の独立した調査に基づいています。

参考文献:
[1]: https://github.com/modelcontextprotocol/inspector "MCP Inspector"
[2]: https://modelcontextprotocol.io/docs/getting-started/intro "What is the Model Context Protocol (MCP)?"
[3]: https://platform.openai.com/docs/guides/function-calling "Function calling"
[4]: https://docs.python.org/3/library/profile.html "The Python Profilers"
[5]: https://github.com/benfred/py-spy "Sampling profiler for Python programs"
[6]: https://github.com/modelcontextprotocol/python-sdk "The official Python SDK for MCP servers and clients"
[7]: https://github.com/modelcontextprotocol/typescript-sdk "The official TypeScript SDK for MCP servers and clients"
[8]: https://json-rpc.org/specification "JSON-RPC 2.0 Specification"
[9]: https://etcd.io/ "A distributed, reliable key-value store for the most critical data of a distributed system"
[10]: https://zookeeper.apache.org/ "What is ZooKeeper?"
[11]: https://aws.amazon.com/cn/secrets-manager/ "AWS Secrets Manager"
[12]: https://developer.hashicorp.com/vault "Manage Secrets & Protect Sensitive Data"
[13]: https://azure.microsoft.com/en-us/products/key-vault "Azure Key Vault"