意味理解 LLM -- MCP 統合ガイド

概要

LLM スタンドアロンアプリケーションには Model Context Protocol (MCP) が完全に統合されており、外部の MCP サーバーと通信する機能を備えています。MCP を通じて、AI アシスタントはリモートのツールやサービスを呼び出し、機能の範囲を拡張できます。

🌟 特長

MCP プロトコルへの完全対応 - 最新の MCP プロトコル仕様に対応
複数の接続タイプ - streamable-http、SSE、stdio などの接続方式に対応
バックグラウンドでのヘルスモニタリング - サーバーの状態と接続の健全性を自動検出
スマートな再接続機構 - 指数バックオフによるリトライ戦略で接続の安定性を確保
ツールの統一的な呼び出し - ローカルツールと MCP ツールをシームレスに統合
設定のホットリロード - MCP サーバーの動的な設定に対応

🔧 設定方法

1. MCP サーバーの設定

utils/mcp/config/mcp_servers.json ファイルを編集します：

{
  "mcpServers": {
    "example-server": {
      "active": true,
      "type": "streamable-http",
      "url": "http://your-server.com:8000/mcp/",
      "timeout": 30,
      "sse_read_timeout": 300
    },
    "another-server": {
      "active": true,
      "type": "streamable-http",
      "url": "http://another-server.com:8001/mcp/",
      "timeout": 30,
      "sse_read_timeout": 300
    },
    "optional-service": {
      "active": false,
      "type": "sse",
      "url": "https://api.example.com/sse",
      "headers": {
        "Authorization": "Bearer your_token_here"
      }
    }
  }
}

2. 接続タイプの説明

streamable-http

双方向のストリーミング通信が必要な場合に適しています：

{
  "type": "streamable-http",
  "url": "http://example.com/mcp/",
  "timeout": 30,
  "sse_read_timeout": 300,
  "terminate_on_close": true
}

SSE (Server-Sent Events)

一方向のストリーミングデータのプッシュに適しています：

{
  "type": "sse",
  "url": "https://api.example.com/sse",
  "timeout": 5,
  "sse_read_timeout": 300,
  "headers": {
    "Authorization": "Bearer token"
  }
}

stdio

ローカルプロセス間の通信に適しています：

{
  "command": "python",
  "args": ["mcp_server.py"],
  "env": {
    "API_KEY": "your_key_here"
  }
}

3. サーバーパラメータの詳細

パラメータ	型	必須	説明
`active`	boolean	はい	このサーバーを有効にするかどうか
`type`	string	はい	接続タイプ：streamable-http/sse/stdio
`url`	string	URL 接続時は必須	サーバーのアドレス
`timeout`	number	いいえ	接続タイムアウト時間（秒）
`sse_read_timeout`	number	いいえ	SSE 読み取りタイムアウト（秒）
`headers`	object	いいえ	リクエストヘッダー情報
`command`	string	stdio 時は必須	実行ファイルのパス
`args`	array	いいえ	コマンドライン引数
`env`	object	いいえ	環境変数

🚀 使用方法

1. 基本的な使い方

MCP ツールは対話システムに自動的に統合されるため、特別な呼び出しは不要です：

# 在对话中直接使用用户配置的MCP工具
user: "调用我的自定义工具处理这个请求"
assistant: # 自动调用用户配置的MCP工具

user: "执行特定的外部服务操作"
assistant: # 自动调用相应的MCP服务器工具

2. プログラミングインターフェース

MCP ツールを直接呼び出します：

from utils.mcp import enhanced_function_call, get_enhanced_function_schemas

# 获取所有可用工具（本地+MCP）
schemas = await get_enhanced_function_schemas()
print(f"可用工具数量: {len(schemas)}")

# 调用特定MCP工具
result = await enhanced_function_call("your_custom_tool", param1="value1")
print(f"检测结果: {result}")

3. ステータスの監視

MCP サーバーの接続状態を確認します：

from utils.mcp import get_mcp_client

client = await get_mcp_client()
status = client.get_server_status()
print(status)

出力例：

{
  "mcp_available": true,
  "total_servers": 3,
  "connected_servers": 2,
  "total_tools": 15,
  "servers": [
    {
      "name": "vision-arm-control",
      "status": "connected",
      "tools_count": 8,
      "last_health_check": "2025-01-15T10:30:00"
    },
    {
      "name": "face-tracker",
      "status": "connected",
      "tools_count": 3,
      "last_health_check": "2025-01-15T10:30:15"
    },
    {
      "name": "amap-maps",
      "status": "failed",
      "error_message": "Connection timeout",
      "next_retry_time": "2025-01-15T10:35:00"
    }
  ]
}

🛠️ MCP サーバーの開発

独自の MCP サーバーを開発する場合は、FastMCP フレームワークの使用をおすすめします：

1. FastMCP のインストール

pip install fastmcp

2. MCP サーバーの作成

# my_mcp_server.py
from fastmcp import FastMCP

mcp = FastMCP("My MCP Server")

@mcp.tool()
def calculate(operation: str, a: float, b: float) -> float:
    """执行数学计算

    Args:
        operation: 运算类型 (add, sub, mul, div)
        a: 第一个数字
        b: 第二个数字

    Returns:
        计算结果
    """
    operations = {
        "add": lambda x, y: x + y,
        "sub": lambda x, y: x - y,
        "mul": lambda x, y: x * y,
        "div": lambda x, y: x / y if y != 0 else None
    }

    if operation in operations:
        return operations[operation](a, b)
    else:
        raise ValueError(f"不支持的运算: {operation}")

if __name__ == "__main__":
    mcp.run(transport="stdio")

3. LLM アプリケーションへの設定

mcp_servers.json に次の内容を追加します：

{
  "my-calculator": {
    "active": true,
    "command": "python",
    "args": ["path/to/my_mcp_server.py"]
  }
}

🔧 トラブルシューティング

よくある問題

接続に失敗する

サーバーの URL が正しいか確認します
ネットワークの疎通を確認します
サーバーのログを確認します

ツールの呼び出しに失敗する

ツールのパラメータ形式が正しいか確認します
サーバーがそのツールに対応しているか確認します
詳細なエラー情報を確認します

再接続の問題

システムは指数バックオフ戦略で自動的にリトライします
サーバーの安定性を確認します
タイムアウトのパラメータを調整します

デバッグモード

詳細ログを有効にします：

import logging
logging.getLogger("utils.mcp").setLevel(logging.DEBUG)

📚 関連リソース

MCP 公式仕様
FastMCP フレームワーク
MCP サーバーの例
プロジェクトのアーキテクチャドキュメント

🎯 ベストプラクティス

サーバー管理

タイムアウト時間を適切に設定します
サーバーの健全性を監視します
ロードバランシングを利用します

ツール設計

明確なツールの説明
十分なパラメータ検証
エラー処理の仕組み

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

HTTPS 接続を使用します
サーバー証明書を検証します
ツールの権限を制限します

MCP を完全に統合することで、LLM スタンドアロンアプリケーションはさまざまな外部サービスやツールに接続でき、AI アシスタントの能力の範囲が大きく広がります。