エージェントアーキテクチャをクリーンアップしたプロトコル。データサイエンスに向けて

数週間前、データチームの誰かが、複雑なエージェントシステムのツールによって設定されているデータベーススキーマを更新できないか尋ねてきました。更新は簡単です。2 つの新しい列がテーブルに追加されます。

デバイス定義は Agent Orchestrator にありました。これの別の同様のバージョンが検証エージェントに存在していました。 3 番目の少し異なる古いバージョンは、誰かが 3 スプリント前に書いたユーティリティモジュールにありました。人間参加型の承認ロジックはグラフのエッジに直接配線され、デバイスごとに 1 つのカスタム実装が行われました。スキーマを変更するには、4 つのファイルを操作し、各エージェントを個別に再テストし、ダウンストリームで何も問題が発生しないことを祈る必要がありました。

私たちはそれを修正しましたが、深刻な疑問が生じました。 なぜこのようにしたのでしょうか?

正直な答えは、他に選択肢がなかったということです。 Langgraph でのツール呼び出しは、設計上の局所的な問題です。アプライアンスを必要な場所で定義し、必要な場所で呼び出し、すべての配管を所有します。エージェントが 2 人だけの場合は問題ありませんが、7 人のエージェントが 1 つのヒューマンゲートで重複するツールを共有している場合は問題になります。

調査を行った結果、エージェントごとにツールをローカルに定義するのではなく、すべてのツールをホストし、すべてのエージェントがツールを使用できる共有リソースを使用する必要があると判断しました。

この記事では

MCPとは何ですか？?
mcpサーバーを作成する
Stdio と HTTP
それをランググラフに接続する
プロトコルの限界における人間参加型
生産の妨げとなるものは何でしょうか?またその理由は何ですか??
MCP がエージェントシステムに与える影響
結論

MCPとは何ですか？

モデルコンテキストプロトコルは、2024 年後半に Anthropic によって公開されたオープンスタンダードです。これは、AI エージェントがツールを検索して呼び出す方法を標準化します。 Orchestrator 内でツールを定義する代わりに、別のサーバー上でツールを実行します。エージェントは実行時にそのサーバーに接続し、利用可能なデバイスを尋ね、リストを取得します。

この記事を読んだ上級エンジニアはすぐに次のような質問をするでしょう。 一元化されたツールレジストリを作成して、起動時に各エージェントに挿入することはできませんか? 私はこれを自問し、別のシステムで MCP の代わりにツールレジストリを使用しました。

はい、可能です。すでにこのような機能を持っている場合、MCP は緊急事態ではありません。特別なレジストリでは得られないもの 相互運用性の制限。 MCP はプロトコルであり、ライブラリではありません。 MCP 準拠のクライアントはサーバーに接続できます。今日は Langgraph ですが、来年には別のフレームワークになります。 TypeScript クライアントは、追加の統合作業を行わずに Python サーバーを呼び出すことができます。ツールレジストリはこの機能を提供しません。

チーム所有権のポイントもあります。私たちの場合、ML チームがツールを持ち、アプリケーションチームがグラフを持っていました。 MCP は、共有コードベースのないクリーンな契約を彼らに与えました。

mcpサーバーを作成する

MCP サーバーは次の 3 つのことを公開できます。道具 (呼び出し可能なアクション)、 リソース (読み取り専用データ)、および信号 (再利用可能なテンプレート)。何らかのアクションを実行する必要があるエージェントシステムの場合、デバイスが最も重要な問題となります。

Python SDKが同梱されています ファストMCP型シグナルからのスキーマ作成を処理し、プロトコルのライフサイクルを管理します。関数を作成してツールデコレーターで装飾すると、残りはサーバーが処理します。

stdio トランスポートに人々を惹きつける 1 つの点: 決して標準出力に書き込まないでください。 MCP プロトコルは、通信チャネルとして stdout を使用します。あらゆる迷走者 print() この呼び出しにより、デバッグが非常に混乱するような形でメッセージストリームが破損します。

import sys
import logging
from mcp.server.fastmcp import FastMCP

logging.basicConfig(level=logging.INFO, stream=sys.stderr)
logger = logging.getLogger("analyst-tools")

mcp = FastMCP("analyst-tools")

@mcp.tool()
async def run_analysis(code: str, dataset: str) -> dict:
    """
    Executes a Python snippet against live data and returns the result.
    Use when the user wants to compute aggregates, filter records,
    or derive insights. The code must assign its final output to a
    variable named 'output'.
    
    Args:
        code: Python code to execute.
        dataset: One of 'sales', 'inventory', 'pipeline'.
    """
    logger.info(f"run_analysis | dataset={dataset}")
    return await execute_in_sandbox(code, dataset)


@mcp.tool()
async def write_to_db(table: str, payload: dict) -> dict:
    """
    Persists a result record to the analyst results table.
    Only call this after run_analysis has returned a verified output.
    
    Args:
        table: Target table name.
        payload: Key-value pairs to write as a new record.
    """
    logger.info(f"write_to_db | table={table}")
    return await persist_result(table, payload)


if __name__ == "__main__":
    mcp.run(transport="stdio")

docstring は、エージェントがどのツールを呼び出すかを決定するのに役立つために LLM によって使用されます。したがって、適切な docstring を作成することが非常に重要です。

Stdio と HTTP

この決定はすべての実稼働デプロイメントで考慮されますが、ほとんどの記事では省略されています。

スタジオ サーバーをクライアントのサブプロセスとして実行します。通信は標準入出力を介して行われます。遅延は 1 桁のミリ秒で、ネットワークは関係せず、セットアップは最小限で済みます。ローカル開発、単一マシンの展開、またはサーバーとクライアントが同じプロセスツリー内にある場所に最適な選択肢です。。

ストリーミング可能なHTTP サーバーを独立したサービスとして実行します。サーバーを複数のクライアントまたはマシン間で共有する必要がある場合、サーバーをコンテナーとしてデプロイする場合、または水平スケーリングが必要な場合に使用します。ここでは、Cloud Run のようなサーバーレスデプロイメントがうまく機能します。 Stdio は、長時間実行されるネイティブプロセスを前提としているため、サーバーレスモデルにはまったく適合しません。

FastMCP でこれらを切り替えるには、次の 1 行を実行します。

mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)

交通手段を変えるしかない mcp.run() それ以外はすべて同じままです。

データ常駐要件については、外部 API に触れることのないアプライアンスを備えたオンプレミスで実行される MCP サーバーが、コンプライアンスチームに明確なストーリーを提供します。このプロトコルでは、サーバーがどこで実行されるかは関係ありません。

それをランググラフに接続する

langchain-mcp-adapters ライブラリサブプロセスは、ライフサイクルを管理し、ツール検出ハンドシェイクを実行し、MCP ツールスキーマを Langchain 準拠のツールオブジェクトに変換します。

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_google_vertexai import ChatVertexAI

llm = ChatVertexAI(
    model="gemini-2.5-flash",
    temperature=0,
    max_tokens=None
)

async def run(query: str):
    async with MultiServerMCPClient({
        "analyst-tools": {
            "command": "python",
            "args": ["./mcp_server.py"],
            "transport": "stdio",
        }
    }) as client:

        tools = await client.get_tools()
        llm_with_tools = llm.bind_tools(tools)

        def agent_node(state: MessagesState):
            return {"messages": [llm_with_tools.invoke(state["messages"])]}

        graph = StateGraph(MessagesState)
        graph.add_node("agent", agent_node)
        graph.add_node("tools", ToolNode(tools))
        graph.add_edge(START, "agent")
        graph.add_conditional_edges("agent", tools_condition)
        graph.add_edge("tools", "agent")

        app = graph.compile()
        result = await app.ainvoke({
            "messages": [{"role": "user", "content": query}]
        })

        print(result["messages"][-1].content)

tools_condition 最後のメッセージにツール呼び出しが含まれているかどうかを確認する組み込みの LangGraph モジュールがあります。「はい」の場合はツール実行プログラムをルート化し、「いいえ」の場合は完了です。独自のルーティング関数を作成する代わりにこれを使用することは、エッジケースや実装上の間違いに対処できるため、合理的です。

知っておく価値のある行動が 1 つあります。 MultiServerMCPClient デフォルトでは、ツール呼び出しごとに新しい MCP セッションが作成されます。 1 つのリクエストに対して 5 回連続してツールを呼び出すと、5 回のハンドシェイクになります。同じマシン上の標準入出力には問題ありませんが、リモートサーバーとの HTTP トランスポートには注意する価値があります。チェーンされたツール呼び出しを伴う運用ワークロードの場合は、次を使用します。 async with client.session("analyst-tools") 1 つのセッションで複数の通話を固定します。

プロトコルの限界における人間参加型

MCP 以前は、私たちの承認はゲートグラフにありました。私たちは実験しました interrupt_before 特定のノードでは、カスタム検証ロジックがグラフのエッジに配線され、新しい機密デバイスが追加されるたびに UI が更新されました。それは機能しましたが、承認が必要なデバイスの追加には 3 チームの調整が必要であることも意味しました。

MCP の後、ゲートは Langgraph エグゼキュータと MCP クライアントの間の層に移動します。機密性ポリシーに一致するデバイスは、サーバーに到達する前にゲートに到達します。グラフにはこれについての知識がありません。

SENSITIVE_TOOLS = frozenset({"write_to_db", "send_notification", "trigger_webhook"})

async def gated_call(tool_name: str, arguments: dict, execute) -> dict:
    if tool_name in SENSITIVE_TOOLS:
        # In production: push to Slack / internal UI / audit queue
        print(f"\nAPPROVAL REQUIRED {tool_name}")
        print(f"Arguments: {arguments}")
        decision = input("Approve? (y/n): ").strip().lower()

        if decision != "y":
            return {
                "status": "rejected",
                "reason": f"Operator declined '{tool_name}'."
            }

    return await execute(tool_name, arguments)

SENSITIVE_TOOLS これは 1 つのセットであり、どのエージェントがトリガーしたかに関係なく、ツール呼び出しごとに参照されます。新しい機密デバイスがサーバーに追加されましたか?このセットに名前を追加します。グラフは変化しません。承認 UI は変わりません。内部システムでは、起動時に構成ファイルからロードされます。製品チームとコンプライアンスチームは、コードをデプロイしなくても更新できます。

生産を妨げるものは何ですか?またその理由は何ですか?

実行中にサーバーがクラッシュしました。 クライアントは、次回のツール呼び出しでエラーを受け取ります。 Langgraph のツールノードは、これをツールエラーメッセージとして LLM に返します。モデルが回復するか混乱したままになるかは、システムのプロンプトによって異なります。少なくとも、サブプロセスの stderr を個別にログに記録して、このデバッグが推測にならずに、サーバーにどのような影響があったかを確認できるようにします。

LLM は間違ったツールを呼び出します。 MCP はこれからあなたを保護しません。ツールの説明が曖昧であったり、意味が重複している場合、モデルは誤ったルーティング決定を下します。特に不適切な説明が問題を引き起こしたため、サーバー内の docstring の調整に多くの時間を費やしました。 write_to_db 最初に呼ばれること run_analysis もう終わりだった。ツールの説明を簡単なエンジニアリングの問題として扱います。

長時間実行されるワークフローの承認ゲート。 人間がツール呼び出しを承認する必要があり、それに 5 分かかる場合、エージェントグラフは待機状態になります。 Langgraph はチェックポイントによるグラフ状態の維持をサポートしているため、決定に達したときにプロセスを終了して再起動できます。これはここで示すものよりも複雑ですが、スレッドを無期限にブロックできないワークフローには適切なアーキテクチャです。

MCP がエージェントシステムに与える影響

7 つのツールをサーバーに移行し、そのうち 3 つが承認されました。彼らに電話をかけたオーケストレーターは、彼らが何をしているのか全く知りません。

工具の重複を完全に排除しました。 今、 run_analysis 最大 7 つのワークフローを 1 か所で同時に提供することが正確に定義されています。出力スキーマを更新するには、サーバーに変更を加えるだけで済みます。その後、各コンシューマーが変更を受け入れます。

新しい機能の追加がより速くなりました。 たとえば、 generate_visualisation 翌週に機器が到着し、エージェントは翌日からそれを使用しました。オーケストレーターの変更は行われていません。

一方のチームがツールを持ち、もう一方のチームがグラフを持ち、両者の間には明確な契約があることがわかりました。アナリストチームが新しい機能を必要とするときは、アプリケーションチームやグラフチームではなく、サーバーについて ML チームに相談します。

MCP では修正できない点を 1 つ共有したいと思います: : これでは、信頼できないデバイスの信頼性は高まりません。 説明が間違っていると、LLM が適切なルーティング決定を行うのに役立ちません。また、可観測性を置き換えるものではありません。ツール呼び出しをログに記録し、実行パスをトレースする必要があります。機械化が容易な構造ですが、作業は依然としてあなたのものです。