[AI Agent Jump Start 基礎編#4] マルチエージェントの理解と実装

4-1. マルチエージェント設計とMCP開発（翻訳サービス連携）

この記事では以下の2点を解説します。

• マルチエージェント設計のベストプラクティス
• MCP（Model Context Protocol）を使った翻訳サービス連携

---

4-1. マルチエージェント設計

① エージェント間の役割分担

単一エージェントに全機能を持たせると以下の問題が発生します。

• プロンプトが肥大化しやすい
• 複雑なツール利用時に管理が困難

👉 そのため、役割ごとにエージェントを分離するのがベストプラクティスです。

---

設計例

• ルーターAgent
  入力を解析し、「どのエージェント（またはツール）に渡すか」を決定する。

• プランナーAgent
  複雑な質問をタスクに分解し、順序を決めて実行する。

• ワーカーAgent
  各タスクを実行（計算、検索、MCP API呼び出しなど）。

---

ルーターAgentの実装例

from agents import Agent

router_agent = Agent(
    name="Router Agent",
    instructions="""
    あなたは入力を解析し、次のアクションを決めます。
    - 「検索系」は Search Agent
    - 「MCP操作系」は MCP Agent
    - 「単純な質問」は Answer Agent
    返答は「ハンドオフ」で対応。
    """,
)

プランナーAgentの実装例

planner_agent = Agent(
    name="Planner Agent",
    instructions="""
    質問をサブタスクに分割し、順序を提示してください。
    可能なら、どのツールやエージェントを使うかも指定してください。
    """
)

MCP呼び出しを含めたフロー設計例

ユーザー入力
   ↓
ルーターAgent → プランナーAgent
   ↓
（サブタスクごとにハンドオフ）
   ↓
Search Agent or MCP Agent or Answer Agent

4-2. MCP開発（翻訳サービス連携）

MCPとは？

MCP（Model Context Protocol） は、AIモデル（LLM）と外部サービスやAPIを安全かつ一貫性のある方法でつなぐための共通ルールです。

---

なぜ必要？

LLMは柔軟な生成能力を持つ一方で、以下のようなリスクを引き起こす可能性があります。

• 予期せぬエンドポイントへのアクセス
  社内の秘密URLや存在しない外部APIを勝手に推測し、アクセスしてしまう危険性。
  → 情報漏洩やセキュリティ事故につながる。

• 危険なリクエストの生成
  「翻訳して」と入力しただけで、不要なパラメータや破壊的操作を付与する可能性。
  → APIの誤作動やデータ破壊のリスク。

• 入力パラメータの暴走
  巨大テキストや不正値を送りつけ、DoS的な挙動を引き起こすリスク。
  → DoS的な障害を引き起こすリスク。

これらは「エージェント側の制御」だけでは完全に防ぎきれず、
プロトコルレベルでのガードレール が必要になります。

---

MCPを利用するメリット

MCPを導入することで、セキュリティ以外にも多くの利点があります。

• スキーマ定義
  translate(text: str, target_lang: str) のように定義 → 余計なSQLやパラメータを生成できない。

• 入力・出力バリデーション
  型・必須チェックで不正入力を排除。

• 実行ガードレール
  定義済みツール以外は呼び出せない。未知のURLを叩けない。

• 一貫性と再利用性
  MCP対応のサービスは、他のエージェント基盤でもそのまま利用可能。
  → ベンダーロックインを避け、横展開が容易。

• 開発効率の向上
  プロンプトで「どう呼び出すか」を細かく書かなくても、MCPのスキーマを参照すればよい。
  → 開発者の負担が減り、実装がシンプルになる。

---

MCPを使う vs 使わない場合の比較

項目	MCPあり	MCPなし
安全性	スキーマで制御	LLMが自由にリクエスト生成して暴走
拡張性	新しいAPIを追加しやすい	APIごとにカスタム実装が必要
ガードレール	定義済みツールのみ呼び出し可能	未知のAPIや危険操作も実行しうる

必要環境とライブラリ

• DeepL APIキーの準備（DeepL公式 で登録）
• Free / Proプラン選択
• 認証キー取得（Auth Key）
• Freeプランエンドポイント: https://api-free.deepl.com
• Proプランエンドポイント: https://api.deepl.com

---

注意点

• キーは公開リポジトリに絶対コミットしない
• 利用量（Freeは月あたり文字数上限あり）を把握する
• 本番は環境変数／Secret管理で注入

---

requirements.txt

fastapi==0.111.0
uvicorn==0.30.1
requests==2.32.3
mcp[cli]==1.12.4   # MCP公式SDK
python-dotenv==1.0.1
openai-agents==0.0.7

MCP開発（翻訳サービス連携）

コード詳細はこちら
https://github.ibm.com/AIAgentJumpStart/-AI-Agent-Jump-Start-4-

インストール方法

pip install -r requirements.txt

.env ファイルにDeepLのAPIキーを設定:

# DeepL
DEEPL_AUTH_KEY=YOUR_DEEPL_AUTH_KEY
# Free/Proで切替（未設定なら Free を既定にします）
DEEPL_API_BASE=https://api-free.deepl.com

# Azure OpenAI
AZURE_OPENAI_API_KEY=""
AZURE_OPENAI_API_VERSION=""
AZURE_OPENAI_ENDPOINT=""
AZURE_OPENAI_DEPLOYMENT="gpt-4o"

フォルダ構成（最小）

mcp-translate/
├─ server/
│  └─ mcp_translate_server.py      # FastAPI + FastMCP（/mcp に公開）
├─ agent_tools/
│  └─ translate_via_mcp.py         # Agents SDK向けのMCPクライアントツール
├─ mcp_agent.py                    # Translate Agent（tools=[translate_via_mcp]）
├─ run_mcp_agent.py                # 実行エントリ（Runner.run_sync）
├─ requirements.txt
├─ .env.example                    # サンプル環境変数

① MCPサーバー実装（翻訳APIラッパー）

server/mcp_translate_server.py

import os
import html
import requests
from typing import Optional, Dict, Any
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

# .env ロード
load_dotenv()

mcp = FastMCP(name="Translator MCP", instructions="DeepLでテキストを翻訳します。")

# 自前でツールレジストリを持つ
tools_registry = {}


# DeepL呼び出し
def _call_deepl(
    text: str, target_lang: str, source_lang: Optional[str]
) -> Dict[str, Any]:
    auth_key = os.getenv("DEEPL_AUTH_KEY")
    if not auth_key:
        raise RuntimeError("DEEPL_AUTH_KEY が未設定です（.env を確認）")

    api_base = os.getenv("DEEPL_API_BASE", "https://api-free.deepl.com")
    url = f"{api_base}/v2/translate"
    headers = {"Authorization": f"DeepL-Auth-Key {auth_key}"}
    payload = {"text": [text], "target_lang": target_lang}
    if source_lang:
        payload["source_lang"] = source_lang

    resp = requests.post(url, json=payload, headers=headers, timeout=15)
    resp.raise_for_status()
    data = resp.json()
    tr = data["translations"][0]
    translated = html.unescape(tr["text"])
    return {
        "provider": "deepl",
        "translation": translated,
        "detected_source_language": tr.get("detected_source_language"),
        "target_lang": target_lang,
    }


# MCP ツール定義
@mcp.tool()
async def translate(
    text: str,
    target_lang: str,
    source_lang: Optional[str] = None,
) -> Dict[str, Any]:
    return _call_deepl(text, target_lang, source_lang)


# ツールを自前でも登録（FastMCPが提供しない場合のため）
tools_registry["translate"] = translate

# FastAPI アプリ本体
app = FastAPI()


@app.post("/mcp")
async def call_mcp(request: Request):
    payload = await request.json()
    try:
        tool_name = payload["tool_name"]
        parameters = payload.get("parameters", {})
        tool_func = tools_registry[tool_name]
        result = await tool_func(**parameters)
        return JSONResponse(content={"structuredContent": result})
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})


@app.get("/health")
def health():
    return {"ok": True}

起動コマンド：

uvicorn server.mcp_translate_server:app --port 9999 --reload

② MCPクライアントツール

agents_tools/translate_via_mcp.py

# TODO translate_via_mcp client tool
import os
import httpx
import json
import uuid
from agents.tool import function_tool


MCP_URL = "http://localhost:9999/mcp"

# 二重呼び出し防止フラグと記憶用変数
_last_call_fingerprint = None
ENABLE_DUPLICATE_CALL_GUARD = (
    os.getenv("ENABLE_DUPLICATE_CALL_GUARD", "true").lower() == "true"
)


def _extract_payload(res) -> dict:
    # res が dict の場合（httpxで得たJSON直後）はここで処理
    if isinstance(res, dict):
        payload = res.get("structuredContent")
        if isinstance(payload, dict):
            return payload
        content = res.get("content")
        if isinstance(content, list):
            for item in content:
                if (
                    isinstance(item, dict)
                    and "json" in item
                    and isinstance(item["json"], dict)
                ):
                    return item["json"]
                if (
                    isinstance(item, dict)
                    and "text" in item
                    and isinstance(item["text"], str)
                ):
                    return {"provider": "unknown", "translation": item["text"]}
        raise RuntimeError(f"Unexpected MCP tool payload: {res!r}")

    # res がオブジェクト形式の場合（Agent SDKやMCPクライアント経由など）
    payload = getattr(res, "structuredContent", None)
    if isinstance(payload, dict):
        return payload
    content = getattr(res, "content", None)
    if isinstance(content, list):
        for item in content:
            if (
                isinstance(item, dict)
                and "json" in item
                and isinstance(item["json"], dict)
            ):
                return item["json"]
            if (
                isinstance(item, dict)
                and "text" in item
                and isinstance(item["text"], str)
            ):
                return {"provider": "unknown", "translation": item["text"]}
    raise RuntimeError(f"Unexpected MCP tool payload: {res!r}")


async def _call_mcp_translate(text, target_lang, source_lang):
    payload = {
        "tool_name": "translate",
        "parameters": {
            "text": text,
            "target_lang": target_lang,
            "source_lang": source_lang,
        },
    }

    try:
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.post(MCP_URL, json=payload)
            resp.raise_for_status()
            res = resp.json()
    except Exception as e:

        raise

    payload = _extract_payload(res)
    return payload


@function_tool
async def translate_via_mcp(text: str, target_lang: str, source_lang=None) -> str:
    global _last_call_fingerprint

    fingerprint = (text, target_lang, source_lang)

    if ENABLE_DUPLICATE_CALL_GUARD:
        if fingerprint == _last_call_fingerprint:
            raise RuntimeError("DUPLICATE_TOOL_CALL")
        _last_call_fingerprint = fingerprint
    else:
        print("[TOOL] Duplicate call guard is disabled (env override)")

    try:
        data = await _call_mcp_translate(text, target_lang, source_lang)
    except Exception as e:
        err = f"MCP_TRANSLATE_ERROR: {type(e).__name__}: {e}"
        raise RuntimeError(err)

    provider = data.get("provider") or "deepl"
    translation = data.get("translation")
    if not translation:
        err = f"MCP_TRANSLATE_EMPTY: payload={data!r}"
        raise RuntimeError(err)

    return f"[{provider}] {translation}"

③ MCP Agent

mcp_agent.py

import os
from dotenv import load_dotenv

load_dotenv()

from agents import Agent
from agent_tools.translate_via_mcp import translate_via_mcp

from agents import Agent, OpenAIChatCompletionsModel, set_tracing_disabled
from openai import AsyncAzureOpenAI


client = AsyncAzureOpenAI(
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    api_version=os.getenv("AZURE_OPENAI_API_VERSION"),  # 例: "2024-08-01-preview"
    azure_endpoint=os.getenv(
        "AZURE_OPENAI_ENDPOINT"
    ),  # 例: "https://<your-resource>.openai.azure.com"
)

model = OpenAIChatCompletionsModel(
    openai_client=client,
    model=os.getenv("AZURE_OPENAI_DEPLOYMENT"),  # ← デプロイ"名"（モデル名ではない）
)

set_tracing_disabled(disabled=True)

mcp_agent = Agent(
    name="Translate Agent",
    instructions="""
    あなたは翻訳に関する質問を処理するエージェントです。
    必ず translate_via_mcp ツールを利用してください。
    """,
    model=model,
    tools=[translate_via_mcp],
)

④ 実行例

run_mcp_agent.py

from agents import Runner
from mcp_agent import mcp_agent

if __name__ == "__main__":
    result = Runner.run_sync(
        mcp_agent, "『私は寿司が好きです』を英語に訳してください。"
    )
    print("\n=== 最終回答 ===\n")
    print(result.final_output)

出力例：

=== 最終回答 ===

『私は寿司が好きです』は英語で『I like sushi.』です。

✅ まとめ

MCPを使うと、翻訳APIを安全にラップし、LLMから安心して利用可能になる。

FastAPIで:9999にMCPサーバーを立て、Agents SDKから呼び出すことで翻訳エージェントが簡単に作れる。

マルチエージェントと組み合わせれば「会議メモを英訳」「外国語記事を日本語要約」など応用自在。