Posted inAI

PythonでカスタムMCPサーバーを構築する：Claude DesktopやAIエージェント向けのツール・データ連携を標準化

AI tutorial - IT technology blog
AI tutorial - IT technology blog

背景と導入理由

ある火曜日の午前2時15分、2つのアベイラビリティゾーンでPagerDutyのアラートが一斉に鳴り響きました。認証クラスタが完全に停止していたのです。チームメイトは4つのターミナルタブを行き来しながら、最初に500エラーを出したマイクロサービスを特定しようと、未整形のJSONダンプをClaudeに貼り付けていました。各内部サービスはそれぞれ独自のラッパースクリプト、カスタムOpenAIスキーマ定義、そして壊れやすいグルーコードに依存していました。APIの軽微な変更ひとつで、障害調査のワークフロー全体が完全に破綻してしまう状態だったのです。

AIクライアントごとに個別のアドホックなAPIコネクタを作成していては、拡張性（スケーラビリティ）を維持できません。Anthropicはこのアーキテクチャの混乱を解決するために、Model Context Protocol（MCP）を発表しました。MCPは、LLMにおけるLanguage Server Protocol（LSP）のようなものだと考えてください。複数のツール向けに連携機能を個別に再実装する代わりに、標準化されたインターフェースを一度実装するだけで済みます。クライアントは、標準入出力（stdio）またはServer-Sent Events（SSE）経由で機能の検出、コンテキストリソースの取得、アクションの実行を行います。

MCPを標準化したことで、私たちのチームのログ調査における平均復旧時間（MTTR）は24分から6分未満に短縮されました。サービスがMCPに対応すれば、Claude Desktop、Cursor、あるいは自作の自律型CLIエージェントなど、互換性のあるあらゆるホストが安全に運用メトリクスを照会し、データベースの状態を検査し、制御された修復手順を実行できるようになります。

インストール

システムツールとのパッケージ衝突を防ぐため、サーバーの依存関係は必ず分離してください。ここではPython 3.10以上を使用します。

1. 仮想環境のセットアップ

mkdir mcp-incident-server
cd mcp-incident-server
python3 -m venv .venv
source .venv/bin/activate

2. 公式MCP SDKのインストール

Anthropicの公式Python SDKに加え、非同期HTTP通信用のhttpxと厳格なスキーマ検証用のpydanticをインストールします：

pip install --upgrade "mcp[cli]" httpx pydantic

3. Python MCPサーバーの実装

server.pyという名前のファイルを作成します。SDKの高水準モジュールであるFastMCPを使用して、2つの運用ツールと1つのリアルタイムメトリクスリソースを公開します：

import sys
from datetime import datetime, timezone
from typing import Any, Dict, List
from mcp.server.fastmcp import FastMCP

# FastMCPサーバーインスタンスの初期化
mcp = FastMCP("production-ops-server")

# モックの運用データストア
INCIDENT_LOGS = [
    {
        "id": "INC-101",
        "service": "auth-api",
        "level": "CRITICAL",
        "message": "PostgreSQL接続プールが枯渇しました（アクティブ接続: 500/500）",
        "timestamp": "2026-10-03T01:58:12Z",
    },
    {
        "id": "INC-102",
        "service": "payment-gw",
        "level": "WARNING",
        "message": "Stripe Webhookゲートウェイのレイテンシが1420msに上昇（p99しきい値: 400ms）",
        "timestamp": "2026-10-03T02:05:44Z",
    },
]

@mcp.tool()
def fetch_incident_logs(service: str = "", severity: str = "") -> List[Dict[str, Any]]:
    """サービス名または重大度レベルでフィルタリングされたアクティブなインシデントログを照会します。"""
    results = INCIDENT_LOGS
    if service:
        results = [log for log in results if log["service"].lower() == service.lower()]
    if severity:
        results = [log for log in results if log["level"].lower() == severity.lower()]
    return results

@mcp.tool()
def restart_microservice_pod(service_name: str, force: bool = False) -> str:
    """障害発生時に異常なKubernetes Podのローリング再起動をトリガーします。"""
    valid_services = ["auth-api", "payment-gw", "order-processor"]
    if service_name not in valid_services:
        return f"エラー: 不明なサービス '{service_name}' です。許可されている対象: {valid_services}"
    
    action = "強制即時終了" if force else "正常なローリング再起動"
    now = datetime.now(timezone.utc).isoformat()
    return f"[{now}] 対象Pod '{service_name}' の{action}を開始しました（ステータス: 202 Accepted）。"

@mcp.resource("system://metrics/summary")
def get_system_metrics() -> str:
    """アクティブなクラスタリソースのテレメトリを構造化コンテキストリソースとして提供します。"""
    return "クラスタヘルス: DEGRADED | CPU: 78.4% | メモリ: 84.1% | アクティブDB接続数: 492/500"

if __name__ == "__main__":
    # クライアントリクエストを標準入出力（stdio）で待機
    mcp.run(transport="stdio")

設定

サーバーの実装が完了したら、AIホストクライアントに登録する必要があります。Claude Desktopは、ローカルテストおよび日常の運用における標準的な環境です。

1. Claude Desktopの設定

お使いのプラットフォームの設定ファイルを開きます：

  • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows: %APPDATA%\Claude\claude_desktop_config.json
  • Linux: ~/.config/Claude/claude_desktop_config.json

mcpServersセクションにサーバーを登録します。システムのPythonではなく、仮想環境内のPythonバイナリの絶対パスを必ず指定してください：

{
  "mcpServers": {
    "ops-debugger": {
      "command": "/home/ubuntu/mcp-incident-server/.venv/bin/python",
      "args": [
        "/home/ubuntu/mcp-incident-server/server.py"
      ],
      "env": {
        "LOG_LEVEL": "INFO",
        "DB_TIMEOUT_MS": "5000"
      }
    }
  }
}

2. セキュリティとI/Oの分離

データベースのパスワードやAPIキーをPythonのソースファイルに直接埋め込むことは避けてください。これらはenvブロックまたはOSの環境変数を介して注入します。

また、stdioモードで動作させる際は、print()を使ってデバッグメッセージを直接出力しないでください。FastMCPはJSON-RPCメッセージのシリアライズにstdoutを排他的に使用します。標準出力に生のテキストを書き込むと、メッセージフレームが破損し、クライアントとの接続が切断されます。内部の診断ログはすべてsys.stderrにルーティングしてください。

動作確認と監視

インフラ関連のツールを検証なしにデプロイしてはいけません。本番のLLMに接続する前に、公式のテストツールを使用してローカルでサーバーを検証しましょう。

1. MCP Inspectorによるテスト

Anthropicは、クライアントとサーバー間のハンドシェイクのシミュレーション、スキーマの検証、生のJSON-RPCトラフィックのキャプチャが可能な対話型Web UI「MCP Inspector」を提供しています。npxで実行します：

npx @modelcontextprotocol/inspector /home/ubuntu/mcp-incident-server/.venv/bin/python /home/ubuntu/mcp-incident-server/server.py

表示されたローカルURL（通常はhttp://localhost:5173）を開きます。左側のサイドバーでfetch_incident_logsを選択し、パラメータに{"service": "auth-api"}を入力して、Run Toolをクリックします。ツールがINC-101のレコードを正常な200レスポンス形式で返すことを確認してください。

2. 実際のクライアントでの確認

Claude Desktopを完全に終了して再起動します。メッセージバーの右下にあるハンマーのアイコンを確認してください。ops-debuggerがアクティブになり、2つのツール（fetch_incident_logs、restart_microservice_pod）とメトリクスリソースが公開されているはずです。

次のプロンプトを使ってエンドツーエンドの動作をテストします：「auth-apiのアクティブなインシデントログを確認し、Podを再起動する必要があるかどうか教えてください。」

Claudeはスキーマを検査し、fetch_incident_logsの実行許可を求めた上で、データベース接続のボトルネックを分析し、ローリング再起動を提案します。

3. よくあるランタイムエラーのトラブルシューティング

  • JSON-RPC Parse Errors（パースエラー）: 不要なprint文や未ハンドルのトレースバックがstdoutを汚染しています。logging.basicConfig(stream=sys.stderr)を使用して、すべてのカスタムログをstderrに向けてください。
  • Subprocess Exit Code 1 / File Not Found（終了コード1 / ファイルが見つからない）: claude_desktop_config.jsonで指定されたバイナリパスが間違っています。有効化した.venv内でwhich pythonを実行し、その正確なパス文字列を設定に貼り付けてください。
  • Client Timeouts（クライアントタイムアウト）: 実行時間の長いSQLクエリやAPI呼び出しがワーカーループをブロックしています。クライアントの接続が切断される前に適切に失敗するよう、外部I/Oをasyncio.wait_for(..., timeout=5.0)でラップしてください。
Share:
Tags: