Week 3 教室 / 最大の伸びしろ週 / 対応: D2全体・D4全体

MCPを「作る側」+構造化抽出パイプライン
— 使い慣れた道具の、設計者の視点

MCPサーバーを何十本も「使って」きた経験を、「作る側」の知識に反転させる週。後半は構造化データ抽出(D4)——スキーマ設計・few-shot・検証リトライ・バッチの4点セット。合わせて配点38%をカバーする。

このレッスンの流れ(約5〜6時間 / 週内に分割OK)

STEP 1MCPサーバーの最小実装を読む

MCPサーバーの正体は「ツールのカタログと実装をまとめて、標準の通信規約(Model Context Protocol)で公開する小さなプログラム」。Python なら公式SDKで20行で書ける。

# 最小のMCPサーバー(天気ツール1個)
# pip install "mcp[cli]"
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather")   # サーバー名

@mcp.tool()
def get_weather(city: str) -> str:
    """指定した都市の現在の天気を返す。

    都市名は日本語・英語どちらでも可(例: 東京, Osaka)。
    現在の天気・気温の質問に使う。過去の天気には get_weather_history を使う。
    """
    # ↑ この docstring がそのまま「ツールの説明文」としてAIに渡る(D2.1の主戦場)
    return f"{city}: 晴れ、28度"

if __name__ == "__main__":
    mcp.run()   # 標準入出力で通信(Claude Code から接続できる)
  • W1でツールカタログ(name / description / input_schema)を手書きしたのを思い出す。MCPはそれを関数定義と docstring から自動生成してくれるだけ——新概念はほぼゼロ
  • だから試験も「サーバーの立て方」ではなく「説明文の書き方・エラーの返し方・ツールの分け方」=設計判断を問う(インフラ・ホスティングは出題範囲外と明言されている)

STEP 2「良いツール」の3条件

① 説明文(D2.1)— AIの唯一の手がかり

上のdocstringに詰めた4要素:何をするか/入力の形式と例/エッジケース/境界(いつ使い・いつ使わないか)。「過去の天気には get_weather_history を」の1行が誤ルーティングを防ぐ肝。

② 構造化エラー(D2.2)— 失敗を情報として返す

@mcp.tool()
def get_weather(city: str) -> dict:
    """(説明文は同じ)"""
    if not weather_api.is_alive():
        return {                              # 一時的な障害 → リトライする価値あり
            "error": True,
            "errorCategory": "transient",
            "isRetryable": True,
            "message": "天気サービスが応答しません。少し待って再試行してください。",
        }
    result = weather_api.lookup(city)
    if result is None:
        return {                              # 入力の問題 → リトライしても無駄
            "error": True,
            "errorCategory": "validation",
            "isRetryable": False,
            "message": f"都市 '{city}' が見つかりません。表記を確認してください。",
        }
    return {"city": city, "weather": result}  # 成功

# 重要な区別: 「検索は成功したが該当0件」はエラーではなく正常な結果として返す

③ 数の絞り込み(D2.3)

1体のエージェントに渡すツールは役割に必要な4〜5個まで。18個渡すと選択の信頼性が落ちる。高頻度の単純ニーズには機能を絞った専用ツール(例:統合係に verify_fact だけ渡す)、複雑ケースはコーディネーター経由——「85/15の分離」が模範解答の型。

STEP 3設定ファイルと「ツール以外のもの」

// .mcp.json(プロジェクト直下・チーム共有・git管理)
{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["servers/weather.py"],
      "env": { "WEATHER_API_KEY": "${WEATHER_API_KEY}" }   // ← 直書き禁止。環境変数展開
    }
  }
}
// 個人の実験用サーバーは ~/.claude.json 側に書く(チームに配らない)
  • MCPリソース:ツール(実行するもの)と別に、参照データのカタログ(issue一覧・DBスキーマ・ドキュメント階層)を公開する仕組み。エージェントが「探るためのツールコール」を繰り返さずに済む
  • 標準的な統合(Jira等)はコミュニティ製の既存サーバーを使う。自作はチーム固有ワークフローだけ——resilience-first の「枯れたものを使う」と同じ判断

STEP 4構造化抽出①:スキーマ設計の3つの知恵(D4.3)

W1のSTEP 7(tool_choice強制でJSONを取る)が土台。その上のスキーマ設計の試験ポイントは3つ:

  1. 無いかもしれない欄は nullable"due_date": {"type": ["string", "null"]}。requiredにすると、モデルが「埋めなきゃ」と値を捏造する。「無いものは無い」と言える設計が捏造防止
  2. enumには "other"+詳細文字列の逃げ道"category": {"enum": ["請求", "返品", "other"]}"category_detail": {"type": "string"}。想定外を無理やり既存カテゴリに押し込ませない。曖昧ケース用に "unclear" を足すのも同じ発想
  3. スキーマは構文を保証するが意味は保証しない:形式は完璧でも「明細の合計≠総額」「値が隣の欄に入る」は起きる。→ 意味の検証は次のSTEP 5で

STEP 5構造化抽出②:検証リトライループ(D4.4)

# retry-with-error-feedback の骨格
def extract_with_retry(document, max_retries=2):
    result = extract(document)                    # tool_choice強制で抽出(W1 STEP7)
    for _ in range(max_retries):
        errors = validate(result)                 # Pydantic等で意味の検証
        if not errors:
            return result
        # ★ 具体的なエラー内容を添えて再依頼するのが肝
        result = extract(
            document,
            feedback=f"前回の抽出はこの検証に失敗した: {errors}。"
                     f"前回の出力: {result}。エラーを修正して再抽出せよ。"
        )
    return flag_for_human_review(result)          # 直らなければ人間レビュー行き
  • リトライが効く失敗:フォーマット不一致・構造ミス/効かない失敗:情報がそもそも文書に無い(何度やっても出ない。nullを許して先へ)
  • 自己検証の仕込み:「stated_total(記載の総額)」と「calculated_total(明細から計算)」を両方抽出させ、不一致なら conflict_detected: true を立てる——data-accuracy の合計突合と同じ技
  • 誤検知の分析用に detected_pattern(何を根拠に判定したか)を出力に含めると、後で「どのパターンが誤検知源か」を集計できる

STEP 6構造化抽出③:バッチ処理(D4.5)

# Message Batches API の骨格
batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": f"doc-{i}",              # ★ 結果との対応付けに必須
            "params": { "model": "claude-haiku-4-5", "max_tokens": 1024,
                        "messages": [...], "tools": [...], "tool_choice": {...} },
        }
        for i, doc in enumerate(documents)        # 100件まとめて投入
    ]
)
# → 後でポーリングして結果回収。失敗した custom_id だけ修正して再投入

暗記3点セット+判断1つ:

  • 50%引き/最大24時間/レイテンシ保証なし
  • 制約:バッチ内では多ターンのツール往復ができない(途中でツールを実行して結果を返す、が不可)
  • 判断:人が待つ処理(マージ前チェック)は同期API、待てる処理(夜間レポート・週次監査)はバッチ——「両方バッチ」「両方同期」は誤答
  • 大量投入の前にサンプル数件でプロンプトを磨く(一発成功率を上げて再投入コストを抑える)

STEP 7演習課題(公式演習3・4の日本語版・約3時間)

私(Claude)に「W3演習やる」と言えば伴走する。

  1. MCPサーバー自作:STEP 1 の weather サーバーに get_weather_history を足し、境界を書き分けた説明文で「昨日の東京の天気」「今の東京の天気」が正しく振り分けられるかを Claude Code に接続してテストする
  2. 構造化エラー:STEP 2 ②の3パターン(transient / validation / 正当な0件)を実装し、Claudeの反応の違い(リトライする/表記を聞き直す/「該当なし」と答える)を観察する
  3. 抽出パイプライン:領収書テキスト5件(うち1件は日付欠落、1件は合計不一致)を用意し、nullable スキーマ+検証リトライで処理。日付欠落が null になり(捏造しない)、合計不一致に conflict_detected が立つことを確認する
  4. few-shot の効果測定:形式の違う文書2種で抽出がブレたら、few-shot例を2個足して安定するかを比較する
  5. (任意)バッチ:同じ抽出を Batches API で10件投入し、custom_id で結果を回収する

理解度チェック(30秒で説明できたらチェック)

終わったら

D2D4のチェックリストを更新 → 模試のD2/D4問題 → Week 4(総仕上げ)へ。

← W2 教室 W4 教室:総仕上げ →