Claude MCP × エージェントワークフロー：実践的な自動化システムの設計と構築

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",
  "CSVデータを集計して統計情報を返す",
  {
    data: z.array(z.record(z.string(), z.number())).describe("集計対象のデータ配列"),
    column: z.string().describe("集計するカラム名"),
    operation: z.enum(["sum", "average", "max", "min"]).describe("集計操作"),
  },
  async ({ data, column, operation }) => {
    const values = data.map(row => row[column]).filter(v => v \!== undefined);
    
    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;
      case "max": result = Math.max(...values); break;
      case "min": result = Math.min(...values); break;
    }
    
    return {
      content: [{
        type: "text",
        text: JSON.stringify({ column, operation, result, count: values.length }),
      }],
    };
  }
);
 
// リソースの定義（テンプレート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

def triage_inquiry(email_content: str) -> dict:
    """問い合わせのトリアージ"""
    
    triage_result = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="""あなたはカスタマーサポートのトリアージ専門家です。
        問い合わせを以下のカテゴリに分類し、緊急度を判定してください：
        - billing: 請求・支払いに関する問題
        - technical: 技術的な問題
        - general: 一般的な質問
        - complaint: クレーム・不満
        
        緊急度: high（即座に人間が対応）/ medium（4時間以内）/ low（24時間以内）
        
        JSON形式で回答してください。""",
        messages=[{"role": "user", "content": email_content}]
    )
    
    return json.loads(triage_result.content[0].text)

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
 
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.utcnow().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.utcnow().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

def sanitize_user_input(user_input: str) -> str:
    """ユーザー入力のサニタイゼーション"""
    # 疑わしいパターンを検出
    suspicious_patterns = [
        "ignore previous instructions",
        "system prompt",
        "forget everything",
    ]
    
    lower_input = user_input.lower()
    for pattern in suspicious_patterns:
        if pattern in lower_input:
            raise ValueError(f"Suspicious input detected: {pattern}")
    
    # 長さ制限
    if len(user_input) > 10000:
        raise ValueError("Input too long")
    
    return user_input