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 サーバーを呼び出させたい | Hosted MCP server tools(HostedMCPTool 経由) |
ローカルまたはリモートで実行している Streamable HTTP サーバーに接続したい | Streamable HTTP MCP servers(MCPServerStreamableHttp 経由) |
Server-Sent Events を用いた HTTP を実装したサーバーとやり取りしたい | HTTP with SSE MCP servers(MCPServerSse 経由) |
ローカルプロセスを起動し stdin/stdout で通信したい | stdio MCP servers(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 connectors にも対応します。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 servers
ネットワーク接続を自分で管理したい場合は、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
は、公開するツールをサブセットに限定できます(Tool filtering を参照)。
3. HTTP with SSE MCP servers
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 servers
ローカルのサブプロセスとして実行する 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_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()
を呼び出します。
トレーシング
Tracing は MCP のアクティビティを自動的に捕捉します。含まれるもの:
- ツールを列挙するための MCP サーバーへの呼び出し。
- ツール呼び出しに関する MCP 関連情報。
参考情報
- Model Context Protocol – 仕様と設計ガイド。
- examples/mcp – 実行可能な stdio、SSE、Streamable HTTP のサンプル。
- examples/hosted_mcp – 承認やコネクタを含む、完全な hosted MCP のデモ。