コアアーキテクチャ"

MCP がどのようにクライアント、サーバー、そして LLM を接続するかを理解する

Model Context Protocol (MCP) は、LLM アプリケーションと統合との間でシームレスな通信を可能にする、柔軟かつ拡張性のあるアーキテクチャ上に構築されています。本ドキュメントでは、コアとなるアーキテクチャのコンポーネントと概念について説明します。

概要

MCP はクライアント・サーバーアーキテクチャに従っています。具体的には:

• ホスト: 接続を開始する LLM アプリケーション（例: Claude Desktop や IDE）
• クライアント: ホストアプリケーション内でサーバーと 1:1 の接続を維持する
• サーバー: クライアントにコンテキスト、ツール、プロンプトを提供する

コアコンポーネント

プロトコル層

プロトコル層は、メッセージのフレーミング、リクエスト／レスポンスの連携、および高レベルな通信パターンを処理します。

class Session(BaseSession[RequestT, NotificationT, ResultT]):
   async def send_request(
       self,
       request: RequestT,
       result_type: type[Result]
   ) -> Result:
       """
       リクエストを送信し、レスポンスを待ちます。レスポンスにエラーが含まれている場合は McpError を発生させます。
       """
       # リクエスト処理の実装

   async def send_notification(
       self,
       notification: NotificationT
   ) -> None:
       """レスポンスを期待しない一方向の通知を送信します。"""
       # 通知処理の実装

   async def _received_request(
       self,
       responder: RequestResponder[ReceiveRequestT, ResultT]
   ) -> None:
       """相手側からのリクエストを処理します。"""
       # リクエスト処理の実装

   async def _received_notification(
       self,
       notification: ReceiveNotificationT
   ) -> None:
       """相手側からの通知を処理します。"""
       # 通知処理の実装

主なクラスには以下が含まれます：

• Protocol
• Client
• Server

トランスポート層

トランスポート層は、クライアントとサーバー間の実際の通信を処理します。MCP は複数のトランスポート機構をサポートしています：

1. Stdio トランスポート
   • 標準入力／出力を使用して通信します
   • ローカルプロセス間での通信に最適です
2. HTTP と SSE トランスポート
   • サーバーからクライアントへのメッセージに Server-Sent Events を使用
   • クライアントからサーバーへのメッセージに HTTP POST を使用

すべてのトランスポートは、メッセージ交換に JSON-RPC 2.0 を使用します。Model Context Protocol のメッセージフォーマットの詳細については、仕様 を参照してください。

メッセージタイプ

MCP には主に以下のメッセージタイプがあります：

1. リクエスト: 相手側からのレスポンスを期待します：

   interface Request {
     method: string;
     params?: { ... };
   }
   

2. 結果: リクエストに対する成功レスポンスです：

   interface Result {
     [key: string]: unknown;
   }
   

3. エラー: リクエストが失敗したことを示します：

   interface Error {
     code: number;
     message: string;
     data?: unknown;
   }
   

4. 通知: レスポンスを期待しない一方向のメッセージです：

   interface Notification {
     method: string;
     params?: { ... };
   }
   

接続ライフサイクル

1. 初期化

1. クライアントは、プロトコルバージョンと能力を含む initialize リクエストを送信します
2. サーバーは、そのプロトコルバージョンと能力を含むレスポンスを返します
3. クライアントは、確認のために initialized 通知を送信します
4. 通常のメッセージ交換が開始されます

2. メッセージ交換

初期化後、以下のパターンがサポートされます：

• リクエスト－レスポンス: クライアントまたはサーバーがリクエストを送信し、相手側がレスポンスを返す
• 通知: どちらの当事者も一方向のメッセージを送信可能

3. 終了

どちらの当事者も接続を終了することができます：

• close() を使用したクリーンなシャットダウン
• トランスポートの切断
• エラー条件

エラー処理

MCP は以下の標準エラーコードを定義しています：

enum ErrorCode {
  // 標準 JSON-RPC エラーコード
  ParseError = -32700,
  InvalidRequest = -32600,
  MethodNotFound = -32601,
  InvalidParams = -32602,
  InternalError = -32603
}

SDK やアプリケーションは、-32000 以上の独自エラーコードを定義できます。

エラーは以下の方法で伝播されます：

• リクエストに対するエラーレスポンス
• トランスポート上のエラーイベント
• プロトコルレベルのエラーハンドラ

実装例

以下は MCP サーバーを実装する基本的な例です：

import asyncio
import mcp.types as types
from mcp.server import Server
from mcp.server.stdio import stdio_server

app = Server("example-server")

@app.list_resources()
async def list_resources() -> list[types.Resource]:
   return [
       types.Resource(
           uri="example://resource",
           name="Example Resource"
       )
   ]

async def main():
   async with stdio_server() as streams:
       await app.run(
           streams[0],
           streams[1],
           app.create_initialization_options()
       )

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

ベストプラクティス

トランスポートの選択

1. ローカル通信
   • ローカルプロセス間の通信には stdio トランスポートを使用
   • 同一マシン上の通信に効率的
   • プロセス管理が簡単
2. リモート通信
   • HTTP 互換性が必要なシナリオには SSE を使用
   • 認証や認可などのセキュリティ上の考慮事項を検討

メッセージ処理

1. リクエスト処理
   • 入力を十分に検証する
   • 型安全なスキーマを使用する
   • エラーは適切に処理する
   • タイムアウトを実装する
2. 進捗報告
   • 長時間の操作には進捗トークンを使用する
   • 進捗を段階的に報告する
   • 既知の場合は全体の進捗を含める
3. エラー管理
   • 適切なエラーコードを使用する
   • 有用なエラーメッセージを含める
   • エラー発生時にリソースを適切にクリーンアップする

セキュリティ上の考慮事項

1. トランスポートのセキュリティ
   • リモート接続には TLS を使用する
   • 接続元を検証する
   • 必要に応じて認証を実装する
2. メッセージの検証
   • すべての受信メッセージを検証する
   • 入力をサニタイズする
   • メッセージサイズの制限を確認する
   • JSON-RPC のフォーマットを検証する
3. リソースの保護
   • アクセス制御を実装する
   • リソースパスを検証する
   • リソース使用状況を監視する
   • リクエストのレート制限を行う
4. エラー処理
   • 機密情報を漏洩させない
   • セキュリティに関するエラーをログに記録する
   • 適切なクリーンアップを実施する
   • DoS シナリオに対応する

デバッグとモニタリング

1. ログ記録
   • プロトコルイベントをログに記録する
   • メッセージの流れを追跡する
   • パフォーマンスを監視する
   • エラーを記録する
2. 診断
   • ヘルスチェックを実装する
   • 接続状態を監視する
   • リソース使用状況を追跡する
   • パフォーマンスのプロファイリングを行う
3. テスト
   • 異なるトランスポートをテストする
   • エラー処理を検証する
   • エッジケースを確認する
   • サーバーのロードテストを実施する

リソース

サーバーから LLM にデータやコンテンツを公開する

リソースは、Model Context Protocol (MCP) の中核となるプリミティブであり、サーバーがクライアントに読み取られ、LLM との相互作用のコンテキストとして利用されるデータやコンテンツを公開できるようにします。

概要

リソースは、MCP サーバーがクライアントに提供したいと考えるあらゆる種類のデータを表します。これには以下が含まれます:

• ファイルの内容
• データベースのレコード
• API レスポンス
• ライブシステムデータ
• スクリーンショットや画像
• ログファイル
• その他多数

各リソースは一意の URI によって識別され、テキストまたはバイナリデータのいずれかを含むことができます。

リソース URI

リソースは、以下の形式に従う URI を使用して識別されます:

[protocol]://[host]/[path]

例えば:

• file:///home/user/documents/report.pdf
• postgres://database/customers/schema
• screen://localhost/display1

プロトコルおよびパスの構造は MCP サーバーの実装によって定義され、サーバーは独自のカスタム URI スキームを定義できます。

リソースの種類

リソースは、2 種類のコンテンツを含むことができます:

テキストリソース

テキストリソースは、UTF-8 エンコードされたテキストデータを含みます。以下に適しています:

• ソースコード
• 設定ファイル
• ログファイル
• JSON/XML データ
• プレーンテキスト

バイナリリソース

バイナリリソースは、base64 エンコードされた生のバイナリデータを含みます。以下に適しています:

• 画像
• PDF
• オーディオファイル
• ビデオファイル
• その他テキスト以外のフォーマット

リソースの発見

クライアントは、以下の 2 つの主要な方法で利用可能なリソースを発見できます:

直接リソース

サーバーは、resources/list エンドポイントを通じて具体的なリソースのリストを公開します。各リソースは以下の情報を含みます:

{
  uri: string;           // リソースの一意識別子
  name: string;          // 人間に読みやすい名称
  description?: string;  // オプションの説明
  mimeType?: string;     // オプションの MIME タイプ
}

リソーステンプレート

動的なリソースの場合、サーバーはクライアントが有効なリソース URI を構築するために使用できる URI テンプレート を公開できます:

{
  uriTemplate: string;   // RFC 6570 に準拠した URI テンプレート
  name: string;          // このタイプの人間に読みやすい名称
  description?: string;  // オプションの説明
  mimeType?: string;     // 一致するすべてのリソースに対するオプションの MIME タイプ
}

リソースの読み取り

リソースを読み取るために、クライアントはリソース URI を指定して resources/read リクエストを送信します。

サーバーは、リソースの内容のリストで応答します:

{
  contents: [
    {
      uri: string;        // リソースの URI
      mimeType?: string;  // オプションの MIME タイプ

      // 以下のいずれか:
      text?: string;      // テキストリソースの場合
      blob?: string;      // バイナリリソースの場合 (base64 エンコード)
    }
  ]
}

リソースの更新

MCP は、以下の 2 つのメカニズムを通じてリソースのリアルタイム更新をサポートします:

リストの変更

サーバーは、利用可能なリソースのリストが変更されたときに、notifications/resources/list_changed 通知を通じてクライアントに通知できます。

コンテンツの変更

クライアントは、特定のリソースの更新を購読できます:

1. クライアントはリソース URI を指定して resources/subscribe を送信します
2. サーバーは、リソースが変更された際に notifications/resources/updated を送信します
3. クライアントは resources/read を使用して最新の内容を取得できます
4. クライアントは resources/unsubscribe で購読を解除できます

実装例

以下は、MCP サーバーにおけるリソースサポートの実装例です:

app = Server("example-server")

@app.list_resources()
async def list_resources() -> list[types.Resource]:
   return [
       types.Resource(
           uri="file:///logs/app.log",
           name="Application Logs",
           mimeType="text/plain"
       )
   ]

@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
   if str(uri) == "file:///logs/app.log":
       log_contents = await read_log_file()
       return log_contents

   raise ValueError("Resource not found")

# サーバーの起動
async with stdio_server() as streams:
   await app.run(
       streams[0],
       streams[1],
       app.create_initialization_options()
   )

ベストプラクティス

リソースサポートを実装する際には、以下の点に留意してください:

1. 明確で説明的なリソース名および URI を使用する
2. LLM の理解を促すために有用な説明を含める
3. 知られている場合は適切な MIME タイプを設定する
4. 動的コンテンツにはリソーステンプレートを実装する
5. 頻繁に変更されるリソースには購読機能を使用する
6. 明確なエラーメッセージでエラーを適切に処理する
7. 大規模なリソースリストにはページネーションを検討する
8. 適切な場合はリソースの内容をキャッシュする
9. 処理前に URI の検証を行う
10. カスタム URI スキームについてドキュメント化する

セキュリティ上の考慮事項

リソースを公開する際には、以下の点に注意してください:

• すべてのリソース URI を検証する
• 適切なアクセス制御を実装する
• ディレクトリトラバーサルを防止するためにファイルパスをサニタイズする
• バイナリデータの取り扱いには注意する
• リソースの読み取りに対してレート制限を検討する
• リソースアクセスを監査する
• 転送中の機密データを暗号化する
• MIME タイプを検証する
• 長時間実行される読み取りにタイムアウトを実装する
• リソースのクリーンアップを適切に処理する

プロンプト

再利用可能なプロンプトテンプレートとワークフローを作成する

プロンプトは、サーバーが再利用可能なプロンプトテンプレートやワークフローを定義し、クライアントがユーザーや LLM に対して容易に提示できるようにするものです。これにより、一般的な LLM とのやり取りを標準化し、共有する強力な方法が提供されます。

概要

MCP におけるプロンプトは、事前に定義されたテンプレートであり、以下の機能を提供します:

• 動的な引数の受け入れ
• リソースからのコンテキストの組み込み
• 複数のやり取りのチェーン
• 特定のワークフローの指針付け
• UI 要素としての提示（例: スラッシュコマンド）

プロンプトの構造

各プロンプトは以下の形式で定義されます:

{
  name: string;              // プロンプトの一意識別子
  description?: string;      // 人間に読みやすい説明
  arguments?: [              // オプションの引数リスト
    {
      name: string;          // 引数の識別子
      description?: string;  // 引数の説明
      required?: boolean;    // 引数が必須かどうか
    }
  ]
}

プロンプトの発見

クライアントは、prompts/list エンドポイントを通じて利用可能なプロンプトを発見できます:

// リクエスト
{
  method: "prompts/list"
}

// レスポンス
{
  prompts: [
    {
      name: "analyze-code",
      description: "コードを解析して潜在的な改善点を見つける",
      arguments: [
        {
          name: "language",
          description: "プログラミング言語",
          required: true
        }
      ]
    }
  ]
}

プロンプトの利用

プロンプトを利用するために、クライアントは prompts/get リクエストを送信します:

// リクエスト
{
  method: "prompts/get",
  params: {
    name: "analyze-code",
    arguments: {
      language: "python"
    }
  }
}

// レスポンス
{
  description: "Pythonコードを解析して潜在的な改善点を見つける",
  messages: [
    {
      role: "user",
      content: {
        type: "text",
        text: "以下のPythonコードを改善の余地がないか分析してください。:\n\n```python\ndef calculate_sum(numbers):\n    total = 0\n    for num in numbers:\n        total = total + num\n    return total\n\nresult = calculate_sum([1, 2, 3, 4, 5])\nprint(result)\n```"
      }
    }
  ]
}

動的プロンプト

プロンプトは動的であり、以下を含むことができます:

組み込みリソースコンテキスト

{
  "name": "analyze-project",
  "description": "プロジェクトのログとコードを解析する",
  "arguments": [
    {
      "name": "timeframe",
      "description": "解析するログの期間",
      "required": true
    },
    {
      "name": "fileUri",
      "description": "レビューするコードのファイルのURI",
      "required": true
    }
  ]
}

prompts/get リクエストの処理例:

{
  "messages": [
    {
      "role": "user",
      "content": {
        "type": "text",
        "text": "これらのシステムログとコードファイルを分析し、問題がないか確認してください:"
      }
    },
    {
      "role": "user",
      "content": {
        "type": "resource",
        "resource": {
          "uri": "logs://recent?timeframe=1h",
          "text": "[2024-03-14 15:32:11] ERROR: Connection timeout in network.py:127\n[2024-03-14 15:32:15] WARN: Retrying connection (attempt 2/3)\n[2024-03-14 15:32:20] ERROR: Max retries exceeded",
          "mimeType": "text/plain"
        }
      }
    },
    {
      "role": "user",
      "content": {
        "type": "resource",
        "resource": {
          "uri": "file:///path/to/code.py",
          "text": "def connect_to_service(timeout=30):\n    retries = 3\n    for attempt in range(retries):\n        try:\n            return establish_connection(timeout)\n        except TimeoutError:\n            if attempt == retries - 1:\n                raise\n            time.sleep(5)\n\ndef establish_connection(timeout):\n    # Connection implementation\n    pass",
          "mimeType": "text/x-python"
        }
      }
    }
  ]
}

マルチステップワークフロー

const debugWorkflow = {
  name: "debug-error",
  async getMessages(error: string) {
    return [
      {
        role: "user",
        content: {
          type: "text",
          text: `以下が私が見つけたエラーです: ${error}`
        }
      },
      {
        role: "assistant",
        content: {
          type: "text",
          text: "このエラーの解析をお手伝いします。これまでに何を試しましたか？"
        }
      },
      {
        role: "user",
        content: {
          type: "text",
          text: "サービスを再起動してみましたが、エラーは依然として発生しています。"
        }
      }
    ];
  }
};

実装例

以下は、MCP サーバーにおけるプロンプトサポートの完全な実装例です:

from mcp.server import Server
import mcp.types as types

# 定義されたプロンプト
PROMPTS = {
   "git-commit": types.Prompt(
       name="git-commit",
       description="Gitのコミットメッセージを生成する",
       arguments=[
           types.PromptArgument(
               name="changes",
               description="Git diff または変更内容の説明",
               required=True
           )
       ],
   ),
   "explain-code": types.Prompt(
       name="explain-code",
       description="コードの仕組みを説明する",
       arguments=[
           types.PromptArgument(
               name="code",
               description="説明すべきコード",
               required=True
           ),
           types.PromptArgument(
               name="language",
               description="プログラミング言語",
               required=False
           )
       ],
   )
}

# サーバーの初期化
app = Server("example-prompts-server")

@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
   return list(PROMPTS.values())

@app.get_prompt()
async def get_prompt(
   name: str, arguments: dict[str, str] | None = None
) -> types.GetPromptResult:
   if name not in PROMPTS:
       raise ValueError(f"Prompt not found: {name}")

   if name == "git-commit":
       changes = arguments.get("changes") if arguments else ""
       return types.GetPromptResult(
           messages=[
               types.PromptMessage(
                   role="user",
                   content=types.TextContent(
                       type="text",
                       text=f"これらの変更について、簡潔かつ説明的なコミットメッセージを作成してください。\n\n{changes}"
                   )
               )
           ]
       )

   if name == "explain-code":
       code = arguments.get("code") if arguments else ""
       language = arguments.get("language", "Unknown") if arguments else "Unknown"
       return types.GetPromptResult(
           messages=[
               types.PromptMessage(
                   role="user",
                   content=types.TextContent(
                       type="text",
                       text=f"この{language}コードの仕組みを説明してください。:\n\n{code}"
                   )
               )
           ]
       )

   raise ValueError("Prompt implementation not found")

ベストプラクティス

プロンプトを実装する際には以下を考慮してください:

1. 明確で説明的なプロンプト名を使用する
2. プロンプトおよび引数の詳細な説明を提供する
3. 必須の引数をすべて検証する
4. 引数が不足している場合は適切に処理する
5. プロンプトテンプレートのバージョン管理を検討する
6. 適切な場合は動的コンテンツをキャッシュする
7. エラー処理を実装する
8. 期待される引数のフォーマットを文書化する
9. プロンプトの組み合わせ可能性を検討する
10. 様々な入力でプロンプトをテストする

UI 統合

プロンプトはクライアントの UI に以下のように提示できます:

• スラッシュコマンド
• クイックアクション
• コンテキストメニュー項目
• コマンドパレットのエントリ
• ガイド付きワークフロー
• インタラクティブフォーム

更新と変更

サーバーは、以下の方法でプロンプトの変更をクライアントに通知できます:

1. サーバーの機能: prompts.listChanged
2. 通知: notifications/prompts/list_changed
3. クライアントがプロンプト一覧を再取得する

セキュリティ上の考慮事項

プロンプトを実装する際には、以下に注意してください:

• すべての引数を検証する
• ユーザー入力をサニタイズする
• レート制限を検討する
• アクセス制御を実装する
• プロンプトの利用状況を監査する
• 機微なデータを適切に取り扱う
• 生成されたコンテンツを検証する
• タイムアウトを実装する
• プロンプトインジェクションのリスクを考慮する
• セキュリティ要件を文書化する

ツール

サーバーを通じて LLM にアクションを実行させる

ツールは、Model Context Protocol (MCP) の強力なプリミティブであり、サーバーがクライアントに実行可能な機能を公開できるようにします。ツールを通じて、LLM は外部システムと連携したり、計算を実行したり、現実世界でアクションを起こすことが可能になります。

:message
ツールは モデル制御 を前提として設計されており、ツールはサーバーからクライアントに公開され、AI モデルが自動的に呼び出すことを意図しています（ただし、承認のために人間が介在する場合もあります）。
:::

概要

MCP におけるツールは、サーバーがクライアントに対して実行可能な関数を公開し、それを LLM がアクションの実行に利用できるようにするものです。ツールの主な特徴は以下の通りです:

• 発見: クライアントは tools/list エンドポイントを通じて利用可能なツールを一覧できます
• 呼び出し: ツールは tools/call エンドポイントを使用して呼び出され、サーバーは要求された操作を実行して結果を返します
• 柔軟性: ツールは単純な計算から複雑な API 連携まで、さまざまな操作をカバーできます

resources と同様に、ツールは一意の名前で識別され、使用方法をガイドするための説明が含まれる場合があります。しかし、リソースと異なり、ツールは状態を変更したり、外部システムと連携する動的な操作を表します。

ツール定義の構造

各ツールは以下の構造で定義されます:

{
  name: string;          // ツールの一意識別子
  description?: string;  // 人間に読みやすい説明
  inputSchema: {         // ツールのパラメータに対する JSON Schema
    type: "object",
    properties: { ... }  // ツール固有のパラメータ
  }
}

ツールの実装

以下は、MCP サーバーにおける基本的なツールの実装例です:

app = Server("example-server")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
   return [
       types.Tool(
           name="calculate_sum",
           description="Add two numbers together",
           inputSchema={
               "type": "object",
               "properties": {
                   "a": {"type": "number"},
                   "b": {"type": "number"}
               },
               "required": ["a", "b"]
           }
       )
   ]

@app.call_tool()
async def call_tool(
   name: str,
   arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
   if name == "calculate_sum":
       a = arguments["a"]
       b = arguments["b"]
       result = a + b
       return [types.TextContent(type="text", text=str(result))]
   raise ValueError(f"Tool not found: {name}")

ツールの使用例パターン

サーバーが提供可能なツールの例をいくつか示します:

システム操作

ローカルシステムと連携するツール:

{
  name: "execute_command",
  description: "Run a shell command",
  inputSchema: {
    type: "object",
    properties: {
      command: { type: "string" },
      args: { type: "array", items: { type: "string" } }
    }
  }
}

API 連携

外部 API をラップするツール:

{
  name: "github_create_issue",
  description: "Create a GitHub issue",
  inputSchema: {
    type: "object",
    properties: {
      title: { type: "string" },
      body: { type: "string" },
      labels: { type: "array", items: { type: "string" } }
    }
  }
}

データ処理

データを変換または解析するツール:

{
  name: "analyze_csv",
  description: "Analyze a CSV file",
  inputSchema: {
    type: "object",
    properties: {
      filepath: { type: "string" },
      operations: {
        type: "array",
        items: {
          enum: ["sum", "average", "count"]
        }
      }
    }
  }
}

ベストプラクティス

ツールを実装する際には以下を考慮してください:

1. 明確で説明的な名前と説明を提供する
2. パラメータのために詳細な JSON Schema 定義を使用する
3. モデルがどのように利用するかを示す例をツールの説明に含める
4. 適切なエラー処理と検証を実装する
5. 長時間の操作には進捗報告を実装する
6. ツールの操作は集中化され、原子的に保つ
7. 期待される戻り値の構造を文書化する
8. 適切なタイムアウトを実装する
9. リソースを多く消費する操作にはレート制限を検討する
10. デバッグやモニタリングのためにツール使用状況をログに記録する

セキュリティ上の考慮事項

ツールを公開する際には:

入力検証

• すべてのパラメータをスキーマに対して検証する
• ファイルパスやシステムコマンドをサニタイズする
• URL や外部識別子を検証する
• パラメータのサイズや範囲を確認する
• コマンドインジェクションを防止する

アクセス制御

• 必要に応じて認証を実装する
• 適切な認可チェックを行う
• ツールの利用状況を監査する
• リクエストに対してレート制限を行う
• 悪用を監視する

エラー処理

• 内部エラーをクライアントに公開しない
• セキュリティに関連するエラーをログに記録する
• タイムアウトを適切に処理する
• エラー発生後にリソースをクリーンアップする
• 戻り値の検証を行う

ツールの発見と更新

MCP は動的なツール発見をサポートしています:

1. クライアントはいつでも利用可能なツールを一覧できます
2. サーバーは notifications/tools/list_changed を使用して、ツールの変更をクライアントに通知できます
3. ツールはランタイム中に追加または削除できます
4. ツール定義は更新可能ですが、慎重に行う必要があります

エラー処理

ツールのエラーは、MCP プロトコルレベルのエラーではなく、結果オブジェクト内で報告されるべきです。これにより、LLM はエラーを確認し、適切な対応（修正や人間による介入の要求など）を行うことができます。ツールでエラーが発生した場合は:

1. 結果オブジェクト内の isError を true に設定する
2. content 配列にエラーの詳細を含める

以下は、ツールの適切なエラー処理の例です:

try:
   # ツールの操作
   result = perform_operation()
   return types.CallToolResult(
       content=[
           types.TextContent(
               type="text",
               text=f"Operation successful: {result}"
           )
       ]
   )
except Exception as error:
   return types.CallToolResult(
       isError=True,
       content=[
           types.TextContent(
               type="text",
               text=f"Error: {str(error)}"
           )
       ]
   )

この方法により、LLM はエラーが発生したことを認識し、必要に応じて修正行動や人間の介入を求めることができます。

ツールのテスト

MCP ツールの包括的なテスト戦略は、以下をカバーすべきです:

• 機能テスト: 有効な入力に対してツールが正しく実行され、無効な入力に対して適切に処理されることを確認する
• 統合テスト: 実際およびモックされた依存関係を使用して、外部システムとのツールの連携をテストする
• セキュリティテスト: 認証、認可、入力のサニタイズ、およびレート制限を検証する
• パフォーマンステスト: 負荷下での動作、タイムアウトの処理、リソースのクリーンアップを確認する
• エラー処理: ツールが MCP プロトコルを通じて適切にエラーを報告し、リソースをクリーンアップすることを確認する

サンプリング

サーバーが LLM に補完リクエストを送る

サンプリングは、サーバーがクライアントを通じて LLM の補完をリクエストできる強力な MCP 機能であり、セキュリティとプライバシーを保ちながら高度なエージェント的動作を実現します。

サンプリングの仕組み

サンプリングのフローは以下のステップに従います:

1. サーバーがクライアントに対して sampling/createMessage リクエストを送信する
2. クライアントはリクエストを確認し、必要に応じて修正する
3. クライアントが LLM から補完をサンプリングする
4. クライアントが補完内容を確認する
5. クライアントが結果をサーバーに返す

この「ヒューマン・イン・ザ・ループ」設計により、ユーザーは LLM が閲覧および生成する内容を制御できます。

メッセージフォーマット

サンプリングリクエストは、標準化されたメッセージフォーマットを使用します:

{
  messages: [
    {
      role: "user" | "assistant",
      content: {
        type: "text" | "image",

        // テキストの場合:
        text?: string,

        // 画像の場合:
        data?: string,             // base64 エンコードされたデータ
        mimeType?: string
      }
    }
  ],
  modelPreferences?: {
    hints?: [{
      name?: string                // 推奨されるモデル名／ファミリー
    }],
    costPriority?: number,         // 0-1, コスト最小化の重要度
    speedPriority?: number,        // 0-1, レイテンシ低減の重要度
    intelligencePriority?: number  // 0-1, 高度な能力の重要度
  },
  systemPrompt?: string,
  includeContext?: "none" | "thisServer" | "allServers",
  temperature?: number,
  maxTokens: number,
  stopSequences?: string[],
  metadata?: Record<string, unknown>
}

リクエストパラメータ

メッセージ

messages 配列には、LLM に送信する会話履歴が含まれます。各メッセージは以下の要素を持ちます:

• role: "user" または "assistant"
• content: メッセージの内容。これは以下のいずれかです:
  • text フィールドを持つテキストコンテンツ
  • data (base64) および mimeType フィールドを持つ画像コンテンツ

モデルの選好

modelPreferences オブジェクトにより、サーバーは利用可能なモデルから適切なモデルを選択するための指針を指定できます:

• hints: クライアントが適切なモデルを選択するためのモデル名の推奨リスト:
  • name: 完全または部分的なモデル名にマッチする文字列 (例: "claude-3", "sonnet")
  • 複数のヒントが優先順に評価されます
• 優先度 (0-1 の正規化値):
  • costPriority: コスト最小化の重要度
  • speedPriority: 低レイテンシ応答の重要度
  • intelligencePriority: 高度なモデル能力の重要度

クライアントは、これらの選好と利用可能なモデルに基づいて最終的なモデル選択を行います。

システムプロンプト

オプションの systemPrompt フィールドにより、サーバーは特定のシステムプロンプトをリクエストできます。クライアントはこれを修正または無視することができます。

コンテキストの包含

includeContext パラメータは、どの MCP コンテキストを含むかを指定します:

• "none": 追加のコンテキストは含まない
• "thisServer": リクエスト元サーバーのコンテキストを含む
• "allServers": 接続されているすべての MCP サーバーのコンテキストを含む

クライアントは実際にどのコンテキストを含むかを制御します。

サンプリングパラメータ

LLM のサンプリングを微調整するために、以下のパラメータを指定できます:

• temperature: ランダム性を制御 (0.0 から 1.0)
• maxTokens: 生成する最大トークン数
• stopSequences: 補完の停止を示すシーケンスの配列
• metadata: その他のプロバイダー固有の追加パラメータ

レスポンスフォーマット

クライアントは、以下の形式で補完結果を返します:

{
  model: string,  // 使用されたモデルの名前
  stopReason?: "endTurn" | "stopSequence" | "maxTokens" | string,
  role: "user" | "assistant",
  content: {
    type: "text" | "image",
    text?: string,
    data?: string,
    mimeType?: string
  }
}

リクエスト例

以下は、クライアントに対してサンプリングをリクエストする例です:

{
  "method": "sampling/createMessage",
  "params": {
    "messages": [
      {
        "role": "user",
        "content": {
          "type": "text",
          "text": "What files are in the current directory?"
        }
      }
    ],
    "systemPrompt": "You are a helpful file system assistant.",
    "includeContext": "thisServer",
    "maxTokens": 100
  }
}

ベストプラクティス

サンプリングを実装する際には以下を考慮してください:

1. 明確で構造化されたプロンプトを必ず提供する
2. テキストと画像の両方のコンテンツを適切に扱う
3. 適切なトークン制限を設定する
4. includeContext を通じて関連するコンテキストを含める
5. 応答を使用する前に必ず検証する
6. エラーを適切に処理する
7. サンプリングリクエストに対してレート制限を考慮する
8. 期待されるサンプリング動作を文書化する
9. 様々なモデルパラメータでテストする
10. サンプリングコストを監視する

ヒューマン・イン・ザ・ループ制御

サンプリングは人間による監視を前提としています:

プロンプトの場合

• クライアントはユーザーに提案されたプロンプトを表示すべきです
• ユーザーはプロンプトを修正または拒否できるべきです
• システムプロンプトはフィルタリングまたは修正可能です
• コンテキストの包含はクライアント側で制御されます

補完の場合

• クライアントはユーザーに補完結果を表示すべきです
• ユーザーは補完結果を修正または拒否できるべきです
• クライアントは補完結果をフィルタリングまたは修正可能です
• 使用するモデルはユーザーが制御します

セキュリティ上の考慮事項

サンプリングを実装する際には、以下に注意してください:

• すべてのメッセージコンテンツを検証する
• 機密情報をサニタイズする
• 適切なレート制限を実装する
• サンプリングの使用状況を監視する
• 転送中のデータを暗号化する
• ユーザーデータのプライバシーを確保する
• サンプリングリクエストを監査する
• コストの露出を制御する
• タイムアウトを実装する
• モデルエラーを適切に処理する

一般的なパターン

エージェント的ワークフロー

サンプリングは、以下のようなエージェント的なパターンを可能にします:

• リソースの読み取りと解析
• コンテキストに基づいた意思決定
• 構造化データの生成
• マルチステップタスクの処理
• インタラクティブな支援の提供

コンテキスト管理

コンテキスト管理のベストプラクティス:

• 必要最小限のコンテキストをリクエストする
• コンテキストを明確に構造化する
• コンテキストサイズの制限に対処する
• 必要に応じてコンテキストを更新する
• 古いコンテキストをクリーンアップする

エラー処理

強固なエラー処理では以下を行うべきです:

• サンプリングの失敗をキャッチする
• タイムアウトエラーを処理する
• レート制限を管理する
• 応答を検証する
• フォールバック動作を提供する
• エラーを適切にログに記録する

制限事項

以下の制限事項に注意してください:

• サンプリングはクライアントの機能に依存します
• ユーザーがサンプリングの動作を制御します
• コンテキストサイズには制限があります
• レート制限が適用される場合があります
• コストに留意する必要があります
• モデルの利用可能性は変動します
• 応答時間は状況により異なります
• すべてのコンテンツタイプがサポートされるわけではありません

ルート

MCP におけるルートの理解

ルート は、サーバーが動作できる境界を定義する MCP の概念です。これにより、クライアントはサーバーに対して、関連するリソースやその位置について情報を提供することができます。

ルートとは？

ルートは、クライアントがサーバーに注目すべき URI を提案するものです。クライアントがサーバーに接続する際、どのルートでサーバーが作業すべきかを宣言します。主にファイルシステムのパスに使用されますが、ルートは HTTP URL を含む任意の有効な URI である可能性があります。

例えば、ルートは次のようになります:

file:///home/user/projects/myapp
https://api.example.com/v1

なぜ ルート を使用するのか？

ルート はいくつかの重要な目的を果たします:

1. ガイダンス: ルートは、サーバーに関連するリソースや位置についての情報を提供します
2. 明確性: ルートにより、どのリソースがワークスペースの一部であるかが明確になります
3. 整理: 複数のルートを使用することで、異なるリソースを同時に扱うことができます

ルート の仕組み

クライアントがルートをサポートする場合、以下のことを行います:

1. 接続時に roots 機能を宣言する
2. サーバーに対して、提案されたルートのリストを提供する
3. （サポートされている場合）ルートが変更された際にサーバーに通知する

ルートは情報提供的なものであり、厳密に強制されるものではありませんが、サーバーは以下のことを行うべきです:

1. 提供されたルートを尊重する
2. ルート URI を使用してリソースを特定し、アクセスする
3. ルートの境界内での操作を優先する

一般的な使用例

ルートは一般的に以下を定義するために使用されます:

• プロジェクトディレクトリ
• リポジトリの場所
• API エンドポイント
• 設定の場所
• リソースの境界

ベストプラクティス

ルート を使用する際には、以下の点に注意してください:

1. 必要なリソースのみを提案する
2. ルートに対して明確で説明的な名前を使用する
3. ルートのアクセス性を監視する
4. ルートの変更を適切に処理する

例

以下は、典型的な MCP クライアントがルートを公開する方法の例です:

{
  "roots": [
    {
      "uri": "file:///home/user/projects/frontend",
      "name": "Frontend Repository"
    },
    {
      "uri": "https://api.example.com/v1",
      "name": "API Endpoint"
    }
  ]
}

この構成は、サーバーがローカルのリポジトリと API エンドポイントの両方に注目するよう提案し、論理的に分離されていることを示しています。

トランスポート

MCP の通信メカニズムについて学ぶ

Model Context Protocol (MCP) におけるトランスポートは、クライアントとサーバー間の通信の基盤を提供します。トランスポートは、メッセージの送受信の基本的な仕組みを処理します。

メッセージフォーマット

MCP は、そのワイヤフォーマットとして JSON-RPC 2.0 を使用します。トランスポート層は、MCP プロトコルメッセージを送信のために JSON-RPC フォーマットに変換し、受信した JSON-RPC メッセージを MCP プロトコルメッセージに戻す役割を担います。

使用される JSON-RPC メッセージのタイプは以下の 3 種類です:

リクエスト

{
  jsonrpc: "2.0",
  id: number | string,
  method: string,
  params?: object
}

レスポンス

{
  jsonrpc: "2.0",
  id: number | string,
  result?: object,
  error?: {
    code: number,
    message: string,
    data?: unknown
  }
}

通知

{
  jsonrpc: "2.0",
  method: string,
  params?: object
}

標準トランスポートタイプ

MCP には、2 つの標準トランスポート実装が含まれています:

標準入力/出力 (stdio)

stdio トランスポートは、標準入力および出力ストリームを通じた通信を可能にします。これは、ローカル統合やコマンドラインツールに特に有用です。

stdio を使用する場面:

• コマンドラインツールの構築
• ローカル統合の実装
• シンプルなプロセス間通信が必要な場合
• シェルスクリプトとの連携

サーバー

app = Server("example-server")

async with stdio_server() as streams:
   await app.run(
       streams[0],
       streams[1],
       app.create_initialization_options()
   )

クライアント

params = StdioServerParameters(
   command="./server",
   args=["--option", "value"]
)

async with stdio_client(params) as streams:
   async with ClientSession(streams[0], streams[1]) as session:
       await session.initialize()

Server-Sent Events (SSE)

SSE トランスポートは、HTTP POST リクエストを使用したクライアントからサーバーへの通信と、サーバーからクライアントへのストリーミングを可能にします。

SSE を使用する場面:

• サーバーからクライアントへのストリーミングのみが必要な場合
• 制限されたネットワーク環境での利用
• シンプルな更新通知の実装

サーバー

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

app = Server("example-server")
sse = SseServerTransport("/messages")

async def handle_sse(scope, receive, send):
   async with sse.connect_sse(scope, receive, send) as streams:
       await app.run(streams[0], streams[1], app.create_initialization_options())

async def handle_messages(scope, receive, send):
   await sse.handle_post_message(scope, receive, send)

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

クライアント

async with sse_client("http://localhost:8000/sse") as streams:
   async with ClientSession(streams[0], streams[1]) as session:
       await session.initialize()

カスタムトランスポート

MCP は、特定のニーズに合わせたカスタムトランスポートの実装を容易にします。任意のトランスポート実装は、Transport インターフェースに準拠する必要があります:

カスタムトランスポートは以下の用途に利用できます:

• カスタムネットワークプロトコル
• 専用の通信チャネル
• 既存システムとの統合
• パフォーマンス最適化

注意: MCP サーバーは通常 asyncio を用いて実装されますが、より広い互換性を得るために、トランスポートなどの低レベルインターフェースは anyio を使用して実装することを推奨します。

@contextmanager
async def create_transport(
   read_stream: MemoryObjectReceiveStream[JSONRPCMessage | Exception],
   write_stream: MemoryObjectSendStream[JSONRPCMessage]
):
   """
   MCP のトランスポートインターフェース。

   Args:
       read_stream: 受信メッセージを読み取るためのストリーム
       write_stream: 送信メッセージを書き込むためのストリーム
   """
   async with anyio.create_task_group() as tg:
       try:
           # メッセージ処理の開始
           tg.start_soon(lambda: process_messages(read_stream))

           # メッセージの送信
           async with write_stream:
               yield write_stream

       except Exception as exc:
           # エラー処理
           raise exc
       finally:
           # クリーンアップ
           tg.cancel_scope.cancel()
           await write_stream.aclose()
           await read_stream.aclose()

エラー処理

トランスポート実装は、以下のエラーシナリオを処理する必要があります:

1. 接続エラー
2. メッセージ解析エラー
3. プロトコルエラー
4. ネットワークタイムアウト
5. リソースのクリーンアップ

エラー処理の例:

注意: MCP サーバーは通常 asyncio を用いて実装されますが、より広い互換性を得るために、トランスポートなどの低レベルインターフェースは anyio を使用して実装することを推奨します。

@contextmanager
async def example_transport(scope: Scope, receive: Receive, send: Send):
   try:
       # 双方向通信のためのストリームを作成
       read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
       write_stream, write_stream_reader = anyio.create_memory_object_stream(0)

       async def message_handler():
           try:
               async with read_stream_writer:
                   # メッセージ処理ロジック
                   pass
           except Exception as exc:
               logger.error(f"Failed to handle message: {exc}")
               raise exc

       async with anyio.create_task_group() as tg:
           tg.start_soon(message_handler)
           try:
               # 通信のためにストリームを yield
               yield read_stream, write_stream
           except Exception as exc:
               logger.error(f"Transport error: {exc}")
               raise exc
           finally:
               tg.cancel_scope.cancel()
               await write_stream.aclose()
               await read_stream.aclose()
   except Exception as exc:
       logger.error(f"Failed to initialize transport: {exc}")
       raise exc

ベストプラクティス

MCP トランスポートの実装または利用においては、以下を考慮してください:

1. 接続ライフサイクルを適切に処理する
2. 適切なエラー処理を実装する
3. 接続終了時にリソースをクリーンアップする
4. 適切なタイムアウトを使用する
5. 送信前にメッセージを検証する
6. デバッグのためにトランスポートイベントをログに記録する
7. 必要に応じて再接続ロジックを実装する
8. メッセージキューのバックプレッシャーに対応する
9. 接続の健全性を監視する
10. 適切なセキュリティ対策を実装する

セキュリティ上の考慮事項

トランスポート実装においては、以下を考慮してください:

認証と認可

• 適切な認証機構を実装する
• クライアントの資格情報を検証する
• セキュアなトークン管理を行う
• 認可チェックを実装する

データセキュリティ

• ネットワークトランスポートには TLS を使用する
• 機密データを暗号化する
• メッセージの整合性を検証する
• メッセージサイズの制限を実装する
• 入力データをサニタイズする

ネットワークセキュリティ

• レート制限を実装する
• 適切なタイムアウトを設定する
• サービス拒否攻撃 (DoS) に対応する
• 異常なパターンを監視する
• 適切なファイアウォールルールを実装する

デバッグトランスポート

トランスポートの問題をデバッグするためのヒント:

1. デバッグログを有効にする
2. メッセージの流れを監視する
3. 接続状態を確認する
4. メッセージフォーマットを検証する
5. エラーシナリオをテストする
6. ネットワーク解析ツールを使用する
7. ヘルスチェックを実装する
8. リソース使用状況を監視する
9. エッジケースをテストする
10. 適切なエラー追跡を実装する

APIのページはこちら

https://www.weather.gov/documentation/services-web-api

まず警報。2文字の州コードを指定する。以下はテキサス州（TX）の場合で、警報はリストで複数入っているので1つだけ抜粋した。

curl https://api.weather.gov/alerts/active/area/TX
    -H "User-Agent: weather-app/1.0" \
    -H "Accept: application/geo+json"  | jq -r .features[0].properties

{
  "@id": "https://api.weather.gov/alerts/urn:oid:2.49.0.1.840.0.f6e51155e2067acf2bf0d9b755c6384ae0cc3558.001.1",
  "@type": "wx:Alert",
  "id": "urn:oid:2.49.0.1.840.0.f6e51155e2067acf2bf0d9b755c6384ae0cc3558.001.1",
  "areaDesc": "Archer; Clay",
  "geocode": {
    "SAME": [
      "048009",
      "048077"
    ],
    "UGC": [
      "TXZ089",
      "TXZ090"
    ]
  },
  "affectedZones": [
    "https://api.weather.gov/zones/forecast/TXZ089",
    "https://api.weather.gov/zones/forecast/TXZ090"
  ],
  "references": [],
  "sent": "2025-04-02T06:44:00-05:00",
  "effective": "2025-04-02T06:44:00-05:00",
  "onset": "2025-04-02T06:44:00-05:00",
  "expires": "2025-04-02T07:15:00-05:00",
  "ends": null,
  "status": "Actual",
  "messageType": "Alert",
  "category": "Met",
  "severity": "Moderate",
  "certainty": "Observed",
  "urgency": "Expected",
  "event": "Special Weather Statement",
  "sender": "w-nws.webmaster@noaa.gov",
  "senderName": "NWS Norman OK",
  "headline": "Special Weather Statement issued April 2 at 6:44AM CDT by NWS Norman OK",
  "description": "At 643 AM CDT, Doppler radar was tracking a strong thunderstorm 10\nmiles east of Olney, moving northeast at 60 mph.\n\nHAZARD...Wind gusts up to 50 mph and penny size hail.\n\nSOURCE...Radar indicated.\n\nIMPACT...Gusty winds could knock down tree limbs and blow around\nunsecured objects. Minor damage to outdoor objects is\npossible.\n\nLocations impacted include...\nShannon, Joy, and Windthorst.",
  "instruction": "If outdoors, consider seeking shelter inside a building.",
  "response": "Execute",
  "parameters": {
    "AWIPSidentifier": [
      "SPSOUN"
    ],
    "WMOidentifier": [
      "WWUS84 KOUN 021144"
    ],
    "NWSheadline": [
      "A strong thunderstorm will impact portions of southwestern Clay and southeastern Archer Counties through 715 AM CDT"
    ],
    "eventMotionDescription": [
      "2025-04-02T11:43:00-00:00...storm...241DEG...53KT...33.4,-98.59"
    ],
    "maxWindGust": [
      "50 MPH"
    ],
    "maxHailSize": [
      "0.75"
    ],
    "BLOCKCHANNEL": [
      "EAS",
      "NWEM",
      "CMAS"
    ],
    "EAS-ORG": [
      "WXR"
    ]
  }
}

次に天気予報。こちらは緯度経度で指定する。以下はテキサス州オースチンのテキサス大学の緯度経度を渡している。

curl https://api.weather.gov/points/30.2851,-97.733 \
    -H "User-Agent: weather-app/1.0" \
    -H "Accept: application/geo+json" | jq -r .properties

{
  "@id": "https://api.weather.gov/points/30.2851,-97.733",
  "@type": "wx:Point",
  "cwa": "EWX",
  "forecastOffice": "https://api.weather.gov/offices/EWX",
  "gridId": "EWX",
  "gridX": 156,
  "gridY": 92,
  "forecast": "https://api.weather.gov/gridpoints/EWX/156,92/forecast",
  "forecastHourly": "https://api.weather.gov/gridpoints/EWX/156,92/forecast/hourly",
  "forecastGridData": "https://api.weather.gov/gridpoints/EWX/156,92",
  "observationStations": "https://api.weather.gov/gridpoints/EWX/156,92/stations",
  "relativeLocation": {
    "type": "Feature",
    "geometry": {
      "type": "Point",
      "coordinates": [
        -97.754358,
        30.30394
      ]
    },
    "properties": {
      "city": "Austin",
      "state": "TX",
      "distance": {
        "unitCode": "wmoUnit:m",
        "value": 2931.4871636478
      },
      "bearing": {
        "unitCode": "wmoUnit:degree_(angle)",
        "value": 135
      }
    }
  },
  "forecastZone": "https://api.weather.gov/zones/forecast/TXZ192",
  "county": "https://api.weather.gov/zones/county/TXC453",
  "fireWeatherZone": "https://api.weather.gov/zones/fire/TXZ192",
  "timeZone": "America/Chicago",
  "radarStation": "KGRK"
}

結果に天気予報のURL（forecast）が含まれるので、これを再度リクエスト。こちらも時間ごとに複数の結果がリストに入ってくるので、1つだけピックアップ。

curl https://api.weather.gov/gridpoints/EWX/156,92/forecast \
    -H "User-Agent: weather-app/1.0" \
    -H "Accept: application/geo+json" | jq -r .properties.periods[0]

{
  "number": 1,
  "name": "Today",
  "startTime": "2025-04-02T08:00:00-05:00",
  "endTime": "2025-04-02T18:00:00-05:00",
  "isDaytime": true,
  "temperature": 88,
  "temperatureUnit": "F",
  "temperatureTrend": "",
  "probabilityOfPrecipitation": {
    "unitCode": "wmoUnit:percent",
    "value": 20
  },
  "windSpeed": "5 to 10 mph",
  "windDirection": "SSE",
  "icon": "https://api.weather.gov/icons/land/day/tsra,20/bkn?size=medium",
  "shortForecast": "Slight Chance Showers And Thunderstorms then Mostly Cloudy",
  "detailedForecast": "A slight chance of showers and thunderstorms before 10am. Mostly cloudy, with a high near 88. South southeast wind 5 to 10 mph, with gusts as high as 25 mph. Chance of precipitation is 20%."
}