OpenAI Agents SDKで複数のMCPサーバーと連携したマルチエージェントを作る

きっかけ

7月末のWeights & Biasesのミートアップ　OpenAI Agents SDKハンズオンに参加してAgents SDKについて学びました。
便利そうなのでさっそく使って、機械設計用の計算をするマルチエージェントを作ってみました。

今回やりたいこと

• OpenAI Agents SDKを使う
• マルチエージェントを構築
• Model Context Protocol (MCP)を利用

ツールは何でもよいので、以前Function Calling用に作成した関数セットなど手元にある計算ツールを再利用しました。また、社内外のMCPの活用を見据えて、既存のコードをMCP化して使いました。

これができると、社内外のAPIなどを自律的に操作してタスクを実行してくれるエージェントを作れるようになります。あとはこの例と同様に、使いたいツールや子エージェントを追加・入替えて、望んだ機能に拡張すればよいです。

MCPサーバーの用意

まず、エージェントに渡すツールの準備をします。

皆様ご存じの通り、MCPはデータやツールとAIモデルとを接続するプロトコルです。
これから手元の関数たちをMCPでLLMに接続できるようにします。

MCPサーバーの構成

今回は以下の4つのMCPサーバーを用意します。アルミフレームを使った構造物の設計支援がテーマです。

• たわみを計算する
• 耐荷重を計算する
• 質量を計算する
• データベースから必要な値を取得する

これらのMCPサーバーで、計算機能と商品の物理量（密度、ヤング率、断面2次モーメントなど）を提供します。
最近のLLMは賢いとはいえ、商品情報を覚えていないし、たまに計算ミスもするので、これらのツールで補います。

MCPサーバーの実装

以下は耐荷重を計算する関数をMCPサーバーにする例です。
各関数の前にmcp.tool()を付けて、FastMCPを定義しrunするところが、今回の追加箇所です。

関数のdocstringには、関数と引数、返り値の説明を書いています。この情報がMCPで渡されて、LLMは関数の使い方を理解します。

mcp_server_load.py

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    "defl_load_MCP",
    description="Model Context Protocolによる耐荷重計算機",
)

@mcp.tool()
def load_calulator_no1(L: float, 
                       I: float, 
                       E: float, 
                       defl_rate: float = 0.001,
                       safety_factor: float = 5) -> float:
    """片持ち梁の先端にかけられる最大の荷重を計算します。

    Args:
        L: (mm) 梁の長さ
        I: (mm^4) 断面2次モーメント
        E: (N/mm^2) ヤング率
        defl_rate: 許容できるたわみ率
        safety_factor: 荷重の安全率 (デフォルト 5).
    Returns:
        最大許容荷重 (N)
    """
    capable_factor = defl_rate / safety_factor
    return capable_factor * 3 * E * I / L**2


def main():
    """Run the MCP server"""
    mcp.run()

if __name__ == "__main__":
    main()

no1という関数名から想像できるかもしれませんが、実際には複数の関数を用意しています。複数の場合も同様に、関数定義の前にデコレータをつけるだけです。冗長なので省略しました。

計算式はミスミが公開しているものを使いました。

OpenAI Agents SDKでマルチエージェントを構築

いよいよエージェントを作ります。

常識となりつつありますが、最近のAIエージェントは、LLMが指示に答えるために必要な手順を自律的に計画・実行・評価して動作するシステムを指します。内部では、必要なツールやデータをLLMが決めた手順に沿って使います。

マルチエージェントの構成

シンプルに、トリアージエージェントを親として、3種の専門エージェントを子に持つ構成にしました。

各計算エージェントは、対応する複数のMCPサーバーを使用して物理量の取得や計算を行います。

準備

APIキーを.envファイルに書いて保存し、

.env

OPENAI_API_KEY=...

ターミナルで読込みます。

set -a
source .env
set +a

MCPをAgentsに渡してマルチエージェントを構築

まずMCPサーバーを設定します。
以下に、耐荷重計算のMCPサーバーmcp_server_loadの設定を紹介します。
複数のMCPサーバーを立てるので、共通して使うコードは関数にまとめています。

multi_agent.py

from agents.mcp import MCPServer, MCPServerStdio
from agents.mcp import create_static_tool_filter
import os, sys

def get_mcp_server(mcp_server_path: str, 
                   mcp_server_name: str, 
                   allowed_tool_names: list[str] = None
                   ) -> MCPServer:
    """ MCP サーバーを取得 """
    
    # Python executable pathを取得
    python_executable = sys.executable
    current_dir = os.path.dirname(os.path.abspath(__file__))
    parent_dir = os.path.dirname(current_dir)

    # 必要ならstatic tool filterを作成
    if type(allowed_tool_names) == list:
        tool_filter = create_static_tool_filter(allowed_tool_names)
    else:
        tool_filter = None

    # MCP serverを作成
    server = MCPServerStdio(
        name=mcp_server_name,
        params={
            "command": python_executable,
            "args": [mcp_server_path],
            "env": {
                "PYTHONPATH": parent_dir
            }
        },
        tool_filter=tool_filter
    )

    return server


### 耐荷重計算の MCP serverを作成
load_mcp_server_path = "PATH to MCP Server..."
mcp_name = "mcp_server_load"
mcp_server_load = get_mcp_server(load_mcp_server_path, mcp_name)

同様にして、たわみ計算mcp_server_deflと質量計算mcp_server_massのMCPも作成します。

次にデータ取得MCPサーバーを用意します。

ここではMCPサーバーの立て方を少し工夫しました。
データ取得機能は一個のMCPサーバーファイルmcp_server_data.pyにまとめていました。しかし、耐荷重・たわみ計算と、質量計算とでは使用する関数が異なり、そのままでは各専門エージェントに余計な関数も渡すことになります。
無駄に間違う可能性をなくすため、以下のように提供する関数をフィルタして、別々のMCPサーバーとして起動するようにしました。

multi_agent.py

### MCP serverのファイルは共通
data_mcp_server_path = "PATH to MCP Server..."
data_mcp_name = "mcp_server_data"

### 耐荷重・たわみ計算用のデータ取得 MCP server
allowed_tool_names = [
    "select_Young_modulus",
    "select_moment_of_inertia",
    "select_length",
    "length_to_mm",
    "load_to_N",
    "load_from_N"
]
mcp_server_defl_data = get_mcp_server(data_mcp_server_path, data_mcp_name, allowed_tool_names)

### 質量計算用のデータ取得 MCP server
allowed_tool_names = [
    "select_density",
    "select_length",
    "length_to_mm",
]
mcp_server_mass_data = get_mcp_server(data_mcp_server_path, data_mcp_name, allowed_tool_names)

次に、各エージェントに与える指示文をそれぞれ定義します。
以下は耐荷重計算エージェント用の指示文です。与える役割や、計算の手順などを明記しています。

multi_agent.py

instructions_defl = """あなたはアルミフレームを使った構造の荷重計算のエキスパートです。
以下のツールを使ってください:
    1. アルミフレーム製品の最大許容荷重を計算
    2. 与えられた製品型番のヤング率、断面2次モーメント、長さを取得

アルミフレームの耐荷重を計算するとき、以下の手順に従ってください:
    1. **ヤング率を取得**: 与えられた型番のヤング率を取得する。
    2. **断面2次モーメントを取得**: 与えられた型番の断面2次モーメントを取得する。
    3. **長さを取得**: 必要なら、与えられた型番の長さを取得する。
    4. **耐荷重を計算**: 取得したヤング率、断面2次モーメント、長さをもとに荷重を計算する。
    5. **耐荷重を回答**: 計算した耐荷重を結果として回答する。

通常、たわみ率には0.001が使われます。アルミフレームの耐荷重では安全率5が使われることがあります。
計算について明確な説明と安全性への注意を常に述べてください。"""

先ほど用意したMCPサーバーを連携させて、エージェントを構築します。
以下は、エージェントの構築から実行まで行う関数です。
複数のMCPサーバーと連携する方法はこちらの記事を参考にしました。

multi_agent.py

# Weights&Biases Weaveでトレーシング
import weave
from weave.integrations.openai_agents.openai_agents import WeaveTracingProcessor
set_trace_processors([WeaveTracingProcessor()])
WEAVE_PROJECT = "weave project name"
weave.init(project_name=WEAVE_PROJECT)

from agents import Agent, Runner

# 使用するモデルを指定
llm = "gpt-4.1"

@weave.op()
async def run(prompt: str):

    async with mcp_server_defl, mcp_server_load, mcp_server_mass, mcp_server_defl_data, mcp_server_mass_data:
        
        # 子エージェントを作成
        load_agent = Agent(
            name="耐荷重計算アシスタント",
            instructions=instructions_load,
            mcp_servers=[mcp_server_load, mcp_server_defl_data],
            model=llm
        )

        defl_agent = Agent(
            name="たわみ計算アシスタント",
            instructions=instructions_defl,
            mcp_servers=[mcp_server_defl, mcp_server_defl_data],
            model=llm
        )

        mass_agent = Agent(
            name="質量計算アシスタント",
            instructions=instruction_mass,
            mcp_servers=[mcp_server_mass, mcp_server_mass_data],
            model=llm
        )

        # 親エージェントを作成
        triage_agent = Agent(
            name="トリアージエージェント",
            instructions=(
    "1. ユーザーへ挨拶\n"
    "2. タスクを決定: たわみ、耐荷重、または質量\n"
    "3. たわみの場合 → defl_agentへハンドオフ\n"
    "4. 耐荷重の場合 → load_agentへハンドオフ\n"
    "5. 質量の場合 → mass_agentへハンドオフ\n"
    "5. ハンドオフを確認"
            ),
            handoffs=[defl_agent, load_agent, mass_agent],
            model=llm,
        )

        # マルチエージェントを実行
        response = await Runner.run(triage_agent, prompt)

        return response.final_output

専門エージェントにはツールとして使ってもらうMCPサーバーを渡しています。

マルチエージェントにするため、親エージェントtriage_agentには、3個の専門エージェントをhandoffsで渡しています。Agents SDKではハンドオフを使うと、あるエージェントがタスクを別のエージェントに任せることができます。

各エージェントでGPT-4.1を指定しました。
エージェントごとにモデルを変えることもできます（例えば、親エージェントや複雑な処理の専門エージェントはo3-miniで、残りはGPT4.1にする）。
また、他社のモデルやローカルLLMを指定することもできるようです。

最後に、マルチエージェントにプロンプトを与えて実行します。

multi_agent.py

import asyncio
from dotenv import load_dotenv
load_dotenv()

if __name__ == "__main__":

    message = """アルミフレーム'HFS6-3030-1000'の許容荷重を計算して。
                単純支持はりで集中荷重の条件で。"""
    print(f"Running: {message}")
    output = asyncio.run(run(prompt=message))
    print(output)

以上で、Pythonスクリプトができました。

【参考】MCPにしない場合

toolsに渡す方法

エージェントに渡すツールがPython関数なら、わざわざMCPにせず、そのままツールとして渡すことができます。実装としてはこちらの方が簡単です。

関数に@function_toolをつけて、Agentsのtoolsに渡すだけです。

multi_agent.py

@function_tool
def load_calulator_no1(L: float,
                       I: float,
                       E: float,
                       defl_rate: float = capable_defl_rate,
                       safety_factor: float = 5) -> float:
    """片持ち梁の先端にかけられる最大の荷重を計算します。
    Args: ...省略
    """
    capable_factor = defl_rate / safety_factor
    return capable_factor * 3 * E * I / L**2

load_agent = Agent(
            name="Load Calculation Assistant",
            instructions=instructions_load,
            tools=[load_calulator_no1, ...その他],
            model=llm
        )

今回は、公開されているMCPとの連携など今後の拡張を考えて、自前の関数をMCPにして使う方法を試しました。

実行

実行してみます。

uv run multi_agent.py

そうすると、MCPでツールを適宜呼び出して、必要な処理をしてくれます。
ここでは、アルミフレーム'HFS6-3030-1000'の許容荷重を計算して。単純支持はりで集中荷重の条件で。という指示を与えています。以下のツールを呼び出して使う想定です。

• ヤング率の取得
• 断面2次モーメントの取得
• 長さの取得
• 耐荷重の計算

最終的に以下の答えが返ってきました。

アルミフレーム『HFS6-3030-1000』の単純支持はりでの集中荷重条件における許容荷重を計算します。

計算手順と条件は以下の通りです：

1. 材料物性値（ヤング率）：約 69,972 N/mm²
2. 断面二次モーメント（I）：28,300 mm⁴
3. スパン長さ（L）：1,000 mm
4. 許容たわみ量：スパンの1/1000（1mm）
5. 安全率：5

この条件で、中央1点集中荷重の場合の許容荷重は 約 19 N です。

---
■安全上の注意
この値は「安全率5」を適用しており、たわみが1 mm以内となるよう計算されています。
短時間での静的荷重基準です。
取り付け方法や 使用実態によってはさらに余裕を持つことをおすすめします。

他の条件（長さや支持方法等）でも計算可能ですのでご相談ください。

必要なパラメータを取得し、正しく「約19N」と答えられました。
これからは、アルミフレームで棚などを設計している同僚から荷重計算を頼まれても、エージェントが僕の代わりに答えてくれそうです。

また、エージェントに渡したinstructionに従って、計算の説明と安全上の注意も生成しています。最後の「安全上の注意」の真偽には議論の余地がある気がしますが、責任を負わないよう安全に振った回答としては良さそうです。

最初に実行するときは、各MCPサーバーを立てるところでタイムアウトになることがありました（ローカルPCが貧弱なためかも）。
そのような場合でも、もう一度コマンドをたたくと実行されました。

おわりに

ミートアップの教材や公式のexamplesを参考にして、思ったものがさくっと実装できました。自由度はありつつミニマルで使いやすく感じました。

Agents SDKには、他にもガードレールや他社モデルを使う機能もあるので、いろいろ試そうと思います。

マルチエージェントの構成はたくさんのデザインパターンがあるので、用途に応じていろいろと試していきたいです。

本記事では、以前LangChainでFunction Callingや商品選定エージェントを作った実験を、新しいAgents SDKを使って発展させました。
今後も、Agents SDKを使ってさらにいろいろなエージェントを作っていこうと思います。

https://zenn.dev/msmtec/articles/function-calling

https://zenn.dev/msmtec/articles/tool-calling-agent

Version情報

• OpenAI Agents SDK 0.2.3
• Python 3.13.5