コンテンツにスキップ

Model context protocol (MCP)

Model context protocol (MCP) は、アプリケーションがツールとコンテキストを言語モデルへ公開する方法を標準化します。公式ドキュメントより:

MCP is an open protocol that standardizes how applications provide context to LLMs. Think of MCP like a USB-C port for AI applications. Just as USB-C provides a standardized way to connect your devices to various peripherals and accessories, MCP provides a standardized way to connect AI models to different data sources and tools.

Agents Python SDK は複数の MCP トランスポートに対応します。これにより既存の MCP サーバーを再利用したり、独自に構築してファイルシステム、HTTP、またはコネクター対応のツールを エージェント に公開できます。

MCP 統合の選択

MCP サーバーを エージェント に接続する前に、ツール呼び出しをどこで実行するか、どのトランスポートに到達できるかを決めます。以下のマトリクスは Python SDK がサポートするオプションをまとめたものです。

必要なこと 推奨オプション
OpenAI の Responses API に、モデルの代理として外部公開可能な MCP サーバーを呼ばせたい ホスト型 MCP サーバー ツール via HostedMCPTool
ローカルまたはリモートで稼働する Streamable HTTP サーバーに接続したい Streamable HTTP MCP サーバー via MCPServerStreamableHttp
Server-Sent Events を用いた HTTP を実装するサーバーと通信したい HTTP with SSE MCP サーバー via MCPServerSse
ローカルプロセスを起動し、stdin/stdout 経由で通信したい stdio MCP サーバー via MCPServerStdio

以下のセクションでは、それぞれのオプションの設定方法と、どのトランスポートを選ぶべきかの目安を説明します。

1. ホスト型 MCP サーバー ツール

ホスト型ツールは、ツールの往復処理全体を OpenAI のインフラ内で完結させます。あなたのコードがツール一覧取得や呼び出しを行う代わりに、HostedMCPTool は サーバーラベル(および任意のコネクター メタデータ)を Responses API に転送します。モデルはリモートサーバーのツールを一覧し、あなたの Python プロセスへの追加のコールバックなしでそれらを実行します。ホスト型ツールは現在、Responses API のホスト型 MCP 連携をサポートする OpenAI モデルで動作します。

基本的なホスト型 MCP ツール

エージェント の tools リストに HostedMCPTool を追加してホスト型ツールを作成します。tool_config の dict は、REST API に送る JSON をそのまま反映します:

import asyncio

from agents import Agent, HostedMCPTool, Runner

async def main() -> None:
    agent = Agent(
        name="Assistant",
        tools=[
            HostedMCPTool(
                tool_config={
                    "type": "mcp",
                    "server_label": "gitmcp",
                    "server_url": "https://gitmcp.io/openai/codex",
                    "require_approval": "never",
                }
            )
        ],
    )

    result = await Runner.run(agent, "Which language is this repository written in?")
    print(result.final_output)

asyncio.run(main())

ホスト型サーバーはツールを自動的に公開します。mcp_servers に追加する必要はありません。

ストリーミングによるホスト型 MCP の実行結果

ホスト型ツールは 関数ツール とまったく同じ方法で ストリーミング をサポートします。Runner.run_streamedstream=True を渡すと、モデルがまだ動作中でも増分の MCP 出力を消費できます:

result = Runner.run_streamed(agent, "Summarise this repository's top languages")
async for event in result.stream_events():
    if event.type == "run_item_stream_event":
        print(f"Received: {event.item}")
print(result.final_output)

任意の承認フロー

サーバーが機微な操作を実行できる場合、各ツール実行前に人間またはプログラムによる承認を必須にできます。tool_configrequire_approval に、単一のポリシー("always""never")またはツール名からポリシーへの dict を設定します。判断を Python 内で行うには、on_approval_request コールバックを指定します。

from agents import MCPToolApprovalFunctionResult, MCPToolApprovalRequest

SAFE_TOOLS = {"read_project_metadata"}

def approve_tool(request: MCPToolApprovalRequest) -> MCPToolApprovalFunctionResult:
    if request.data.name in SAFE_TOOLS:
        return {"approve": True}
    return {"approve": False, "reason": "Escalate to a human reviewer"}

agent = Agent(
    name="Assistant",
    tools=[
        HostedMCPTool(
            tool_config={
                "type": "mcp",
                "server_label": "gitmcp",
                "server_url": "https://gitmcp.io/openai/codex",
                "require_approval": "always",
            },
            on_approval_request=approve_tool,
        )
    ],
)

コールバックは同期・非同期のどちらでもよく、モデルが実行を継続するために承認データを必要とするたびに呼び出されます。

コネクター対応のホスト型サーバー

ホスト型 MCP は OpenAI コネクターにも対応しています。server_url を指定する代わりに、connector_id とアクセストークンを指定します。Responses API が認証を処理し、ホスト型サーバーはコネクターのツールを公開します。

import os

HostedMCPTool(
    tool_config={
        "type": "mcp",
        "server_label": "google_calendar",
        "connector_id": "connector_googlecalendar",
        "authorization": os.environ["GOOGLE_CALENDAR_AUTHORIZATION"],
        "require_approval": "never",
    }
)

ストリーミング、承認、コネクターを含む完全なホスト型ツールのサンプルは examples/hosted_mcp にあります。

2. Streamable HTTP MCP サーバー

ネットワーク接続を自分で管理したい場合は、MCPServerStreamableHttp を使用します。Streamable HTTP サーバーは、トランスポートを自分で制御したい場合や、サーバーを自社インフラ内で動かしつつレイテンシを低く抑えたい場合に最適です。

import asyncio
import os

from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp
from agents.model_settings import ModelSettings

async def main() -> None:
    token = os.environ["MCP_SERVER_TOKEN"]
    async with MCPServerStreamableHttp(
        name="Streamable HTTP Python Server",
        params={
            "url": "http://localhost:8000/mcp",
            "headers": {"Authorization": f"Bearer {token}"},
            "timeout": 10,
        },
        cache_tools_list=True,
        max_retry_attempts=3,
    ) as server:
        agent = Agent(
            name="Assistant",
            instructions="Use the MCP tools to answer the questions.",
            mcp_servers=[server],
            model_settings=ModelSettings(tool_choice="required"),
        )

        result = await Runner.run(agent, "Add 7 and 22.")
        print(result.final_output)

asyncio.run(main())

コンストラクターは次の追加オプションを受け付けます:

  • client_session_timeout_seconds は HTTP の読み取りタイムアウトを制御します。
  • use_structured_contenttool_result.structured_content をテキスト出力より優先するかどうかを切り替えます。
  • max_retry_attemptsretry_backoff_seconds_baselist_tools()call_tool() に自動リトライを追加します。
  • tool_filter は公開するツールをサブセットに絞り込めます(ツールのフィルタリング を参照)。

3. HTTP with SSE MCP サーバー

MCP サーバーが HTTP with SSE トランスポートを実装している場合は、MCPServerSse をインスタンス化します。トランスポート以外は、API は Streamable HTTP サーバーと同一です。

from agents import Agent, Runner
from agents.model_settings import ModelSettings
from agents.mcp import MCPServerSse

workspace_id = "demo-workspace"

async with MCPServerSse(
    name="SSE Python Server",
    params={
        "url": "http://localhost:8000/sse",
        "headers": {"X-Workspace": workspace_id},
    },
    cache_tools_list=True,
) as server:
    agent = Agent(
        name="Assistant",
        mcp_servers=[server],
        model_settings=ModelSettings(tool_choice="required"),
    )
    result = await Runner.run(agent, "What's the weather in Tokyo?")
    print(result.final_output)

4. stdio MCP サーバー

ローカルのサブプロセスとして動作する MCP サーバーには、MCPServerStdio を使用します。SDK はプロセスを起動し、パイプを開いたまま維持し、コンテキストマネージャの終了時に自動でクローズします。これは、迅速なプロトタイプ作成や、サーバーがコマンドラインのエントリポイントのみを公開している場合に役立ちます。

from pathlib import Path
from agents import Agent, Runner
from agents.mcp import MCPServerStdio

current_dir = Path(__file__).parent
samples_dir = current_dir / "sample_files"

async with MCPServerStdio(
    name="Filesystem Server via npx",
    params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", str(samples_dir)],
    },
) as server:
    agent = Agent(
        name="Assistant",
        instructions="Use the files in the sample directory to answer questions.",
        mcp_servers=[server],
    )
    result = await Runner.run(agent, "List the files available to you.")
    print(result.final_output)

ツールのフィルタリング

各 MCP サーバーはツールフィルターをサポートしており、エージェント に必要な機能だけを公開できます。フィルタリングは、構築時にも、実行ごとの動的制御でも可能です。

静的なツールフィルタリング

create_static_tool_filter を使用して、シンプルな許可/ブロックリストを設定します:

from pathlib import Path

from agents.mcp import MCPServerStdio, create_static_tool_filter

samples_dir = Path("/path/to/files")

filesystem_server = MCPServerStdio(
    params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", str(samples_dir)],
    },
    tool_filter=create_static_tool_filter(allowed_tool_names=["read_file", "write_file"]),
)

allowed_tool_namesblocked_tool_names の両方が指定された場合、SDK はまず許可リストを適用し、その後残った集合からブロック対象を除外します。

動的なツールフィルタリング

より高度なロジックが必要な場合は、ToolFilterContext を受け取る呼び出し可能オブジェクトを渡します。呼び出し可能オブジェクトは同期・非同期いずれも可能で、ツールを公開すべきときに True を返します。

from pathlib import Path

from agents.mcp import MCPServerStdio, ToolFilterContext

samples_dir = Path("/path/to/files")

async def context_aware_filter(context: ToolFilterContext, tool) -> bool:
    if context.agent.name == "Code Reviewer" and tool.name.startswith("danger_"):
        return False
    return True

async with MCPServerStdio(
    params={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", str(samples_dir)],
    },
    tool_filter=context_aware_filter,
) as server:
    ...

フィルターコンテキストは、アクティブな run_context、ツールを要求している agent、および server_name を公開します。

プロンプト

MCP サーバーは、エージェントの instructions を動的に生成する プロンプト も提供できます。プロンプトをサポートするサーバーは次の 2 つのメソッドを公開します:

  • list_prompts() は利用可能なプロンプトテンプレートを列挙します。
  • get_prompt(name, arguments) は、任意のパラメーター付きで具体的なプロンプトを取得します。
from agents import Agent

prompt_result = await server.get_prompt(
    "generate_code_review_instructions",
    {"focus": "security vulnerabilities", "language": "python"},
)
instructions = prompt_result.messages[0].content.text

agent = Agent(
    name="Code Reviewer",
    instructions=instructions,
    mcp_servers=[server],
)

キャッシュ

各 エージェント 実行は、各 MCP サーバーに対して list_tools() を呼び出します。リモートサーバーは顕著なレイテンシを導入しうるため、すべての MCP サーバークラスは cache_tools_list オプションを公開しています。ツール定義が頻繁に変わらないと確信できる場合にのみ True を設定してください。後で新しい一覧を強制するには、サーバーインスタンスで invalidate_tools_cache() を呼び出します。

トレーシング

トレーシング は、以下を含む MCP の活動を自動的に取得します:

  1. ツール一覧のための MCP サーバーへの呼び出し。
  2. ツール呼び出しに関する MCP 関連情報。

MCP トレーシングのスクリーンショット

参考資料