MCP 応用 — サンプリング・通知・本番運用

MCP 入門の続き。本番サーバー実装・Sampling・Notifications の高度な機能。

1. Sampling — モデルへの問い合わせをサーバーが要求する

通常の MCP フロー（Tool Use）:
  ホスト（Claude） → [call_tool] → MCP サーバー → ツール実行 → 結果を返す

Sampling:
  MCP サーバー → [create_message リクエスト] → ホスト → Claude に問い合わせ → 結果をサーバーに返す
  ↑ サーバー側が「Claude に判断させたい」ときに使う

用途:
  - 取得したデータの要約を Claude に作らせる
  - 複数の選択肢から Claude に最適なものを選ばせる
  - エラーの原因を Claude に分析させる

from mcp.server import Server
from mcp.types import (
    CreateMessageRequest,
    CreateMessageResult,
    TextContent,
    SamplingMessage,
)

server = Server("intelligent-analyzer")

@server.call_tool()
async def analyze_logs(arguments: dict) -> list[TextContent]:
    """ログを取得し、Claude に分析させる"""
    # 1. ログデータを取得
    raw_logs = fetch_cloudwatch_logs(
        log_group=arguments["log_group"],
        hours=arguments.get("hours", 1)
    )
    
    # 2. Sampling で Claude に分析を依頼
    result: CreateMessageResult = await server.request_context.session.create_message(
        CreateMessageRequest(
            messages=[
                SamplingMessage(
                    role="user",
                    content=TextContent(
                        type="text",
                        text=f"""以下のログを分析してエラーパターンと原因を特定してください:

{raw_logs[:5000]}  # 長すぎる場合は切り詰め

出力形式:
1. 発見されたエラー（件数付き）
2. 推定される根本原因
3. 対処の優先順位"""
                    )
                )
            ],
            maxTokens=1024,
            system="あなたはSREエンジニアです。ログを分析し、簡潔かつ実用的な回答をしてください。"
        )
    )
    
    # 3. Claude の分析結果を返す
    return [TextContent(type="text", text=result.content.text)]

2. Notifications — 長時間処理の進捗通知

from mcp.types import ProgressNotification

@server.call_tool()
async def run_deployment(arguments: dict) -> list[TextContent]:
    """デプロイ処理（長時間）の進捗を通知しながら実行"""
    steps = [
        ("ビルド中", build_image),
        ("テスト中", run_tests),
        ("ECR プッシュ中", push_to_ecr),
        ("ECS デプロイ中", deploy_to_ecs),
        ("ヘルスチェック", verify_health),
    ]
    
    for i, (description, func) in enumerate(steps):
        # クライアント（Claude）に進捗を通知
        await server.request_context.session.send_notification(
            ProgressNotification(
                method="notifications/progress",
                params={
                    "progressToken": arguments.get("progressToken"),
                    "progress": i / len(steps),
                    "total": 1.0,
                    "message": description
                }
            )
        )
        
        await func(arguments)
    
    return [TextContent(type="text", text="デプロイ完了")]

3. Resources — 動的コンテンツの提供

from mcp.types import Resource, ReadResourceResult

@server.list_resources()
async def list_resources() -> list[Resource]:
    """利用可能なリソースの一覧を返す"""
    # 動的に生成（例: S3 バケット内のファイル一覧）
    files = list_s3_files("my-config-bucket")
    return [
        Resource(
            uri=f"s3://my-config-bucket/{f}",
            name=f,
            mimeType="application/json" if f.endswith(".json") else "text/plain"
        )
        for f in files
    ]

@server.read_resource()
async def read_resource(uri: str) -> ReadResourceResult:
    """リソースの内容を返す"""
    # URI からバケット・キーを解析
    parts = uri.replace("s3://", "").split("/", 1)
    bucket, key = parts[0], parts[1]
    
    content = get_s3_object(bucket, key)
    
    return ReadResourceResult(
        contents=[
            TextContent(type="text", text=content)
        ]
    )

4. 本番 MCP サーバーの構成

開発:
  claude <--(stdio)--> mcp-server.py

本番（ネットワーク越し）:
  Claude Desktop ──(HTTP/SSE)──> MCP サーバー (FastAPI) ──> AWS

セキュリティ考慮:
  - 認証: API キー / mTLS
  - 権限: 最小権限の IAM ロール
  - ログ: 全 tool call を CloudWatch に記録
  - レート制限: 1ユーザーあたりの呼び出し数制限

# HTTP/SSE トランスポートを使った本番 MCP サーバー
from mcp.server.sse import SseServerTransport
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

app = FastAPI()

# CORS 設定（Claude Desktop からのアクセスを許可）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://claude.ai"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# API キー認証
async def verify_api_key(request: Request):
    api_key = request.headers.get("X-API-Key")
    if api_key != os.environ["MCP_API_KEY"]:
        raise HTTPException(status_code=401, detail="Invalid API key")

@app.get("/sse", dependencies=[Depends(verify_api_key)])
async def sse_endpoint(request: Request):
    transport = SseServerTransport("/messages")
    await server.run(
        transport.connect_sse(request),
        InitializationOptions(server_name="infra-tools", server_version="1.0.0")
    )

@app.post("/messages", dependencies=[Depends(verify_api_key)])
async def message_endpoint(request: Request):
    return await transport.handle_post_message(request)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

5. claude_desktop_config.json への登録

{
  "mcpServers": {
    "infra-tools-local": {
      "command": "python",
      "args": ["/path/to/mcp_server.py"],
      "env": {
        "AWS_PROFILE": "default"
      }
    },
    "infra-tools-remote": {
      "url": "https://mcp.example.com/sse",
      "headers": {
        "X-API-Key": "your-secret-key"
      }
    }
  }
}

6. デバッグとテスト

# MCP Inspector でサーバーをテスト
npx @modelcontextprotocol/inspector python mcp_server.py

# ログ出力（stderr に書く）
import sys
print("デバッグメッセージ", file=sys.stderr)

# ユニットテスト
from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client

async def test_tool():
    async with stdio_client(
        server_params=StdioServerParameters(
            command="python", args=["mcp_server.py"]
        )
    ) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool(
                "analyze_logs",
                {"log_group": "/aws/lambda/test"}
            )
            assert result.content[0].text  # レスポンスがあることを確認

🔗 関連リソース

• 01_MCP入門.md — 基礎から始める
• Anthropic MCP ドキュメント
• MCP Skilljar: MCP Advanced Topics