Model context protocol (MCP)

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

MCP は、アプリケーションが LLMs にコンテキストを提供する方法を標準化するオープンプロトコルです。MCP は AI アプリケーション向けの USB‑C ポートのようなものだと考えてください。USB‑C がさまざまな周辺機器やアクセサリにデバイスを接続するための標準化された方法を提供するのと同様に、MCP は AI モデルをさまざまなデータソースやツールに接続するための標準化された方法を提供します。

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

MCP 統合の選択

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

What you need	Recommended option
Let OpenAI's Responses API call a publicly reachable MCP server on the model's behalf	Hosted MCP server tools via HostedMCPTool
Connect to Streamable HTTP servers that you run locally or remotely	Streamable HTTP MCP servers via MCPServerStreamableHttp
Talk to servers that implement HTTP with Server-Sent Events	HTTP with SSE MCP servers via MCPServerSse
Launch a local process and communicate over stdin/stdout	stdio MCP servers via MCPServerStdio

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

1. Hosted MCP server tools

Hosted ツールは、ツールの往復処理全体を OpenAI のインフラに委ねます。あなたのコードがツールを列挙・呼び出す代わりに、HostedMCPTool が サーバーラベル（および任意のコネクタメタデータ）を Responses API に転送します。モデルはリモートサーバーのツールを列挙し、あなたの Python プロセスへの追加のコールバックなしにそれらを起動します。Hosted ツールは現在、Responses API の hosted MCP 統合をサポートする OpenAI モデルで動作します。

基本的な hosted MCP ツール

エージェントの tools リストに HostedMCPTool を追加して hosted ツールを作成します。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())

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

hosted MCP の結果の ストリーミング

Hosted ツールは、関数ツールとまったく同じ方法で結果の ストリーミング をサポートします。Runner.run_streamed に stream=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_config の require_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,
        )
    ],
)

コールバックは同期または非同期のいずれでも構いません。モデルが実行を続けるために承認データを必要とするたびに呼び出されます。

コネクタで支えられた hosted サーバー

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

import os

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

ストリーミング、承認、コネクタを含む完全に動作する hosted ツールのサンプルは 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_content は、テキスト出力よりも tool_result.structured_content を優先するかどうかを切り替えます。
• max_retry_attempts と retry_backoff_seconds_base は、list_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 を使用して、シンプルな allow/block リストを設定します:

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_names と blocked_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 関連情報。

参考資料

• Model Context Protocol – 仕様および設計ガイド。
• examples/mcp – 実行可能な stdio、SSE、Streamable HTTP のサンプル。
• examples/hosted_mcp – 承認やコネクタを含む完全な hosted MCP のデモ。