OpenAIエージェントSDKをMCPサーバーに接続する方法

OpenAI Agents SDKは、エージェント機能を備えたAIアプリケーションを構築するための強力な方法を提供します。その主な機能の1つは、アプリケーションが大規模言語モデル（LLM）にコンテキストとツールを提供する方法を標準化するモデルコンテキストプロトコル（MCP）をサポートしていることです。この記事では、OpenAI Agent SDKをMCPサーバーに接続する方法を、詳細な手順とサンプルコードを使用してガイドします。

モデルコンテキストプロトコル（MCP）の理解

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

MCPサーバーの種類

MCP仕様は、トランスポートメカニズムに基づいて2種類のサーバーを定義します：

1. スタンダード入力サーバー：これは、アプリケーションのサブプロセスとして実行され、基本的に「ローカル」として実行されます。
2. SSEサーバー：これはリモートで実行され、HTTP経由でサーバー送信イベント（SSE）を使用してURLを介して接続されます。

OpenAI Agent SDKは、これら2種類のサーバータイプに対応するクラスを提供します：

• MCPServerStdio：ローカルのスタンダード入力サーバー用
• MCPServerSse：リモートSSEサーバー用

前提条件

始める前に、以下を確認してください：

Python 3.10以上がインストールされていること

OpenAI Agent SDKがインストールされていること：

pip install openai-agents

ローカルのスタンダード入力サーバーを使用するには、npx（JavaScriptベースのMCPサーバー用）などの追加ツールが必要になる場合があります。

MCPスタンダード入力サーバーへの接続

まず、ローカルのスタンダード入力MCPサーバーに接続する方法を見てみましょう。ファイルシステムMCPサーバーを例として使います。

ステップ1：必要な依存関係のインストール

ファイルシステムMCPサーバーを使用する予定がある場合、Node.jsとNPXが必要です：

# Node.jsをインストールします（すでにインストールされていない場合）
# Ubuntu/Debianの場合
sudo apt update
sudo apt install nodejs npm

# macOSの場合
brew install node

# インストールを確認します
node --version
npx --version

ステップ2：プロジェクト構造の設定

基本的なプロジェクト構造を作成します：

my_agent_project/
├── sample_data/
│   ├── file1.txt
│   └── file2.md
├── main.py
└── README.md

ステップ3：MCPスタンダード入力サーバーに接続する

ここでは、MCPServerStdioを使用してファイルシステムMCPサーバーに接続する完全な例を示します：

import asyncio
import os
import shutil
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerStdio

async def run_agent_with_mcp():
    # サンプルデータ用のディレクトリパスを取得します
    current_dir = os.path.dirname(os.path.abspath(__file__))
    sample_data_dir = os.path.join(current_dir, "sample_data")

    # MCPサーバーを作成して設定します
    async with MCPServerStdio(
        name="ファイルシステムサーバー",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", sample_data_dir],
        },
    ) as mcp_server:
        # MCPサーバーを使用するエージェントを作成します
        agent = Agent(
            name="ファイルアシスタント",
            instructions="ファイルシステムツールを使用して、sample_dataディレクトリ内のファイルを読み取り分析します。",
            mcp_servers=[mcp_server]
        )

        # ユーザーのクエリでエージェントを実行します
        message = "ディレクトリ内のすべてのファイルをリストし、その内容を要約してください。"
        print(f"クエリを実行中: {message}\\\\n")

        # デバッグ用にトレースIDを生成します
        trace_id = gen_trace_id()
        with trace(workflow_name="MCPファイルシステム例", trace_id=trace_id):
            print(f"トレースを表示: <https://platform.openai.com/traces/{trace_id}\\\\n>")
            result = await Runner.run(starting_agent=agent, input=message)
            print(result.final_output)

if __name__ == "__main__":
    # npxがインストールされているか確認します
    if not shutil.which("npx"):
        raise RuntimeError("npxがインストールされていません。`npm install -g npx`でインストールしてください。")

    asyncio.run(run_agent_with_mcp())

この例では：

1. MCPServerStdioインスタンスを作成してファイルシステムMCPサーバーをサブプロセスとして実行します。
2. mcp_serversパラメーターを介してこのサーバーをAgentコンストラクターに渡します。
3. エージェントが実行されると、自動的にMCPサーバーでlist_tools()を呼び出し、LLMに使用可能なツールを認識させます。
4. LLMがMCPツールのいずれかを使用することを決定した場合、SDKはサーバーでcall_tool()を呼び出します。

MCP SSEサーバーへの接続

次に、SSEを使用してリモートMCPサーバーに接続する方法を見てみましょう：

ステップ1：SSE MCPサーバーの理解

SSE（Server-Sent Events）MCPサーバーはリモートで実行され、HTTPエンドポイントを介してその機能を公開します。スタンダード入力サーバーとは異なり、アプリケーションのサブプロセスとしては実行されません。

ステップ2：MCP SSEサーバーに接続する

以下は、MCP SSEサーバーに接続するサンプルコードです：

import asyncio
from agents import Agent, Runner, gen_trace_id, trace
from agents.mcp import MCPServerSse
from agents.model_settings import ModelSettings

async def run_agent_with_remote_mcp():
    # SSE MCPサーバー接続を作成して設定します
    async with MCPServerSse(
        name="天気サービス",
        params={
            "url": "<https://example.com/mcp/sse>",
            # オプションの認証パラメータ
            "headers": {
                "Authorization": "Bearer your_api_key_here"
            }
        },
    ) as mcp_server:
        # リモートMCPサーバーを使用するエージェントを作成します
        agent = Agent(
            name="天気アシスタント",
            instructions="天気ツールを使用して、現在の天気に関する質問に答えます。",
            mcp_servers=[mcp_server],
            # 利用可能な場合はツールを使用するようにエージェントに指示します
            model_settings=ModelSettings(tool_choice="required")
        )

        # ユーザーのクエリでエージェントを実行します
        message = "今日の東京の天気はどうですか？"
        print(f"クエリを実行中: {message}\\\\n")

        trace_id = gen_trace_id()
        with trace(workflow_name="天気MCP例", trace_id=trace_id):
            print(f"トレースを表示: <https://platform.openai.com/traces/{trace_id}\\\\n>")
            result = await Runner.run(starting_agent=agent, input=message)
            print(result.final_output)

if __name__ == "__main__":
    asyncio.run(run_agent_with_remote_mcp())

シンプルなローカルMCP SSEサーバーの作成

MCPがどのように機能するかを完全に理解するために、シンプルなMCPサーバーを実装することが役立ちます。以下は、最小限のMCP SSEサーバーの例です：

import asyncio
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from typing import Dict, Any, List, Optional

app = FastAPI()

# MCPサーバーが提供するツールを定義します
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "add",
            "description": "二つの数字を足します",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "number", "description": "最初の数値"},
                    "b": {"type": "number", "description": "二つ目の数値"}
                },
                "required": ["a", "b"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "特定の場所の天気を取得します",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "都市名"}
                },
                "required": ["location"]
            }
        }
    }
]

# 実際のツール機能を実装します
async def call_tool(tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
    if tool_name == "add":
        return {"result": parameters["a"] + parameters["b"]}

    elif tool_name == "get_weather":
        # 実際の実装では、実際の天気APIを呼び出します
        weather_data = {
            "Tokyo": {"condition": "晴れ", "temperature": 25},
            "New York": {"condition": "曇り", "temperature": 18},
            "London": {"condition": "雨", "temperature": 15}
        }
        location = parameters["location"]
        if location in weather_data:
            return {"weather": weather_data[location]}
        return {"error": f"{location}の天気データは利用できません"}

    return {"error": f"不明なツール: {tool_name}"}

async def sse_event_generator(request: Request):
    # リクエストボディを読み取ります
    body_bytes = await request.body()
    body = json.loads(body_bytes)

    # 異なるMCP操作を処理します
    if body["action"] == "list_tools":
        yield f"data: {json.dumps({'tools': TOOLS})}\\\\n\\\\n"

    elif body["action"] == "call_tool":
        tool_name = body["tool_name"]
        parameters = body["parameters"]
        result = await call_tool(tool_name, parameters)
        yield f"data: {json.dumps({'result': result})}\\\\n\\\\n"

@app.post("/sse")
async def sse_endpoint(request: Request):
    return StreamingResponse(
        sse_event_generator(request),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

高度な機能

ツールリストのキャッシュ

パフォーマンスを向上させるために、MCPサーバーからツールリストをキャッシュできます：

# ツールキャッシュを使用してMCPサーバーを作成します
async with MCPServerSse(
    name="天気サービス",
    params={"url": "<https://example.com/mcp/sse>"},
    cache_tools_list=True  # キャッシングを有効にする
) as mcp_server:
    # 以前と同様にサーバーを使用します...

cache_tools_list=Trueに設定すると、SDKはMCPサーバーでのlist_tools()を1回のみ呼び出し、以降のエージェント実行に結果を再利用します。これにより、特にリモートサーバーでのレイテンシが減少します。

必要に応じてキャッシュを無効にするには：

mcp_server.invalidate_tools_cache()

MCP操作のトレース

OpenAI Agents SDKには、MCP操作を自動的にキャプチャする組み込みトレース機能が含まれています：

1. ツールをリストするためのMCPサーバーへの呼び出し
2. 関数呼び出しに関するMCP関連の情報

上記の例のようにtraceコンテキストマネージャを使用すると、これらのトレースを

https://platform.openai.com/traces/で表示できます。

複数のMCPサーバーの使用

エージェントを複数のMCPサーバーに同時に接続することで、より広範なツールにアクセスできます：

async def run_with_multiple_servers():
    async with MCPServerStdio(
        name="ファイルシステム",
        params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./data"]}
    ) as fs_server, MCPServerSse(
        name="天気API",
        params={"url": "<https://example.com/weather/mcp/sse>"}
    ) as weather_server:
        # 両方のMCPサーバーを持つエージェントを作成します
        agent = Agent(
            name="マルチツールアシスタント",
            instructions="利用可能なすべてのツールを使用してユーザーを支援します。",
            mcp_servers=[fs_server, weather_server]
        )

        # エージェントを実行します
        result = await Runner.run(
            starting_agent=agent,
            input="まず東京の天気をチェックし、その後report.txtファイルの内容を読み取ります。"
        )
        print(result.final_output)

エラーハンドリングとデバッグ

MCPサーバーを使用する際に、さまざまな問題に直面する可能性があります。以下は、一般的な問題とその対処方法です：

接続の問題

MCPサーバーが応答していない場合：

try:
    async with MCPServerSse(
        name="天気サービス",
        params={"url": "<https://example.com/mcp/sse>"}
    ) as mcp_server:
        # サーバーを使用します...
except Exception as e:
    print(f"MCPサーバーへの接続に失敗しました: {e}")
    # フォールバック戦略を実装します

ツール実行エラー

ツールの実行が失敗した場合は、優雅に処理します：

try:
    result = await Runner.run(starting_agent=agent, input=user_query)
    print(result.final_output)
except Exception as e:
    print(f"エージェントの実行中にエラーが発生しました: {e}")
    # 詳細なエラー情報を確認するためにトレースログをチェックすることをお勧めします

結論

OpenAI Agents SDKのMCPサポートにより、さまざまなツールと機能を備えたエージェントを拡張できます。ローカルのスタンダード入力サーバーを使用している場合でも、リモートSSEエンドポイントを使用している場合でも、統合は簡単かつ強力です。

MCPサーバーに接続することで、エージェントはファイルシステム、天気API、データベース、およびMCPインターフェースを公開するほぼすべての外部ツールやデータソースにアクセスできます。この柔軟性により、OpenAI Agent SDKは洗練されたAIアプリケーションを構築するための強力な基盤となります。

パフォーマンスを最適化するためにツールキャッシュ機能を活用し、組み込みのトレース機能を使用して、MCPサーバーとのエージェントの相互作用をデバッグおよび監視することを忘れないでください。

OpenAI AgentsとMCPで楽しくコーディングしてください！