STUDY NOTES

第20回: MCP × LangGraph — `create_react_agent` から MCP サーバの Tool を呼ぶナレッジエージェント¶

第19回で「MCP サーバを単独で立てて Cursor などのホストから JSON-RPC で叩く」基本を学んだ。第20回はその次のステップ:

MCP サーバが提供する Tool を、LangGraph の create_react_agent から透過的に呼べるようにする。

ここで肝になるのは、MCP の Tool と LangChain の StructuredTool は別物だということ。MCP の Tool は「JSON-RPC 経由でサブプロセスを叩く非同期 RPC」であり、LangGraph の ReAct ループは「同期関数として呼べる BaseTool の配列」を期待する。この 2 つの世界をつなぐアダプタ層が mcp_manager.py の役割で、本回の最大の学びどころ。

サンプルアプリは「Tavily で Web 検索 → 信頼性スコアと要約をつけて SQLite に保存 → 過去のナレッジを SQL で検索」するナレッジエージェント。MCP サーバ側に 7 つの Tool（Web 検索 / URL 抽出 / 保存 / 取得 / スキーマ参照 / SELECT 実行）を載せ、LangGraph の ReAct エージェントがそれらをツール呼び出しとして消費する。

全体像¶

20/
├── mcp_config.json              ← どの MCP サーバをどう起動するか
├── langgraph.json               ← LangGraph Studio が graph を発見するエントリポイント
├── src/
│   ├── sd_20/                   ← LangGraph 側（MCPクライアント + エージェント）
│   │   ├── __main__.py          ← CLI エントリ
│   │   ├── agent.py             ← create_react_agent でグラフ組み立て
│   │   ├── mcp_manager.py       ← MCP Tool → LangChain StructuredTool アダプタ ★中心
│   │   ├── state.py             ← トリム付きカスタム State
│   │   └── prompts/system.txt   ← ナレッジエージェントの行動原理
│   └── mcp_servers/             ← MCP サーバ側
│       ├── server.py            ← FastMCP + @mcp.tool() の 7 ツール
│       └── database.py          ← SQLite 操作（永続化）

create_react_agent が prebuilt として内部に組み立てる LangGraph 構造（このサンプルでは StateGraph を手で書いていない代わりに、これが裏で生成される）:

flowchart TD
    START([START]) --> agent
    agent["agent ノード<br/>LLM (Claude Sonnet 4.5) を呼んで<br/>AIMessage を生成"](<../../../"agent ノード<br/>LLM (Claude Sonnet 4.5) を呼んで<br/>AIMessage を生成">)
    agent -->|tool_calls あり| tools
    agent -->|tool_calls なし<br/>(最終回答)| END([END])
    tools["tools ノード<br/>ToolNode が tool_calls を並列実行<br/>→ ToolMessage を messages に追加"](<../../../"tools ノード<br/>ToolNode が tool_calls を並列実行<br/>→ ToolMessage を messages に追加".md>)
    tools --> agent

ノード 2 つ + 条件分岐 1 本だけのシンプルな循環。これが ReAct ループ（思考 → 行動 → 観察 → 思考 …）の最小実装
第07/08/09/12回のように StateGraph().add_node(...) を明示的に書かなくても、このグラフが create_react_agent の 1 行で構築される
自分のコードで可視化したいときは print(graph.get_graph().draw_mermaid()) で同じ構造を吐ける

ランタイムの動的フロー（CLI 起動時）:

sequenceDiagram
    participant CLI as __main__.py
    participant Agent as create_react_agent(graph)
    participant Mgr as mcp_manager
    participant Sub as MCP Server subprocess<br/>(uv run -m src.mcp_servers.server)
    participant Tavily as Tavily API
    participant DB as SQLite (data.db)

    CLI->>Agent: graph.stream(user_input, ...)
    Note over Agent: import 時に create_agent() が走り、<br/>load_all_mcp_tools() で MCP Tool を全部ロード済み

    rect rgb(245,245,220)
    Note over Mgr,Sub: 【ツールロード（import 時 1 回）】
    Mgr->>Sub: subprocess 起動 + initialize
    Mgr->>Sub: list_tools()
    Sub-->>Mgr: [search_web, save_search_result, ...]
    Mgr-->>Agent: List[StructuredTool] （args_schema は MCP Tool の inputSchema を流用）
    end

    loop ReAct ループ
        Agent->>Agent: LLM (Claude Sonnet 4.5) が tool_call を生成
        Agent->>Mgr: tool_func(**kwargs)
        rect rgb(220,235,250)
        Note over Mgr,Sub: 【ツール呼び出しのたびに subprocess を立て直す】
        Mgr->>Sub: subprocess 再起動 + initialize
        Mgr->>Sub: call_tool("search_web", {query: ...})
        Sub->>Tavily: search(query)
        Tavily-->>Sub: results
        Sub-->>Mgr: ToolCallResult
        end
        Mgr-->>Agent: ToolMessage に変換されて messages に追加

        Agent->>Mgr: save_search_result(...)
        Mgr->>Sub: 再起動 + call_tool
        Sub->>DB: INSERT / UPDATE
        DB-->>Sub: result_id
        Sub-->>Mgr: 保存完了メッセージ
        Mgr-->>Agent: ToolMessage
    end

    Agent-->>CLI: 最終 AIMessage（日本語レポート）

⚠️ 重要な落とし穴: 上の図の青色ブロックを見ると、ツール呼び出しのたびに MCP サーバ subprocess を起動し直している。mcp_manager.create_langchain_tool の中で「ツール呼び出し関数」を async with stdio_client(server_params) で包んでいるためで、これは「セッションの寿命 = 1 ツール呼び出しだけ」という設計。当然オーバーヘッドが大きい。後述「現代版に移植するなら」の langchain-mcp-adapters を使えば常駐セッションになる。

使用ライブラリ・原理¶

`mcp.client.stdio` / `ClientSession` — MCP の Python クライアント¶

from mcp.client.stdio import StdioServerParameters, stdio_client
from mcp.client.session import ClientSession

StdioServerParameters(command, args, env): 「このコマンドを subprocess として起動し、stdin/stdout を JSON-RPC 双方向パイプにしてね」という指示書
stdio_client(params): async context manager。入ると subprocess を起動して (read_stream, write_stream) を返す。抜けると subprocess は kill される
ClientSession(read, write): JSON-RPC 2.0 のセッション層。initialize() で capabilities 交換 → list_tools() / call_tool() で MCP プリミティブにアクセス

LSP との対比: LSP クライアント（VSCode）が言語サーバ（pyright）を subprocess で立てて TextDocument を stdin/stdout で投げる構造と完全に同じ。プロトコルが違うだけで形は LSP。

`langchain_core.tools.structured.StructuredTool` — LangChain ツールの「箱」¶

StructuredTool.from_function(func, name, description, args_schema) で「Python 関数を LangChain ツール化」する API。args_schema には Pydantic モデルまたは JSON Schema dictを渡せて、LLM はこれを見て「どんな引数で呼べばいいか」を判断する。

本サンプルでは args_schema=tool_item.inputSchema のように MCP Tool が宣言した JSON Schema をそのまま流用している。@mcp.tool() デコレータが Python の型ヒントから自動生成した JSON Schema が、ここで LangChain に渡る形。

`create_react_agent`（LangGraph prebuilt）¶

第11回でも出てきた「ReAct パターンの定型グラフ」。今回新しいのは:

state_schema=CustomAgentState で 既定の state を上書き（メッセージ蓄積をトリム付きに）
checkpointer=MemorySaver() で 会話の thread_id ごとに state を保存（__main__.py で uuid.uuid4() を thread_id にしている）

`langgraph.managed.IsLastStep` / `RemainingSteps`¶

CustomAgentState の不思議なフィールド:

class CustomAgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_and_trim_messages]
    is_last_step: IsLastStep
    remaining_steps: RemainingSteps

IsLastStep と RemainingSteps は LangGraph の managed values（フレームワークがランタイムに自動注入する特殊な型）。create_react_agent が「再帰上限（recursion_limit）に達しそうか」を判定するために内部で参照する。ユーザコードでは触らないが、state_schema をカスタム定義するときは型に含めないと create_react_agent が動かない。

`trim_messages` + `count_tokens_approximately`¶

長い会話で context が肥大化するのを防ぐ仕組み。add_messages で結合した後、最新メッセージから貪欲に詰めて 128k トークン以内に切り詰める。strategy="last" は「最新優先」、include_system=True は「SystemMessage は必ず残す」の意味。

ファイル別の役割¶

ファイル	役割
`mcp_config.json`	MCP サーバの起動コマンド定義（`mcpServers` キーが Claude Desktop / Cursor 設定と互換）
`langgraph.json`	LangGraph Studio / `langgraph dev` が `graph` シンボルを探す場所を指定
`src/sd_20/__main__.py`	CLI から `uv run -m src.sd_20 "質問"` で起動。`thread_id` を毎回新規発行 → MemorySaver は実質「単発実行」になっている
`src/sd_20/agent.py`	起動時に `load_all_mcp_tools()` を同期実行 → ツール一覧をプロンプトに埋めて `create_react_agent` を組む
`src/sd_20/mcp_manager.py`	本回の中核。`mcp_config.json` を読む → 各サーバ subprocess を立てて `list_tools()` → 各ツールを `StructuredTool` 化
`src/sd_20/state.py`	`add_and_trim_messages` で「結合 + 128k トリム」を reducer 化したカスタム State
`src/sd_20/prompts/system.txt`	行動原理（信頼性スコアの基準 0.0-1.0、検索クエリの組み立て方など）と `{tool_descriptions}` / `{current_date}` の差し込み口
`src/mcp_servers/server.py`	`FastMCP` に Tavily 連携の `search_web` / `extract_urls` と SQLite 操作の 5 ツールを `@mcp.tool()` で登録
`src/mcp_servers/database.py`	SQLite の CRUD。`init_database()` がモジュールロード時に走り、`search_results` テーブルとインデックスを冪等に作る

行レベルの工夫（中核ロジックの抜粋）¶

① `mcp_manager.create_langchain_tool` — MCP Tool を LangChain Tool に変換する核心 (`mcp_manager.py:86-123`)¶

async def create_langchain_tool(
    tool_name, tool_desc, prefix, server_name, server_params, tool_item,
) -> StructuredTool:
    full_tool_name = f"{prefix}{tool_name}"                                  # ①
    full_tool_desc = f"[{server_name}] {tool_desc}" if server_name else tool_desc

    # 非同期の MCP 呼び出し関数を定義
    async def call_mcp_tool_async(**kwargs: Any) -> Any:                     # ②
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                result = await session.call_tool(tool_name, arguments=kwargs)
                return result

    # 同期呼び出し用にラップする
    def tool_func(**kwargs: Any) -> Any:                                     # ③
        return asyncio.run(call_mcp_tool_async(**kwargs))

    return StructuredTool.from_function(                                     # ④
        func=tool_func,
        name=full_tool_name,
        description=full_tool_desc,
        args_schema=tool_item.inputSchema,
    )

行	やってること	なぜそうする
①	`prefix` でツール名に `knowledge-db__search_web` のように名前空間を付ける	複数 MCP サーバを束ねるとき同名ツールが衝突しないよう、サーバ名でスコープを切る
②	`call_mcp_tool_async` は「呼ばれるたびに subprocess を起動 → initialize → call_tool → 後始末」	`async with stdio_client` を抜けると subprocess が落ちる仕様。セッションを使い回さず、毎回新規起動しているのが本サンプルの単純化ポイント
③	`asyncio.run(call_mcp_tool_async(...))` で同期関数化	`create_react_agent` のツールノードは同期 `BaseTool.invoke` を呼ぶ。`async` のままだと使えないので、外側で `asyncio.run` を挟んでイベントループを毎回回す
④	`args_schema=tool_item.inputSchema`	MCP Tool が公開している JSON Schema をそのまま LangChain に渡すことで、Pydantic モデルを書き直す手間を消している（`from_function` が dict 形式の JSON Schema を受け付ける）

ハマりどころ: ③ の asyncio.run は「現在のスレッドにイベントループがないこと」が前提。create_react_agent を await graph.ainvoke(...) で非同期実行している環境（=既にループが回っている）に持ち込むと RuntimeError: asyncio.run() cannot be called from a running event loop で落ちる。今回の __main__.py は同期 graph.stream() だから動く。

② `agent.py` — グラフ組み立て (`agent.py:15-57`)¶

def create_agent():
    tools = asyncio.run(load_all_mcp_tools())                                # ①
    tool_descriptions = "\n\n".join(
        [f"### {tool.name}\n{tool.description}" for tool in tools]
    )
    current_date = datetime.now().strftime("%Y年%m月%d日")

    with open("src/sd_20/prompts/system.txt", "r") as f:
        prompt = f.read()
    prompt = prompt.format(                                                  # ②
        tool_descriptions=tool_descriptions,
        current_date=current_date,
    )

    model = ChatAnthropic(
        model_name="claude-3-7-sonnet-20250219",
        timeout=None, stop=None, max_tokens=4_096,
    )
    graph = create_react_agent(                                              # ③
        model, tools=tools, prompt=prompt,
        state_schema=CustomAgentState,
        checkpointer=MemorySaver(),
    )
    return graph

graph = create_agent()                                                       # ④

行	やってること	なぜそうする
①	起動時に MCP ツールを 1 回ロード	ツール一覧を確定させてプロンプトに埋め込みたいので、ここで同期的に取りに行く
②	`system.txt` 内の `{tool_descriptions}` を実際のツール一覧で差し替え	LLM に「使えるツールは何か」をプロンプトに書き下すことで、ReAct の思考精度を上げる（OpenAI Tools spec を信用せず冗長化する古典テク）
③	`state_schema=CustomAgentState` で reducer をトリム付きに差し替え	長期会話で context が爆発しないようにする。`create_react_agent` の既定 state は単純な `add_messages`
④	モジュールトップで `graph` を実体化している	`langgraph.json` の `"agent": "./src/sd_20/agent.py:graph"` が指すシンボルになるため。LangGraph Studio がこの `graph` を import するタイミングで MCP サーバが起動する点に注意（import 副作用が大きい）

③ `state.py` — トリム付き reducer (`state.py:11-44`)¶

MAX_TOKENS = 128_000

def add_and_trim_messages(left_messages, right_messages) -> Messages:
    combined_messages = add_messages(left_messages, right_messages)          # ①
    trimmed_messages = trim_messages(                                        # ②
        combined_messages,
        max_tokens=MAX_TOKENS,
        token_counter=count_tokens_approximately,
        strategy="last",
        include_system=True,
    )
    return trimmed_messages

class CustomAgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_and_trim_messages]        # ③
    is_last_step: IsLastStep
    remaining_steps: RemainingSteps

行	やってること	なぜそうする
①	既存メッセージリストに新規メッセージを追加（LangGraph 標準の `add_messages` を内部利用）	tool_call と tool_result のペアリングなど、LangGraph 固有のメッセージマージ規則を再実装しないため
②	結合後に最新優先で 128k トークンに収まるよう削る	Claude 3.7 Sonnet の context は 200k だが、Web 検索結果が長くなりがちなので保険として早めに切る
③	`Annotated[..., add_and_trim_messages]` の構文が「LangGraph の reducer 指定」	`Annotated` の第2引数を LangGraph が読み取り、「このフィールドへの書き込み = この関数で merge」と解釈する。LangGraph 特有の慣習なので初見は驚く

学んだこと（要点）¶

MCP × LangGraph 統合は「ツールアダプタ層」が肝。mcp.types.Tool と langchain.tools.StructuredTool は別物で、間に「JSON-RPC 呼び出しを Python 関数に見せかける」薄いラッパが必要
第19回との違いは「ホストが Cursor / Claude Desktop ではなく自作のクライアント」になったこと。stdio_client を直接叩くと、第19回で見たプロトコルが Python から覗ける
本サンプルは毎回 subprocess 起動という富豪実装で、production では遅すぎる。langchain-mcp-adapters を使うとセッションを使い回せる（後述）
create_react_agent は state_schema を差し替えられる柔軟性がある。が、IsLastStep / RemainingSteps は必ず含めないと動かない（managed values の仕様）
MCP サーバ側は @mcp.tool() デコレータを付けるだけで JSON Schema 自動生成 → クライアント側に inputSchema として渡る。Python の型ヒントが API スキーマになる
mcp_config.json の形式は Claude Desktop と互換なので、同じ MCP サーバを Claude Desktop と LangGraph 両方から使える設計
検索 + 保存 + 検索の往復をエージェントに任せると、プロンプトで「先に DB を見てから web 検索しろ」と書かないと毎回新規検索になる（system.txt の「3.1 既存情報の確認」がそれ）

拡張アイデア¶

MCP サーバを 2 個に増やす — 既存の knowledge-db に加えて、別途 filesystem MCP（公式実装 @modelcontextprotocol/server-filesystem）を mcp_config.json に追加し、load_all_mcp_tools がきちんと両方束ねられることを確認する
subprocess 使い回し化 — mcp_manager.py を改造し、ClientSession をモジュール初期化時に 1 回だけ起動して使い回す。AsyncExitStack で寿命管理しないと resource leak するので、その学びも込みで
langchain-mcp-adapters に置き換え — LangChain 公式の MCP アダプタ（pip install langchain-mcp-adapters）に差し替えると、本サンプルの mcp_manager.py ≒ 200 行が 10 行で済む。差分を比較すれば「フレームワークが何を肩代わりしているか」が分かる
MemorySaver → SqliteSaver — 現状は in-memory で thread_id を毎回 uuid 生成しているので「会話継続」ができていない。from langgraph.checkpoint.sqlite import SqliteSaver に置き換えて、CLI に --thread-id 引数を足す
信頼性スコアの自動評価 — 現状は LLM がプロンプトの基準を見て主観で 0.0-1.0 を付けているだけ。LLM-as-a-Judge スタイルで「保存前にもう一度別のプロンプトで再評価」する evaluate_reliability ツールを追加
SELECT 以外の禁止を強化 — database.execute_select_query は startswith("select") だけで判定しているので、SELECT ... ; DROP TABLE ... の複文を弾けない。sqlite3 の executescript を避けつつ ; を含む文を拒否する処理を追加

現代版に移植するなら¶

1. `langchain-mcp-adapters` を使えば `mcp_manager.py` は不要になる¶

LangChain 公式が 2025 年に出した langchain-mcp-adapters を使うと、本サンプルの mcp_manager.py ≒ 200 行が以下に縮む:

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({
    "knowledge-db": {
        "command": "uv",
        "args": ["run", "-m", "src.mcp_servers.server"],
        "transport": "stdio",
    }
})
tools = await client.get_tools()  # subprocess は session 内で使い回される

ポイントは MultiServerMCPClient がセッション寿命を管理してくれる点で、本サンプルのように「ツール呼び出しのたびに subprocess を立て直す」無駄が消える。

2. `prompt=` 文字列より `ChatPromptTemplate` のほうが安全¶

現状の prompt = prompt.format(...) は普通の Python の str.format なので、ツール説明文に { が混ざるとクラッシュする。ChatPromptTemplate.from_messages([...]) + MessagesPlaceholder("messages") で組むほうが堅い。

3. `claude-3-7-sonnet-20250219` → 最新モデルへ¶

2026 年現在、Claude 4.x 系（claude-sonnet-4-6 等）が利用可能。model_name を差し替えるだけで動く（API は互換）。max_tokens=4_096 も Sonnet 4.x なら 8192 まで上げて余裕を持たせる選択肢あり。

4. `MemorySaver` + 毎回新規 `uuid` → 会話継続が効いていない¶

__main__.py で thread_id = uuid.uuid4() を CLI 実行ごとに発行しているので、せっかくの MemorySaver が単発実行扱いになっている。CLI に --thread-id を追加し、SqliteSaver に差し替えると永続会話になる。

5. プリント文 → `logging`¶

mcp_manager.py は print() まみれで stdio 通信と print が同じストリームに乗ると MCP の JSON-RPC を壊す可能性がある（サーバ側 print は特に危険）。logging モジュールで stderr に出すよう統一すべき。サーバ側 server.py も print("MCPサーバーの初期化を開始します...") を stdout に書いており、SDK の挙動次第で破綻する余地あり。

既知の不具合・注意点¶

asyncio.run 多重起動問題: 本サンプルは CLI 同期実行を前提にしているので動くが、Jupyter Notebook や FastAPI といった既にイベントループが回っている環境に組み込むと RuntimeError: asyncio.run() cannot be called from a running event loop で落ちる
MCP サーバの stdout への print: server.py の print("MCPサーバーの初期化を開始します...") は stdio トランスポートの JSON-RPC ストリームを汚染する可能性がある。今は MCP SDK が tolerant に動いているが、本来 stderr に出すべき
get_recent_results の SQL バインド: params.append(str(limit)) で LIMIT 値を文字列化している。SQLite は受けてくれるが、本来は int のまま渡すべき（他 DB だと型エラーになる）
tavily_client = TavilyClient(api_key=None) が許される問題: TAVILY_API_KEY 未設定でも TavilyClient(api_key=None) が初期化できてしまい、実際に search_web を叩いた時点で初めて 401 が返る。if not TAVILY_API_KEY: raise で fail-fast したほうが良い
with open("src/sd_20/prompts/system.txt", ...) が相対パス: CWD がリポジトリルートでないと FileNotFoundError。Path(__file__).parent / "prompts" / "system.txt" のほうが安全

記事参照¶

Software Design 2025 年 5 月号（推定）連載第20回「MCP と LangGraph の統合」
第19回 STUDY_NOTES: 19/STUDY_NOTES.md — MCP の Tool / Prompt / Resource の 3 プリミティブを最初に解説した回
第11回 STUDY_NOTES: 11/STUDY_NOTES.md — create_react_agent 単体の解説
公式 MCP Python SDK: https://github.com/modelcontextprotocol/python-sdk
LangChain MCP アダプタ: https://github.com/langchain-ai/langchain-mcp-adapters

作成: 2026-05-25 ／最終更新: 2026-06-10

STUDY NOTES

第20回: MCP × LangGraph — create_react_agent から MCP サーバの Tool を呼ぶナレッジエージェント¶

全体像¶

使用ライブラリ・原理¶

mcp.client.stdio / ClientSession — MCP の Python クライアント¶

langchain_core.tools.structured.StructuredTool — LangChain ツールの「箱」¶

create_react_agent（LangGraph prebuilt）¶

langgraph.managed.IsLastStep / RemainingSteps¶

trim_messages + count_tokens_approximately¶