コンテンツにスキップ

Model context protocol (MCP)

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

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

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

MCP 統合の選択

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

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

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

1. Hosted MCP server tools

Hosted ツールは、ツールの往復処理全体を OpenAI のインフラに移します。あなたのコードがツールを列挙・呼び出す代わりに、HostedMCPTool がサーバーラベル(および任意のコネクタメタデータ)を Responses API に転送します。モデルはリモートサーバーのツールを列挙し、あなたの Python プロセスへの追加コールバックなしにそれらを呼び出します。Hosted ツールは現在、Responses API のホスト型 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())

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

ストリーミングによる hosted MCP の結果取得

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

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

コネクタを裏付けとする hosted サーバー

Hosted 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",
    }
)

ストリーミング、承認、コネクタを含む完全な 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_attemptsretry_backoff_seconds_base は、list_tools()call_tool() に自動リトライを追加します。
  • tool_filter は公開するツールをサブセットに制限できます(ツールのフィルタリング を参照)。

3. HTTP with SSE MCP サーバー

Warning

MCP プロジェクトは Server‑Sent Events トランスポートを非推奨化しました。新しい統合には Streamable HTTP または stdio を推奨し、SSE はレガシーサーバーにのみ残してください。

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() を呼び出します。

トレーシング

Tracing は MCP のアクティビティを自動的に捕捉します。含まれるもの:

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

MCP Tracing Screenshot

参考資料