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 サーバーを呼び出させる	`HostedMCPTool` 経由のホスト型 MCP サーバーツール
ローカルまたはリモートで実行する Streamable HTTP サーバーに接続する	`MCPServerStreamableHttp` 経由の Streamable HTTP MCP サーバー
Server-Sent Events を用いた HTTP を実装しているサーバーと通信する	`MCPServerSse` 経由の SSE 付き HTTP MCP サーバー
ローカルプロセスを起動し、stdin/stdout 経由で通信する	`MCPServerStdio` 経由の stdio MCP サーバー

以下のセクションでは、各オプション、設定方法、あるトランスポートを別のトランスポートより優先すべき場合について説明します。

エージェントレベルの MCP 設定

トランスポートの選択に加えて、Agent.mcp_config を設定することで、MCP ツールの準備方法を調整できます。

from agents import Agent

agent = Agent(
    name="Assistant",
    mcp_servers=[server],
    mcp_config={
        # Try to convert MCP tool schemas to strict JSON schema.
        "convert_schemas_to_strict": True,
        # If None, MCP tool failures are raised as exceptions instead of
        # returning model-visible error text.
        "failure_error_function": None,
        # Prefix local MCP tool names with their server name.
        "include_server_in_tool_names": True,
    },
)

注:

convert_schemas_to_strict はベストエフォートです。スキーマを変換できない場合は、元のスキーマが使用されます。
failure_error_function は、MCP ツール呼び出しの失敗をモデルにどのように提示するかを制御します。
failure_error_function が未設定の場合、SDK はデフォルトのツールエラーフォーマッターを使用します。
サーバーレベルの failure_error_function は、そのサーバーについて Agent.mcp_config["failure_error_function"] を上書きします。
include_server_in_tool_names はオプトインです。有効にすると、各ローカル MCP ツールは、決定的なサーバープレフィックス付きの名前でモデルに公開されます。これにより、複数の MCP サーバーが同じ名前のツールを公開している場合の衝突を避けやすくなります。生成される名前は ASCII セーフで、関数ツール名の長さ制限内に収まり、同じエージェント上の既存のローカル関数ツール名と有効なハンドオフ名を避けます。SDK は引き続き、元のサーバー上で元の MCP ツール名を呼び出します。

トランスポート間で共通するパターン

トランスポートを選択した後、多くの統合では同じような追加判断が必要です。

ツールのサブセットのみを公開する方法（ツールフィルタリング）。
サーバーが再利用可能なプロンプトも提供するかどうか（プロンプト）。
list_tools() をキャッシュすべきかどうか（キャッシュ）。
MCP アクティビティがトレースにどのように表示されるか（トレーシング）。

ローカル MCP サーバー（MCPServerStdio、MCPServerSse、MCPServerStreamableHttp）では、承認ポリシーと呼び出しごとの _meta ペイロードも共通の概念です。Streamable HTTP セクションでは最も完全な例を示しており、同じパターンは他のローカルトランスポートにも適用できます。

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

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

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

HostedMCPTool をエージェントの tools リストに追加して、ホスト型ツールを作成します。tool_config dict は、REST API に送信する JSON と同じ構造です。

import asyncio

from agents import Agent, HostedMCPTool, Runner

async def main() -> None:
    agent = Agent(
        name="Assistant",
        instructions="Use the DeepWiki hosted MCP server to inspect openai/openai-agents-python.",
        tools=[
            HostedMCPTool(
                tool_config={
                    "type": "mcp",
                    "server_label": "deepwiki",
                    "server_url": "https://mcp.deepwiki.com/mcp",
                    "require_approval": "never",
                }
            )
        ],
    )

    result = await Runner.run(
        agent,
        "Which language is the repository openai/openai-agents-python written in?",
    )
    print(result.final_output)

asyncio.run(main())

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

ホスト型ツール検索でホスト型 MCP サーバーを遅延読み込みしたい場合は、tool_config["defer_loading"] = True を設定し、ToolSearchTool をエージェントに追加します。これは OpenAI Responses モデルでのみサポートされます。ツール検索の完全な設定と制約については、ツールを参照してください。

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

ホスト型ツールは、関数ツールとまったく同じ方法で実行結果のストリーミングをサポートします。Runner.run_streamed を使用して、モデルがまだ動作している間に、増分 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_wiki_structure", "read_wiki_contents", "ask_question"}

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": "deepwiki",
                "server_url": "https://mcp.deepwiki.com/mcp",
                "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_content は、テキスト出力より tool_result.structured_content を優先するかどうかを切り替えます。
max_retry_attempts と retry_backoff_seconds_base は、list_tools() と call_tool() に自動再試行を追加します。
tool_filter により、ツールのサブセットのみを公開できます（ツールフィルタリングを参照）。
require_approval は、ローカル MCP ツールで人間参加型の承認ポリシーを有効にします。
failure_error_function は、モデルに見える MCP ツール失敗メッセージをカスタマイズします。代わりにエラーを発生させるには、None に設定します。
tool_meta_resolver は、call_tool() の前に呼び出しごとの MCP _meta ペイロードを注入します。

ローカル MCP サーバーの承認ポリシー

MCPServerStdio、MCPServerSse、MCPServerStreamableHttp はすべて require_approval を受け付けます。

サポートされる形式:

すべてのツールに対する "always" または "never"。
True / False（always/never と同等）。
ツールごとのマップ。例: {"delete_file": "always", "read_file": "never"}。
グループ化されたオブジェクト: {"always": {"tool_names": [...]}, "never": {"tool_names": [...]}}。

async with MCPServerStreamableHttp(
    name="Filesystem MCP",
    params={"url": "http://localhost:8000/mcp"},
    require_approval={"always": {"tool_names": ["delete_file"]}},
) as server:
    ...

完全な一時停止/再開フローについては、人間参加型および examples/mcp/get_all_mcp_tools_example/main.py を参照してください。

`tool_meta_resolver` による呼び出しごとのメタデータ

MCP サーバーが _meta 内にリクエストメタデータ（たとえば、テナント ID やトレースコンテキスト）を想定している場合は、tool_meta_resolver を使用します。以下の例では、Runner.run(...) に context として dict を渡すことを想定しています。

from agents.mcp import MCPServerStreamableHttp, MCPToolMetaContext


def resolve_meta(context: MCPToolMetaContext) -> dict[str, str] | None:
    run_context_data = context.run_context.context or {}
    tenant_id = run_context_data.get("tenant_id")
    if tenant_id is None:
        return None
    return {"tenant_id": str(tenant_id), "source": "agents-sdk"}


server = MCPServerStreamableHttp(
    name="Metadata-aware MCP",
    params={"url": "http://localhost:8000/mcp"},
    tool_meta_resolver=resolve_meta,
)

実行コンテキストが Pydantic モデル、dataclass、またはカスタムクラスの場合は、属性アクセスでテナント ID を読み取ってください。

MCP ツール出力: テキストと画像

MCP ツールが画像コンテンツを返すと、SDK はそれを画像ツール出力エントリーに自動的にマッピングします。テキスト/画像が混在するレスポンスは出力項目のリストとして転送されるため、エージェントは通常の関数ツールからの画像出力を利用するのと同じ方法で MCP 画像実行結果を利用できます。

3. SSE 付き HTTP MCP サーバー

Warning

MCP プロジェクトでは Server-Sent Events トランスポートが非推奨になりました。新しい統合には Streamable HTTP または stdio を優先し、SSE はレガシーサーバー向けにのみ維持してください。

MCP サーバーが SSE 付き HTTP トランスポートを実装している場合は、 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)

5. MCP サーバーマネージャー

複数の MCP サーバーがある場合は、MCPServerManager を使用して事前に接続し、接続済みのサブセットをエージェントに公開します。コンストラクターオプションと再接続の動作については、MCPServerManager API リファレンスを参照してください。

from agents import Agent, Runner
from agents.mcp import MCPServerManager, MCPServerStreamableHttp

servers = [
    MCPServerStreamableHttp(name="calendar", params={"url": "http://localhost:8000/mcp"}),
    MCPServerStreamableHttp(name="docs", params={"url": "http://localhost:8001/mcp"}),
]

async with MCPServerManager(servers) as manager:
    agent = Agent(
        name="Assistant",
        instructions="Use MCP tools when they help.",
        mcp_servers=manager.active_servers,
    )
    result = await Runner.run(agent, "Which MCP tools are available?")
    print(result.final_output)

主な動作:

active_servers には、drop_failed_servers=True（デフォルト）の場合、正常に接続されたサーバーのみが含まれます。
失敗は failed_servers と errors で追跡されます。
最初の接続失敗で例外を発生させるには、strict=True を設定します。
失敗したサーバーを再試行するには reconnect(failed_only=True) を呼び出し、すべてのサーバーを再起動するには reconnect(failed_only=False) を呼び出します。
ライフサイクルの動作を調整するには、connect_timeout_seconds、cleanup_timeout_seconds、connect_in_parallel を使用します。

共通のサーバー機能

以下のセクションは、MCP サーバートランスポート全体に適用されます（正確な API サーフェスはサーバークラスによって異なります）。

ツールフィルタリング

各 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_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 サーバーは、エージェントの指示を動的に生成するプロンプトも提供できます。プロンプトをサポートするサーバーは、次の 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 アクティビティを自動的にキャプチャします。

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

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

参考資料

Model Context Protocol – 仕様と設計ガイド。
examples/mcp – 実行可能な stdio、SSE、Streamable HTTP サンプル。
examples/hosted_mcp – 承認とコネクターを含む完全なホスト型 MCP デモ。