Closed9

OpenAI「Agents SDK」③ツール・MCP

OpenAI
Agent
Agents SDK
kun432kun432

ツール

https://openai.github.io/openai-agents-python/ja/tools/

エージェントに、ユーザのクエリに応じて動的な処理を実行させて、コンテキストを追加するのがツール。ツールは3つの方法がある。

  • ホスト型ツール
    • LLMモデルと同じサーバ上で実行、つまりモデルプロバイダのマネージドなツールと言える。
    • OpenAIの場合は、リトリーバル、Web 検索、コンピュータ操作が提供されている。
  • 関数ツール（Function calling）
    • 任意の Python 関数 をツールとして実行
  • ツールとしての エージェント
    • エージェント をツールとして使用する
    • ハンドオフ せずに他の エージェント を呼び出せる。

ホスト型ツール

OpenAIの場合は以下のツールに対応している

  • WebSearchTool: Web検索。
  • FileSearchTool: OpenAIベクトルストア から情報を取得。
  • ComputerTool: コンピュータ操作 タスクの自動化。
  • CodeInterpreterTool: サンドボックス環境でコードを実行。
  • HostedMCPTool: リモートの MCP サーバ のツールを実行。
  • ImageGenerationTool: プロンプトから画像を生成。
  • LocalShellTool: 同一マシン上でシェルコマンドを実行

WebSearchToolを使用した例

from agents import Agent, Runner, WebSearchTool 
import asyncio

agent = Agent(
    name="エージェント",
    instructions="日本語で回答する。",
    tools=[WebSearchTool()],
)

async def main():
    result = await Runner.run(agent, "神戸の天気は？")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
本日、2025年8月17日（日）の神戸市の天気予報は以下の通りです。

## 〒659-0096 兵庫県芦屋市山手町XX-XXの天候:
現在の状態: おおむね曇り、86°F (30°C)

日次予測:
* 日曜日, 8月 17: 低: 81°F (27°C)、高: 93°F (34°C)、説明: 一部の地域で雷雨
* 月曜日, 8月 18: 低: 81°F (27°C)、高: 95°F (35°C)、説明: 曇りのち薄日が差す
* 火曜日, 8月 19: 低: 81°F (27°C)、高: 94°F (35°C)、説明: 所により晴れ
* 水曜日, 8月 20: 低: 81°F (27°C)、高: 95°F (35°C)、説明: 一部の地域で雷雨
* 木曜日, 8月 21: 低: 81°F (27°C)、高: 95°F (35°C)、説明: 所により晴れ
* 金曜日, 8月 22: 低: 83°F (28°C)、高: 96°F (36°C)、説明: 曇りのち薄日が差す
* 土曜日, 8月 23: 低: 82°F (28°C)、高: 97°F (36°C)、説明: 上層雲からの晴れ間

悪天候アラート:
* 芦屋市: 雷注意報発令中。発表者: 気象庁、、開始時刻: 日曜日, 8月 17, 15:00:00 UTC、終了時刻: 不明


現在、芦屋市には雷注意報が発令されています。雷雨の可能性があるため、外出の際は十分ご注意ください。

関数ツール

Python関数をそのままツールとして使用できる。所定のルールに従えば、Agents SDK が自動でツールをセットアップしてくれる。

  • ツール名は、Python 関数 の名前から設定される（名前を指定することも可能）
  • ツールの説明は、関数の docstring から取得される（説明を指定することも可能）
  • 関数入力のスキーマは関数の引数から自動生成される
  • 関数入力の説明は、（無効化しない限り）関数の docstring から取得される

実際にどのような関数のスキーマが設定されるかを確認してみる。

import json
from typing_extensions import TypedDict, Any
from agents import Agent, FunctionTool, RunContextWrapper, function_tool

class Location(TypedDict):
    lat: float
    long: float

@function_tool  
async def fetch_weather(location: Location) -> str:
    """与えられた場所の天気を取得する。

    Args:
        location: 天気を取得したい場所名
    """
    # 実際には天気APIから取得する
    return "sunny"

# デコレータの引数でツール名や説明などの上書き, docstringのスタイルを定義できる様子
@function_tool(name_override="fetch_data", description_override="与えられたファイルの内容を取得する。" )  
def read_file(ctx: RunContextWrapper[Any], path: str, directory: str | None = None) -> str:
    """ファイルの内容を取得。

    Args:
        path: 読み込むファイルのパス
        directory: 読み込むファイルのディレクトリ
    """
    # 実際にはファイルシステムから読み込む
    return "<file contents>"


agent = Agent(
    name="Assistant",
    tools=[fetch_weather, read_file],  
)

for tool in agent.tools:
    if isinstance(tool, FunctionTool):
        print(tool.name)
        print(tool.description)
        print(json.dumps(tool.params_json_schema, indent=2, ensure_ascii=False))
        print()
出力
fetch_weather
与えられた場所の天気を取得する。
{
  "$defs": {
    "Location": {
      "properties": {
        "lat": {
          "title": "Lat",
          "type": "number"
        },
        "long": {
          "title": "Long",
          "type": "number"
        }
      },
      "required": [
        "lat",
        "long"
      ],
      "title": "Location",
      "type": "object",
      "additionalProperties": false
    }
  },
  "properties": {
    "location": {
      "description": "天気を取得したい場所名",
      "properties": {
        "lat": {
          "title": "Lat",
          "type": "number"
        },
        "long": {
          "title": "Long",
          "type": "number"
        }
      },
      "required": [
        "lat",
        "long"
      ],
      "title": "Location",
      "type": "object",
      "additionalProperties": false
    }
  },
  "required": [
    "location"
  ],
  "title": "fetch_weather_args",
  "type": "object",
  "additionalProperties": false
}

fetch_data
与えられたファイルの内容を取得する。
{
  "properties": {
    "path": {
      "description": "読み込むファイルのパス",
      "title": "Path",
      "type": "string"
    },
    "directory": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "description": "読み込むファイルのディレクトリ",
      "title": "Directory"
    }
  },
  "required": [
    "path",
    "directory"
  ],
  "title": "fetch_data_args",
  "type": "object",
  "additionalProperties": false
}

内部的には

  • inspect モジュールで関数シグネチャを抽出
  • griffe で docstring を解析
  • pydanticでスキーマ生成

を行っているらしい。

カスタム関数ツール

Pythonの関数 をツールとして使わずに、FunctionTool で直接ツールを作成することができる。その場合は以下を個別に指定する。

  • name
  • description
  • params_json_schema: 引数のJSONスキーマ
  • on_invoke_tool: ToolContext と JSON 文字列を引数として受け取って、文字列としてツール出力を返す非同期関数

・・・とあるがいまいちユースケースがわからない。

一応こんな感じなのかな？

from typing import Any
from pydantic import BaseModel
from agents import Agent, RunContextWrapper, FunctionTool, Runner
import asyncio

def do_some_work(data: str) -> str:
    return "done"

class FunctionArgs(BaseModel):
    username: str
    age: int

async def run_function(ctx: RunContextWrapper[Any], args: str) -> str:
    parsed = FunctionArgs.model_validate_json(args)
    return do_some_work(data=f"{parsed.username} は {parsed.age} 歳です。")

# 
tool = FunctionTool(
    name="process_user",
    description="抽出したユーザデータを処理する",
    params_json_schema=FunctionArgs.model_json_schema(),
    on_invoke_tool=run_function,
)

agent = Agent(
    name="Assistant",
    instructions="日本語で回答する。",
    tools=[tool],  
)

async def main():
    result = await Runner.run(agent, "ユーザー名は田中、年齢は25歳です。")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
ありがとうございます。ユーザー情報の処理が完了しました。

関数を直接実行するのではなく、run_functionをFunctionToolでラップしてツール化、run_functionの中で 入力からパースされた文字列を do_some_workに渡して処理する、って感じに見える。

引数と docstring の自動解析

上にも書いてあったけど、ツールの定義は自動的に解析される。いくつか注意点。

  • シグネチャ解析は inspect で行われる。
    • 型アノテーションで引数の型を把握、全体のスキーマを Pydantic モデルで動的に構築。
    • Python の基本型、Pydantic モデル、TypedDict など、ほとんどの型をサポート。
  • docstring の解析は griffe で行われる。
    • サポートする docstring 形式は google、sphinx、numpy。
    • docstring形式の検出は自動で行うが、ベストエフォートになるため、function_tool 呼び出し時に明示的に設定することができる。
    • use_docstring_info=Falseを設定すると docstring 解析を無効化できる。

スキーマ抽出のコードは agents.function_schema を参照。

ツールとしての エージェント

マルチエージェントでハンドオフするのではなく、エージェントをツールとして使う。.as_tool()でエージェントをツールに変換できる。

from agents import Agent, Runner
import asyncio
import json

spanish_agent = Agent(
    name="Spanish agent",
    instructions="ユーザのメッセージをスペイン語に翻訳する。",
)

english_agent = Agent(
    name="English agent",
    instructions="ユーザのメッセージを英語に翻訳する。",
)

orchestrator_agent = Agent(
    name="orchestrator_agent",
    instructions=(
        "あなたは翻訳エージェントです。与えられたツールを使って翻訳してください。"
        "複数の翻訳を求められた場合は、該当するツールを呼び出してください。"
    ),
    tools=[
        spanish_agent.as_tool(
            tool_name="translate_to_spanish",
            tool_description="ユーザのメッセージをスペイン語に翻訳する。",
        ),
        english_agent.as_tool(
            tool_name="translate_to_english",
            tool_description="ユーザのメッセージを英語に翻訳する。",
        ),
    ],
)

async def main():
    result = await Runner.run(orchestrator_agent, input="「こんにちは！ご機嫌いかがですか？」を英語に翻訳して。")
    print("最終回答:\n", result.final_output)
    print("\n会話履歴:\n", json.dumps(result.to_input_list(), indent=2, ensure_ascii=False))

if __name__ == "__main__":
    asyncio.run(main())
出力
最終回答:
 英語に翻訳すると、「Hello! How are you doing?」です。

会話履歴:
 [
  {
    "content": "「こんにちは！ご機嫌いかがですか？」を英語に翻訳して。",
    "role": "user"
  },
  {
    "arguments": "{\"input\":\"こんにちは！ご機嫌いかがですか？\"}",
    "call_id": "call_eXRPCdb7FHQCEx4goqR2MP5f",
    "name": "translate_to_english",
    "type": "function_call",
    "id": "fc_68a1acb4df0c819ba8bbdd8c86576d160a425b5fcc13894d",
    "status": "completed"
  },
  {
    "call_id": "call_eXRPCdb7FHQCEx4goqR2MP5f",
    "output": "Hello! How are you doing?",
    "type": "function_call_output"
  },
  {
    "id": "msg_68a1acb73fe0819bab97459ef9d8f1a40a425b5fcc13894d",
    "content": [
      {
        "annotations": [],
        "text": "英語に翻訳すると、「Hello! How are you doing?」です。",
        "type": "output_text",
        "logprobs": []
      }
    ],
    "role": "assistant",
    "status": "completed",
    "type": "message"
  }
]

ツールとして実行されているのがわかる。

ツール化したエージェントのカスタマイズ

エージェントに対して .as_tool はお手軽に使える代わりに、細かい設定などはサポートされていない場合もあり、例えばエージェントループの最大回数を指定する max_turns などは設定できない。こういった場合は、ツール内で直接 Runner.run() を呼び出すことができる。

from agents import Agent, Runner, function_tool
import asyncio
import json

@function_tool
async def run_spanish_agent(input: str) -> str:
    """スペイン語に翻訳するエージェントを実行するツール"""
    agent = Agent(
        name="Spanish agent",
        instructions="ユーザのメッセージをスペイン語に翻訳する。",
    )
    result = await Runner.run(
        agent,
        input,
        # カスタムな設定をここでおこなう
        max_turns=5,
    )
    return str(result.final_output)

orchestrator_agent = Agent(
    name="orchestrator_agent",
    instructions=(
        "あなたは翻訳エージェントです。与えられたツールを使って翻訳してください。"
        "複数の翻訳を求められた場合は、該当するツールを呼び出してください。"
    ),
    tools=[run_spanish_agent],
)

async def main():
    result = await Runner.run(orchestrator_agent, input="「こんにちは！ご機嫌いかがですか？」をスペイン語に翻訳して。")
    print("最終回答:\n", result.final_output)
    print("\n会話履歴:\n", json.dumps(result.to_input_list(), indent=2, ensure_ascii=False))

if __name__ == "__main__":
    asyncio.run(main())
出力
最終回答:
 「こんにちは！ご機嫌いかがですか？」のスペイン語訳は「¡Hola! ¿Cómo estás?」です。

会話履歴:
 [
  {
    "content": "「こんにちは！ご機嫌いかがですか？」をスペイン語に翻訳して。",
    "role": "user"
  },
  {
    "arguments": "{\"input\":\"こんにちは！ご機嫌いかがですか？\"}",
    "call_id": "call_uCnUsa07b3iox2bVdSmgkpMd",
    "name": "run_spanish_agent",
    "type": "function_call",
    "id": "fc_68a1bb0286e081988c802d61e4b3c9c70fa42716911182ef",
    "status": "completed"
  },
  {
    "call_id": "call_uCnUsa07b3iox2bVdSmgkpMd",
    "output": "¡Hola! ¿Cómo estás?",
    "type": "function_call_output"
  },
  {
    "id": "msg_68a1bb04d4c48198b826a9ed4fd3cc7d0fa42716911182ef",
    "content": [
      {
        "annotations": [],
        "text": "「こんにちは！ご機嫌いかがですか？」のスペイン語訳は「¡Hola! ¿Cómo estás?」です。",
        "type": "output_text",
        "logprobs": []
      }
    ],
    "role": "assistant",
    "status": "completed",
    "type": "message"
  }
]

出力のカスタム抽出

エージェントをツール化した場合、オーケストレータとなるエージェントに返す前に、ツール化したエージェントの出力を加工したり、というようなユースケースがある。例えば以下。

  • サブエージェントのチャット履歴から特定の情報（例: JSON ペイロード）を抽出。
  • エージェント の最終回答を変換・再整形する（例: Markdown をプレーンテキストや CSV に変換）。
  • 出力を検証し、エージェント の応答が欠落 or 不正な場合にフォールバック値を提供。

この場合は、as_tool に custom_output_extractor を渡してここで加工用の関数などを指定する。

ここはドキュメントの断片的なコードだけだとイマイチ理解できず、それを踏まえた完全なサンプルコードも書けなかったので、ドキュメントのまま。

async def extract_json_payload(run_result: RunResult) -> str:
    # エージェントの出力を逆順に走査して、ツール呼び出しからJSONっぽいメッセージを探す
    for item in reversed(run_result.new_items):
        if isinstance(item, ToolCallOutputItem) and item.output.strip().startswith("{"):
            return item.output.strip()
    # Fallback to an empty JSON object if nothing was found
    return "{}"


json_tool = data_agent.as_tool(
    tool_name="get_data_json",
    tool_description="データエージェントを実行して、JSONペイロードのみを返す。",
    custom_output_extractor=extract_json_payload,
)

ツールの条件付き有効化

as_toolのis_enabledを使うと、特定の条件を満たした場合のみ、ツール化エージェントを有効化する、といったことができる

・・・のだが、どうやらこれはつい最近マージされたばかりで、執筆時点のパッケージには反映されていない。

https://github.com/openai/openai-agents-python/issues/1097

サンプルだけ。

import asyncio
from agents import Agent, AgentBase, Runner, RunContextWrapper
from pydantic import BaseModel

class LanguageContext(BaseModel):
    language_preference: str = "japanese_english"

def english_enabled(ctx: RunContextWrapper[LanguageContext], agent: AgentBase) -> bool:
    """プリファレンスが "english_japanese" の場合、英語のツールを有効にする。"""
    return ctx.context.language_preference == "english_japanese"

# Create specialized agents
english_agent = Agent(
    name="english_agent",
    instructions="あなたは英語で応答する。ユーザの質問に対して、常に英語で応答してください。",
)

japanese_agent = Agent(
    name="japanese_agent",
    instructions="日本語で回答する。ユーザの質問に対して、常に日本語で応答してください。",
)

# Create orchestrator with conditional tools
orchestrator = Agent(
    name="orchestrator",
    instructions=(
        "あなたは多言語対応のアシスタントです。"
        "あなたは与えられたツールを使用して、ユーザーの質問に応答してください。"
        "すべてのツールを呼び出して、異なる言語で回答してください。"
        "あなたは自分で言語で回答してはいけません。常に提供されたツールを使用してください。"
    ),
    tools=[
        japanese_agent.as_tool(
            tool_name="respond_japanese",
            tool_description="ユーザの質問に日本語で回答する。",
            is_enabled=True,  # 常に有効
        ),
        english_agent.as_tool(
            tool_name="respond_english",
            tool_description="ユーザの質問に英語で回答する。",
            is_enabled=english_enabled,
        ),
    ],
)

async def main():
    context = RunContextWrapper(LanguageContext(language_preference="english_japanese"))
    result = await Runner.run(orchestrator, "こんにちは！", context=context.context)
    print(result.final_output)

asyncio.run(main())

is_enabledは以下を指定可能

  • 真偽値: True（常に有効）/ False（常に無効）
  • 呼び出し可能関数: (context, agent) を受け取って、真偽値を返す関数
  • 非同期関数: 複雑な条件ロジック向けの async 関数

無効化されたツール（is_enabled=False）はLLMから隠蔽されるため、以下のような用途に有用

  • ユーザー権限によって使える・使えないを振り分ける
  • 環境別のツール可用性（dev / prod で分ける）
  • 異なるツール構成の A/B テスト
  • 実行時の状態に基づいてツールを動的にフィルタリングする

関数ツールでのエラー処理

Python関数を@function_toolでツール化する際にfailure_error_functionで、ツール呼び出し失敗時に呼び出す関数を指定でき、LLMにエラーレスポンスを渡すことができる。

  • デフォルト（failure_error_functionに何も指定しない）では、エラーが発生したことをLLMに返すdefault_tool_error_functionが実行される
  • 独自のエラー関数を渡した場合はソレが実行され、その結果がLLMに返される。
  • 明示的にNoneを指定した場合、ツール呼び出しエラーは再スローされるので、自分で処理する必要がある。モデルが不正なJSONを生成した場合や ModelBehaviorError、コードがクラッシュした場合は UserError などがスローされる。
from agents import Agent, Runner, RunContextWrapper, function_tool
from typing import Any
import asyncio

def my_custom_error_function(context: RunContextWrapper[Any], error: Exception) -> str:
    """ユーザフレンドリーなエラーメッセージを返すカスタムなエラー関数"""
    print(f"ツール呼び出しで以下のエラーが発生しました: {error}")
    return "内部エラーが発生しました。しばらくしてから再度お試しください。"

@function_tool(failure_error_function=my_custom_error_function)
def get_user_profile(user_id: str) -> str:
    """ユーザのプロフィールを取得する"""
    # 以下のユーザID以外はエラーとする
    if user_id == "user_123":
        return "ユーザID: user_123 は、ユーザ名: 山田太郎、年齢: 25歳です。"
    else:
        raise ValueError(f"ユーザID: {user_id} のプロフィールを取得できませんでした。APIがエラーを返しました。")

agent = Agent(
    name="Assistant",
    instructions="日本語で回答する。",
    tools=[get_user_profile],  
)

async def main():
    result = await Runner.run(agent, "user_123のプロフィールを取得して。")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
ユーザID: user_123 のプロフィール情報は以下の通りです。

- **ユーザ名**: 山田太郎
- **年齢**: 25歳

入力を変えてみる

(snip)
    result = await Runner.run(agent, "user_456のプロフィールを取得して。")
(snip)
出力
ツール呼び出しで以下のエラーが発生しました: ユーザID: user_456 のプロフィールを取得できませんでした。APIがエラーを返しました。
プロフィール取得中にエラーが発生しました。しばらく待ってからもう一度試してみてください。申し訳ありません。

FunctionToolを手動で作成した場合は、on_invoke_tool内でエラー処理を行う必要がある。

kun432kun432

Model Context Protocol（MCP）

https://openai.github.io/openai-agents-python/ja/mcp/

MCPの説明は割愛。

MCP サーバ

MCPのトランスポートは現時点では3種類

  1. stdio: サーバはアプリケーションのサブプロセスとして実行される、「ローカル」で動作する形。
  2. HTTP over SSE: サーバはリモートで動作し、URLで接続する。
  3. Streamable HTTP: サーバは、MCP 仕様で定義された Streamable HTTP トランスポートを使用してリモートで動作する。

それぞれに合わせて、MCPServerStdio、MCPServerSse、MCPServerStreamableHttp の3クラスが用意されている。

なお、MCP仕様にもあるが、

2. ストリーム対応 HTTP

とあるので、今後2は徐々に減っていくのだろう。

とりあえず stdio と Streamable HTTPの実装例

まず、stdio。MCP公式のfilesystem サーバをサンプルとして使う。事前にこういうディレクトリ構成を用意しておく。

出力
samples_dir/
├── favorite_derby_winners.txt
├── favorite_takaraduka_winners.txt
└── favorite_arima_winners.txt

mkdir samples_dir
echo -e "トウカイテイオー\nナリタブライアン\nディープインパクト\n" > samples_dir/favorite_derby_winners.txt
echo -e "クロノジェネシス\nゴールドシップ\nサイレンススズカ\n" > samples_dir/favorite_takaraduka_winners.txt
echo -e "トウカイテイオー\nディープインパクト\nクロノジェネシス\n" > samples_dir/favorite_arima_winners.txt

コード

from agents import Agent, Runner
from agents.mcp import MCPServer, MCPServerStdio
import asyncio
import os

async def run(mcp_server: MCPServer):
    agent = Agent(
        name="Haiku agent",
        instructions="日本語で回答する。",
        mcp_servers=[mcp_server]
    )
    result = await Runner.run(agent, input="アクセスできるディレクトリ内のファイルをリストアップして。")
    print(result.final_output)
    print("-" * 20)
   
    result = await Runner.run(agent, input="私のお気に入りのダービー馬は？さっきのファイルのどれかに書いてあるよ。")
    print(result.final_output)
    print("-" * 20)

async def main():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    samples_dir = os.path.join(current_dir, "samples_dir")

    async with MCPServerStdio(
        name="Filesystem Server, via npx",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir],
        },
    ) as server:
        await run(server)

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

実行結果

出力
アクセス可能なディレクトリ内のファイルは以下の通りです：

- favorite_arima_winners.txt
- favorite_derby_winners.txt
- favorite_takaraduka_winners.txt
--------------------
あなたのお気に入りのダービー馬は「トウカイテイオー」「ナリタブライアン」「ディープインパクト」です。
--------------------

次に Streamable HTTP。Ramen APIのMCPを使用させていただく。

from agents import Agent, Runner
from agents.mcp import MCPServer, MCPServerStreamableHttp
import asyncio
import os

async def run(mcp_server: MCPServer):
    agent = Agent(
        name="Haiku agent",
        instructions="日本語で回答する。",
        mcp_servers=[mcp_server]
    )
    result = await Runner.run(agent, input="横浜のラーメン屋を知りたい。")
    print(result.final_output)
   
async def main():
    async with MCPServerStreamableHttp(
        # ref: https://github.com/yusukebe/ramen-api
        name="Ramen API MCP",
        params={
            "url": "https://ramen-api.dev/mcp"
        },
    ) as server:
        await run(server)

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

実行結果。

出力
横浜のラーメン屋をいくつか紹介します。

1. **吉村家**
   ![吉村家](https://ramen-api.dev/images/yoshimuraya/yoshimuraya-001.jpg)

2. **杉田家**
   ![杉田家](https://ramen-api.dev/images/sugitaya/sugitaya-001.jpg)

3. **たかさご家**
   ![たかさご家](https://ramen-api.dev/images/takasagoya/takasagoya-001.jpg)

4. **上々家**
   ![上々家](https://ramen-api.dev/images/jyoujyouya/jyoujyouya-001.jpg)

5. **とらきち家**
   ![とらきち家](https://ramen-api.dev/images/torakichiya/torakichiya-001.jpg)
   ![とらきち家](https://ramen-api.dev/images/torakichiya/torakichiya-002.jpg)

どのラーメン屋に興味がありますか？詳細を知りたい場合は教えてください。
kun432kun432

MCPサーバの使用

上の例でもやっているが、エージェントの mcp_servers で、エージェントにMCPサーバを追加して使用することができる。エージェントは実行されるたびに、

  • list_tools()でツールを認識
  • LLMにツールを渡す
  • LLMがツール使用を判断したら、call_tool() でツールを呼び出す

という流れになる。

自分は、MCPのシーケンスをちゃんと理解していないのだが、Strands AgentsでMCPをやったときと同じように、with の中で処理をするような形になってるのはMCPが初期化プロセスのようなものを必要とするからだろうと思っている。ツールのようなシンプルな指定の仕方とはやや異なるね。

複数のMCPサーバを指定する場合はどうするか？上の２つのMCPサーバを組み合わせてみる。

from agents import Agent, Runner
from agents.mcp import MCPServer, MCPServerStreamableHttp, MCPServerStdio
import asyncio
import os
   
async def main():
    ramen_mcp = MCPServerStreamableHttp(
        name="Ramen API MCP",
        params={
            "url": "https://ramen-api.dev/mcp"
        },
    )

    current_dir = os.path.dirname(os.path.abspath(__file__))
    samples_dir = os.path.join(current_dir, "samples_dir")

    filesystem_mcp = MCPServerStdio(
        name="Filesystem Server, via npx",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir],
        },
    )

    async with ramen_mcp, filesystem_mcp:
        agent = Agent(
            name="Haiku agent",
            instructions="日本語で回答する。",
            mcp_servers=[ramen_mcp, filesystem_mcp]
        )
        result = await Runner.run(agent, input="横浜のラーメン屋を調べて、その結果をファイルにMarkdownに出力して")
        print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())
出力
横浜のラーメン屋情報をMarkdown形式でファイルに保存しました。以下の内容です：

### ファイルパス
`/Users/kun432/work/agents-sdk-work/samples_dir/横浜ラーメン/README.md`

### 内容
- **吉村家**
  - ![吉村家](https://ramen-api.dev/images/yoshimuraya/yoshimuraya-001.jpg)

- **杉田家**
  - ![杉田家](https://ramen-api.dev/images/sugitaya/sugitaya-001.jpg)

- **たかさご家**
  - ![たかさご家](https://ramen-api.dev/images/takasagoya/takasagoya-001.jpg)

- **上々家**
  - ![上々家](https://ramen-api.dev/images/jyoujyouya/jyoujyouya-001.jpg)

- **とらきち家**
  - ![とらきち家1](https://ramen-api.dev/images/torakichiya/torakichiya-001.jpg)
  - ![とらきち家2](https://ramen-api.dev/images/torakichiya/torakichiya-002.jpg)

他にご質問があれば教えてください。
出力
samples_dir/
(snip)
└── 横浜ラーメン
    └── README.md

参考

https://zenn.dev/msmtec/articles/openai-agents-sdk-multiagent

kun432kun432

ツールのフィルタリング

MCPサーバが提供するツールの許可・ブロックが設定できる。静的・動的の2種類。

静的ツールフィルタリング

シンプルに許可・ブロックするツールの指定を tool_filter=create_static_tool_filter() で行う。

  • 許可: allowed_tool_namesで、許可するツールをリスト指定。
  • ブロック: blocked_tool_namesで、ブロックするツールをリスト指定。
  • allowed_tool_names と blocked_tool_names が両方指定されている場合は以下となる。
    1. allowed_tool_namesを適用。指定されているツールだけが残る。
    2. blocked_tool_names を適用。指定されたツールを1から除外。
    3. 1&2の結果、残ったツールだけが使える

こんな感じで確認できた。

from agents import Agent, RunContextWrapper
from agents.mcp import MCPServerStdio, create_static_tool_filter
import asyncio
import os


async def main():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    samples_dir = os.path.join(current_dir, "samples_dir")

    print("-" * 10, "利用可能なツール一覧（デフォルト）", "-" * 10)
    async with MCPServerStdio(
        name="Filesystem Server, via npx",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir],
        },
    ) as mcp:
        tools = await mcp.list_tools()
        print("Number of tools:", len(tools))
        for tool in tools:
            print("-", tool.name)

    print("-" * 10, "利用可能なツール一覧（許可）", "-" * 10)

    async with MCPServerStdio(
        name="Filesystem Server, via npx",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir],
        },
        tool_filter=create_static_tool_filter(
            allowed_tool_names=["read_file", "list_directory"],
        )
    ) as mcp:
        # フィルタする場合はエージェントとコンテキストが必要らしい
        agent = Agent(name="agent", mcp_servers=[mcp])
        tools = await agent.get_mcp_tools(run_context=RunContextWrapper(None))
        print("Number of tools:", len(tools))
        for tool in tools:
            print("-", tool.name)

    print("-" * 10, "利用可能なツール一覧（ブロック）", "-" * 10)

    async with MCPServerStdio(
        name="Filesystem Server, via npx",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir],
        },
        tool_filter=create_static_tool_filter(
            blocked_tool_names=["read_file", "list_directory"],
        )
    ) as mcp:
        # フィルタする場合はエージェントとコンテキストが必要らしい
        agent = Agent(name="agent", mcp_servers=[mcp])
        tools = await agent.get_mcp_tools(run_context=RunContextWrapper(None))
        print("Number of tools:", len(tools))
        for tool in tools:
            print("-", tool.name)

if __name__ == "__main__":
    asyncio.run(main())
出力
---------- 利用可能なツール一覧（デフォルト） ----------
(snip)
Number of tools: 14
- read_file
- read_text_file
- read_media_file
- read_multiple_files
- write_file
- edit_file
- create_directory
- list_directory
- list_directory_with_sizes
- directory_tree
- move_file
- search_files
- get_file_info
- list_allowed_directories
---------- 利用可能なツール一覧（許可） ----------
(snip)
Number of tools: 2
- read_file
- list_directory
---------- 利用可能なツール一覧（ブロック） ----------
(snip)
Number of tools: 12
- read_text_file
- read_media_file
- read_multiple_files
- write_file
- edit_file
- create_directory
- list_directory_with_sizes
- directory_tree
- move_file
- search_files
- get_file_info
- list_allowed_directories

動的ツールフィルタリング

フィルタリングのロジックを自分で書くこともできる。こちらが動的なフィルタリングのサンプル。

from agents import Agent, RunContextWrapper
from agents.mcp import MCPServerStdio
import asyncio
import os

from agents.mcp import ToolFilterContext

# シンプルな同期フィルタの場合
def custom_filter(context: ToolFilterContext, tool) -> bool:
    """カスタムツールフィルタの例"""
    # ツール名のパターンに基づいたフィルタロジック
    return tool.name.startswith("list_")

# コンテキストに基づいたフィルタの場合
def context_aware_filter(context: ToolFilterContext, tool) -> bool:
    """コンテキスト情報にもとづいてツールをフィルタする。"""
    # エージェント情報にアクセスする
    agent_name = context.agent.name

    # サーバー情報にアクセスする
    server_name = context.server_name

    # カスタムなフィルタロジックをここに実装する
    return some_filtering_logic(agent_name, server_name, tool)

# 非同期フィルタの場合
async def async_filter(context: ToolFilterContext, tool) -> bool:
    """非同期フィルタの例"""
    # 非同期操作が必要な場合はここに実装する
    result = await some_async_check(context, tool)
    return result

async def main():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    samples_dir = os.path.join(current_dir, "samples_dir")

    async with MCPServerStdio(
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir],
        },
        # ここではシンプルなツール名パターンでのフィルタを使用
        tool_filter=custom_filter  # context_aware_filter や async_filter もそれぞれ実装して指定できる
    ) as mcp:
        # フィルタする場合はエージェントとコンテキストが必要らしい
        agent = Agent(name="agent", mcp_servers=[mcp])
        tools = await agent.get_mcp_tools(run_context=RunContextWrapper(None))
        print("Number of tools:", len(tools))
        for tool in tools:
            print("-", tool.name)

if __name__ == "__main__":
    asyncio.run(main())
出力
Number of tools: 3
- list_directory
- list_directory_with_sizes
- list_allowed_directories
kun432kun432

プロンプト

MCPサーバのプロンプト を使って、インストラクションをカスタマイズしたりできる。

プロンプトの使用

プロンプトをサポートする MCPサーバー のプロンプトは、以下の2つで取得できる。

  • list_prompts(): MCPサーバで利用可能なすべてのプロンプトを一覧表示
  • get_prompt(name, arguments): 任意のパラメータをつけて、特定のプロンプトを取得

プロンプトに対応したMCPサーバを探してみると以下があったので、これを使う。

https://github.com/reading-plus-ai/mcp-server-deep-research/

ここで紹介されている

https://dev.classmethod.jp/articles/trial-prompts-of-mcp-by-mcp-server-deep-research/

list_prompts() でプロンプトの情報を取得してみる。

from agents import Agent, RunContextWrapper
from agents.mcp import MCPServerStdio, create_static_tool_filter
import asyncio
import os


async def main():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    samples_dir = os.path.join(current_dir, "samples_dir")

    async with MCPServerStdio(
        name="AWS Documentation MCP Server",
        params={
            "command": "uvx",
            "args": ["deep-research-mcp-server"]
        }
    ) as mcp:
        # プロンプト情報を取得
        prompts = await mcp.list_prompts()
        for prompt in prompts.prompts:
            print(f"[{prompt.name}]:")
            print(f"Description: {prompt.description}")
            # 引数を取得
            print("Args:")
            for arg in prompt.arguments:
                print(f"- {arg.name}: {arg.description}")

if __name__ == "__main__":
    asyncio.run(main())
出力
[deep-research]:
Description: A prompt to conduct deep research on a question
Args:
- research_question: The research question to investigate

この結果を使って、次に get_prompt(name, arguments) を使ってプロンプトを取得してみる。

from agents import Agent, RunContextWrapper
from agents.mcp import MCPServerStdio, create_static_tool_filter
import asyncio
import os


async def main():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    samples_dir = os.path.join(current_dir, "samples_dir")

    async with MCPServerStdio(
        name="Deep Research MCP Server",
        params={
            "command": "uvx",
            "args": ["deep-research-mcp-server"]
        }
    ) as mcp:
        # プロンプトを取得
        prompt = await mcp.get_prompt(
            "deep-research",
            {"research_question": "競馬の重要な魅力"}
        )
        print(prompt.messages[0].content.text)

if __name__ == "__main__":
    asyncio.run(main())
出力
You are a professional researcher tasked with conducting thorough research on a topic and producing a structured, comprehensive report. Your goal is to provide a detailed analysis that addresses the research question systematically.

The research question is:

<research_question>
競馬の重要な魅力
</research_question>

Follow these steps carefully:

1. <question_elaboration>
   Elaborate on the research question. Define key terms, clarify the scope, and identify the core issues that need to be addressed. Consider different angles and perspectives that are relevant to the question.
</question_elaboration>

2. <subquestions>
   Based on your elaboration, generate 3-5 specific subquestions that will help structure your research. Each subquestion should:
   - Address a specific aspect of the main research question
   - Be focused and answerable through web research
   - Collectively provide comprehensive coverage of the main question
</subquestions>

3. For each subquestion:
   a. <web_search_results>
      Search for relevant information using web search. For each subquestion, perform searches with carefully formulated queries.
      Extract meaningful content from the search results, focusing on:
      - Authoritative sources
      - Recent information when relevant
      - Diverse perspectives
      - Factual data and evidence
      
      Be sure to properly cite all sources and avoid extensive quotations. Limit quotes to less than 25 words each and use no more than one quote per source.
   </web_search_results>

   b. Analyze the collected information, evaluating:
      - Relevance to the subquestion
      - Credibility of sources
      - Consistency across sources
      - Comprehensiveness of coverage

4. Create a beautifully formatted research report as an artifact. Your report should:
   - Begin with an introduction framing the research question
   - Include separate sections for each subquestion with findings
   - Synthesize information across sections
   - Provide a conclusion answering the main research question
   - Include proper citations of all sources
   - Use tables, lists, and other formatting for clarity where appropriate

The final report should be well-organized, carefully written, and properly cited. It should present a balanced view of the topic, acknowledge limitations and areas of uncertainty, and make clear, evidence-based conclusions.

Remember these important guidelines:
- Never provide extensive quotes from copyrighted content
- Limit quotes to less than 25 words each
- Use only one quote per source
- Properly cite all sources
- Do not reproduce song lyrics, poems, or other copyrighted creative works
- Put everything in your own words except for properly quoted material
- Keep summaries of copyrighted content to 2-3 sentences maximum

Please begin your research process, documenting each step carefully.
kun432kun432

キャッシュ

デフォルトでは、エージェントが実行されるたびに、MCPサーバで list_tools() が呼び出されるため、レイテンシーに影響する可能性がある。これを回避するためにキャッシュが用意されている。

  • MCPサーバ定義時に cache_tools_list=True を指定するとツール一覧がキャッシュされる。
    • ツール一覧が途中で変更されないことが重要。
  • invalidate_tools_cache() を呼び出すと、キャッシュは無効化される。
このスクラップは1ヶ月前にクローズされました