MCP エージェントの実装でつまずいた4箇所 — 集計ツール・並列実行・入力検証を実測で見直す

mkdir my-mcp-server
cd my-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node ts-node

import { McpServer, ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
 
const server = new McpServer({
  name: "data-analyzer",
  version: "1.0.0",
});
 
// データ集計ツールの定義
server.tool(
  "aggregate_data",
  "表形式データの指定カラムを集計して統計情報を返す",
  {
    // 実データには文字列カラムが必ず混ざる。値の型はここでは緩く受け、集計側で絞り込む
    data: z.array(z.record(z.string(), z.unknown())).describe("集計対象の行配列"),
    column: z.string().describe("集計するカラム名"),
    operation: z.enum(["sum", "average", "max", "min"]).describe("集計操作"),
  },
  async ({ data, column, operation }) => {
    const raw = data.map(row => row[column]);
    // undefined だけでなく null・数値文字列・NaN・Infinity も落とす
    const values = raw.filter((v): v is number => typeof v === "number" && Number.isFinite(v));
    const skipped = raw.length - values.length;
 
    // 空集合を数値で答えない。Claude に「無かった」と言葉で伝える
    if (values.length === 0) {
      return {
        isError: true,
        content: [{
          type: "text",
          text: `カラム "${column}" に数値が1件もありませんでした（${raw.length} 行を検査）。` +
                `カラム名の綴りか、値が文字列として格納されていないかをご確認ください。`,
        }],
      };
    }
 
    let result: number;
    switch (operation) {
      case "sum": result = values.reduce((a, b) => a + b, 0); break;
      case "average": result = values.reduce((a, b) => a + b, 0) / values.length; break;
      // spread は要素数が増えるとコールスタックを溢れさせるため reduce で畳む
      case "max": result = values.reduce((a, b) => (b > a ? b : a)); break;
      case "min": result = values.reduce((a, b) => (b < a ? b : a)); break;
      default: {
        const exhaustive: never = operation;
        throw new Error(`未対応の operation です: ${String(exhaustive)}`);
      }
    }
 
    return {
      content: [{
        type: "text",
        // skipped を返すと、Claude が「一部しか集計できていない」ことに気づける
        text: JSON.stringify({ column, operation, result, count: values.length, skipped }),
      }],
    };
  }
);
 
// リソースの定義（テンプレートURI）
server.resource(
  "report",
  new ResourceTemplate("report://{date}", { list: undefined }),
  async (uri, { date }) => ({
    contents: [{
      uri: uri.href,
      text: `${date}のレポートデータ（サンプル）`,
    }],
  })
);
 
// STDIOトランスポートで起動
const transport = new StdioServerTransport();
await server.connect(transport);

{
  "mcpServers": {
    "data-analyzer": {
      "command": "node",
      "args": ["/path/to/my-mcp-server/dist/index.js"],
      "env": {
        "NODE_ENV": "production"
      }
    }
  }
}

from anthropic import Anthropic
 
client = Anthropic()
 
def daily_report_workflow(sources: list[str]) -> str:
    """毎日のレポート生成ワークフロー"""
    
    # Step 1: 各ソースから情報収集（並列）
    collection_prompt = f"""
    以下のニュースソースから今日の重要な情報を収集してください：
    {', '.join(sources)}
    
    各ソースについて：
    1. fetch_webpageツールでページを取得
    2. 最新5件の記事タイトルと概要を抽出
    3. 重要度を1-5で評価
    """
    
    collection_result = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=8192,
        messages=[{"role": "user", "content": collection_prompt}]
    )
    
    # Step 2: 収集した情報を分析・要約
    summary_prompt = f"""
    収集した情報を分析し、今日のエグゼクティブサマリーを作成してください。
    
    収集情報：
    {collection_result.content[0].text}
    
    サマリーの要件：
    - 最重要トピック3件をトップに
    - 各トピックは3行以内で簡潔に
    - 業界への影響と推奨アクションを含める
    - Slack での読みやすさを考慮したMarkdown形式
    """
    
    summary = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        messages=[{"role": "user", "content": summary_prompt}]
    )
    
    return summary.content[0].text

import json
 
TRIAGE_TOOL = {
    "name": "record_triage",
    "description": "問い合わせの分類結果を記録する",
    "input_schema": {
        "type": "object",
        "properties": {
            "category": {"type": "string", "enum": ["billing", "technical", "general", "complaint"]},
            "urgency": {"type": "string", "enum": ["high", "medium", "low"]},
            "reason": {"type": "string", "description": "その判断に至った根拠"},
        },
        "required": ["category", "urgency", "reason"],
    },
}
 
def triage_inquiry(email_content: str) -> dict:
    """問い合わせのトリアージ。構造化された結果だけを受け取る"""
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system=(
            "あなたはカスタマーサポートのトリアージ担当です。"
            "問い合わせを分類し、record_triage ツールで結果を記録してください。\n"
            "urgency の目安: high は即座に人間が対応、medium は4時間以内、low は24時間以内。"
        ),
        tools=[TRIAGE_TOOL],
        tool_choice={"type": "tool", "name": "record_triage"},  # 必ずこのツールを使わせる
        messages=[{"role": "user", "content": email_content}],
    )
 
    for block in response.content:
        if getattr(block, "type", None) == "tool_use" and block.name == "record_triage":
            return block.input  # 既にスキーマ検証済みの dict
 
    # ここに来るのは想定外。人間のキューへ回す
    raise RuntimeError("トリアージ結果を取得できませんでした")

import time
import random
from typing import Callable, TypeVar
 
T = TypeVar('T')
 
def with_retry(
    func: Callable[[], T],
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0
) -> T:
    """指数バックオフ + ジッターによるリトライ"""
    
    for attempt in range(max_attempts):
        try:
            return func()
        except Exception as e:
            if attempt == max_attempts - 1:
                raise  # 最後の試行は例外を再スロー
            
            # 指数バックオフ + ランダムジッター
            delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.1f}s...")
            time.sleep(delay)

import logging
import json
from datetime import datetime, timezone
 
class AgentLogger:
    def __init__(self, workflow_id: str):
        self.workflow_id = workflow_id
        self.logger = logging.getLogger(f"agent.{workflow_id}")
    
    def log_tool_call(self, tool_name: str, input_data: dict, output_data: dict, duration_ms: float):
        """ツール呼び出しのログ"""
        self.logger.info(json.dumps({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "workflow_id": self.workflow_id,
            "event": "tool_call",
            "tool": tool_name,
            "input_tokens": len(str(input_data)),
            "output_tokens": len(str(output_data)),
            "duration_ms": duration_ms,
        }))
    
    def log_agent_decision(self, agent_id: str, decision: str, reasoning: str):
        """エージェントの判断ログ"""
        self.logger.info(json.dumps({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "workflow_id": self.workflow_id,
            "event": "agent_decision",
            "agent_id": agent_id,
            "decision": decision,
            "reasoning": reasoning[:500],  # 長すぎる場合は切り詰め
        }))

response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=4096,
    system=[
        {
            "type": "text",
            "text": "（長いシステムプロンプト...）",
            "cache_control": {"type": "ephemeral"}  # キャッシュを有効化
        }
    ],
    messages=[{"role": "user", "content": user_message}]
)

def summarize_history(messages: list, threshold: int = 20) -> list:
    """古いメッセージを要約して圧縮"""
    if len(messages) <= threshold:
        return messages
    
    old_messages = messages[:-threshold]
    recent_messages = messages[-threshold:]
    
    summary_response = client.messages.create(
        model="claude-haiku-4-5",  # 要約にはHaikuで十分
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"以下の会話履歴を簡潔に要約してください：\n\n{json.dumps(old_messages)}"
        }]
    )
    
    summary = summary_response.content[0].text
    return [{"role": "assistant", "content": f"[会話履歴の要約]: {summary}"}] + recent_messages

# 副作用のあるツールと、読み取りだけのツールを分けて持つ
WRITE_TOOLS = {"send_email", "delete_file", "post_message", "create_pull_request"}
READ_TOOLS = {"search_docs", "fetch_webpage", "query_database"}
 
def gate_tool_call(tool_name: str, tool_input: dict, *, context_is_untrusted: bool) -> str:
    """ツール実行の直前に通す関門。入力の文言ではなく、実行される操作で判断する"""
    if tool_name not in WRITE_TOOLS | READ_TOOLS:
        return "deny"  # 知らないツールは既定で拒否する
 
    # 外部から取り込んだテキストが文脈に混ざっている間は、副作用を人間の承認待ちにする
    if tool_name in WRITE_TOOLS and context_is_untrusted:
        return "require_human_approval"
 
    return "allow"