LiteLLM × LangGraphでGPTとClaudeを混在させたA2Aエージェント連携を構築してみる

Python 3.13
a2a-sdk 0.3.26
litellm 1.83.14
langgraph 1.1.10
langchain 1.0.0
langchain-litellm 0.6.4
fastapi
uvicorn
httpx

$ uv pip install "a2a-sdk>=0.3,<1.0" litellm langgraph langchain langchain-litellm fastapi uvicorn httpx

"""A2A プロトコル準拠の翻訳エージェントサーバー。

公式 `a2a-sdk` の AgentExecutor で LangChain (ChatLiteLLM) を包み、
JSON-RPC `message/send` を `A2AFastAPIApplication` が処理する。

Agent Card は `/.well-known/agent-card.json` で公開される。

起動: uvicorn translator_server:app --port 8001
"""

import litellm
import uvicorn
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.apps.jsonrpc import A2AFastAPIApplication
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
    Part,
    TextPart,
)
from langchain_litellm import ChatLiteLLM

litellm.modify_params = True

class TranslatorAgentExecutor(AgentExecutor):
    """LangChain (ChatLiteLLM) を A2A AgentExecutor で包む。

    `execute()` が JSON-RPC `message/send` リクエストごとに呼ばれる。
    成果物（翻訳結果）は TaskUpdater で artifact として登録する。
    """

    def __init__(self) -> None:
        # 翻訳品質を担保するため Claude Sonnet を採用
        self._llm = ChatLiteLLM(model="anthropic/claude-sonnet-4-6")

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        user_input = context.get_user_input()
        # クライアントが Message.metadata で指定した skill_id を取り出す
        # 翻訳方向は LLM が入力言語から自動判定し、skill_id ではトーンを切り替える
        skill_id = (context.message.metadata or {}).get("skill_id", "translate_natural")
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.start_work()

        if skill_id == "translate_natural":
            prompt = (
                "次の文を、日常会話やSNSで自然に通じる日本語/英語に翻訳してください。"
                "口語表現・短い文・親しみやすい言い回しを優先します。"
                "翻訳方向は入力言語から自動判定し、翻訳のみ出力してください。\n\n"
                f"{user_input}"
            )
        elif skill_id == "translate_business":
            prompt = (
                "次の文を、ビジネスメールや社外文書としてそのまま使える"
                "フォーマルな日本語/英語に翻訳してください。"
                "敬語・丁寧表現・定型句を適切に使い、口語表現は避けます。"
                "翻訳方向は入力言語から自動判定し、翻訳のみ出力してください。\n\n"
                f"{user_input}"
            )
        else:
            await updater.add_artifact(
                parts=[Part(root=TextPart(text=f"未対応の skill_id: {skill_id}"))]
            )
            await updater.complete()
            return

        response = await self._llm.ainvoke(prompt)
        await updater.add_artifact(
            parts=[Part(root=TextPart(text=str(response.content)))]
        )
        await updater.complete()

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.cancel()

# Agent Card: /.well-known/agent-card.json で公開される自己紹介データ
agent_card = AgentCard(
    name="translator-ja-en",
    description="日本語 ↔ 英語の双方向翻訳エージェント（Claude Sonnet 4.6）",
    url="http://localhost:8001/",
    version="1.0.0",
    capabilities=AgentCapabilities(streaming=True, push_notifications=False),
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    skills=[
        AgentSkill(
            id="translate_natural",
            name="自然訳（カジュアル）",
            description="日本語 ↔ 英語の双方向翻訳。日常会話・SNS で自然に通じる口語スタイル。",
            tags=["translation", "casual", "natural"],
        ),
        AgentSkill(
            id="translate_business",
            name="ビジネス文体訳",
            description="日本語 ↔ 英語の双方向翻訳。ビジネスメール・社外文書向けのフォーマルスタイル。",
            tags=["translation", "business", "formal"],
        ),
    ],
)

task_store = InMemoryTaskStore()
executor = TranslatorAgentExecutor()
request_handler = DefaultRequestHandler(
    agent_executor=executor,
    task_store=task_store,
)

# A2AFastAPIApplication が JSON-RPC ルートと Agent Card 公開を組み立てる
app_builder = A2AFastAPIApplication(
    agent_card=agent_card,
    http_handler=request_handler,
)
app = app_builder.build()

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

$ export ANTHROPIC_API_KEY="sk-ant-..."
$ uvicorn translator_server:app --port 8001
INFO:     Started server process [...]
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8001

# 別ターミナルで Agent Card を取得
$ curl -s http://localhost:8001/.well-known/agent-card.json | jq
{
  "capabilities": {
    "pushNotifications": false,
    "streaming": true
  },
  "defaultInputModes": [
    "text/plain"
  ],
  "defaultOutputModes": [
    "text/plain"
  ],
  "description": "日本語 ↔ 英語の双方向翻訳エージェント（Claude Sonnet 4.6）",
  "name": "translator-ja-en",
  "preferredTransport": "JSONRPC",
  "protocolVersion": "0.3.0",
  "skills": [
    {
      "description": "日本語 ↔ 英語の双方向翻訳。日常会話・SNS で自然に通じる口語スタイル。",
      "id": "translate_natural",
      "name": "自然訳（カジュアル）",
      "tags": [
        "translation",
        "casual",
        "natural"
      ]
    },
    {
      "description": "日本語 ↔ 英語の双方向翻訳。ビジネスメール・社外文書向けのフォーマルスタイル。",
      "id": "translate_business",
      "name": "ビジネス文体訳",
      "tags": [
        "translation",
        "business",
        "formal"
      ]
    }
  ],
  "url": "http://localhost:8001/",
  "version": "1.0.0"
}

"""A2A クライアント: LangGraph オーケストレータが A2A サーバーをツール経由で呼ぶ。

`A2ACardResolver` で Agent Card を取得（`/.well-known/agent-card.json`）し、
`ClientFactory.connect()` で JSON-RPC `message/send` で対話する。
"""

import asyncio
import uuid

import httpx
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import Message, Part, Role, TaskState, TextPart
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_litellm import ChatLiteLLM

TRANSLATOR_URL = "http://localhost:8001"

async def discover_agent(base_url: str):
    """Agent Card を取得して能力を確認する。"""
    async with httpx.AsyncClient() as http_client:
        resolver = A2ACardResolver(httpx_client=http_client, base_url=base_url)
        return await resolver.get_agent_card()

async def call_translator(skill_id: str, text: str) -> str:
    """A2A サーバーへ JSON-RPC で message/send し、結果を返す。

    Card で宣言された skill_id を Message.metadata で指定することで、
    サーバー側の TranslatorAgentExecutor が方向別に分岐できるようにする。
    """
    config = ClientConfig(streaming=False, accepted_output_modes=["text/plain"])
    async with await ClientFactory.connect(TRANSLATOR_URL, client_config=config) as client:
        message = Message(
            message_id=uuid.uuid4().hex,
            role=Role.user,
            parts=[Part(root=TextPart(text=text))],
            metadata={"skill_id": skill_id},
        )
        async for event in client.send_message(message):
            if isinstance(event, tuple):
                task, _ = event
                if task.status.state == TaskState.completed and task.artifacts:
                    for artifact in task.artifacts:
                        for part in artifact.parts:
                            if hasattr(part.root, "text"):
                                return part.root.text
            else:
                for part in event.parts:
                    if hasattr(part.root, "text"):
                        return part.root.text
    return "（応答なし）"

@tool
async def translate_natural(text: str) -> str:
    """日本語 ↔ 英語の双方向翻訳（カジュアル/自然な口語スタイル）。
    日常会話・SNS など親しみやすい文体が必要なときに使う。
    A2A skill: translate_natural"""
    return await call_translator("translate_natural", text)

@tool
async def translate_business(text: str) -> str:
    """日本語 ↔ 英語の双方向翻訳（ビジネスメール・社外文書向けのフォーマル文体）。
    敬語・丁寧表現・定型句を伴う公式な文章が必要なときに使う。
    A2A skill: translate_business"""
    return await call_translator("translate_business", text)

async def main() -> None:
    # まず Agent Card を取得して能力を確認
    card = await discover_agent(TRANSLATOR_URL)
    print(f"接続先エージェント: {card.name}")
    print(f"  説明: {card.description}")
    print(f"  スキル: {[s.name for s in card.skills]}\n")

    # オーケストレータ自身は安価な GPT、翻訳は Claude エージェントに委譲する構成
    llm = ChatLiteLLM(model="openai/gpt-5-mini")
    agent = create_agent(
        model=llm,
        tools=[translate_natural, translate_business],
        system_prompt=(
            "翻訳が必要なときは外部エージェントのツールを使ってください。"
            "ユーザーの依頼から **カジュアル（日常会話・SNS）か、ビジネス（メール・社外文書）か** を読み取り、"
            "translate_natural / translate_business のいずれかを選んでください。"
            "自分で翻訳するのではなく、必ずツール呼び出しで処理してください。"
        ),
    )

    # ビジネスメール向けの翻訳依頼
    result = await agent.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": (
                        "次の日本語をビジネスメールとして送れる英語にして:"
                        "「打ち合わせの件、添付資料をご確認のうえフィードバックをお願いします」"
                    ),
                }
            ]
        }
    )
    print("[business] " + result["messages"][-1].content)

    # カジュアルな翻訳依頼
    result = await agent.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "「明日ヒマだったらランチしようよ」を友達に送るくらいのカジュアルな英語にして",
                }
            ]
        }
    )
    print("[natural ] " + result["messages"][-1].content)

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

# 別ターミナルで A2A サーバーを起動済みとする
$ export OPENAI_API_KEY="sk-..."
$ python orchestrator.py
接続先エージェント: translator-ja-en
  説明: 日本語 ↔ 英語の双方向翻訳エージェント（Claude Sonnet 4.6）
  スキル: ['自然訳（カジュアル）', 'ビジネス文体訳']

[business] Regarding the matter of our upcoming meeting, we kindly request that you review the attached materials and provide us with your feedback at your earliest convenience.
[natural ] If you're free tomorrow, let's grab lunch!

from a2a.types import APIKeySecurityScheme, SecurityScheme

agent_card = AgentCard(
    # ...既存のフィールドはそのまま...
    security_schemes={
        "api_key": SecurityScheme(
            root=APIKeySecurityScheme(name="X-API-Key", in_="header")
        ),
    },
    # api_key を必須要件として宣言
    security=[{"api_key": []}],
)

import os

from fastapi import HTTPException, Request

API_KEY = os.environ["A2A_API_KEY"]
app = app_builder.build()

@app.middleware("http")
async def api_key_middleware(request: Request, call_next):
    # Card 取得は無認証にする運用が一般的（クライアントが Card を見て認証方式を決めるため）
    if request.url.path.startswith("/.well-known/"):
        return await call_next(request)
    if request.headers.get("X-API-Key") != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return await call_next(request)

from a2a.auth.user import User
from a2a.server.apps.jsonrpc.jsonrpc_app import CallContextBuilder
from a2a.server.context import ServerCallContext

# 本番ではこのマッピングを DB / シークレットマネージャーから取得する
API_KEYS: dict[str, str] = {
    "key-team-alpha": "team-alpha",
    "key-team-beta": "team-beta",
}

class APIKeyUser(User):
    """API キー認証済みユーザーの最小実装。"""

    def __init__(self, user_id: str) -> None:
        self._user_id = user_id

    @property
    def is_authenticated(self) -> bool:
        return True

    @property
    def user_name(self) -> str:
        return self._user_id

class APIKeyContextBuilder(CallContextBuilder):
    def build(self, request) -> ServerCallContext:
        key = request.headers.get("X-API-Key", "")
        user_id = API_KEYS.get(key)
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid API key")
        return ServerCallContext(
            user=APIKeyUser(user_id=user_id),
            state={"headers": dict(request.headers)},
        )

app_builder = A2AFastAPIApplication(
    agent_card=agent_card,
    http_handler=request_handler,
    context_builder=APIKeyContextBuilder(),  # ← ここで挿す
)

async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
    user = context.call_context.user if context.call_context else None
    if user is None or not user.is_authenticated:
        # 通常は ContextBuilder で弾けているが、念のため
        ...
    # 例: ユーザーごとのレート制限・課金カウント・監査ログ
    audit_log.info("translate request from user=%s", user.user_name)
    ...

import os

from a2a.client import ClientConfig, ClientFactory
from a2a.client.auth import AuthInterceptor, InMemoryContextCredentialStore

store = InMemoryContextCredentialStore()
await store.set_credential(
    session_id="default",
    security_scheme_name="api_key",
    credential=os.environ["A2A_API_KEY"],
)

config = ClientConfig(
    streaming=False,
    accepted_output_modes=["text/plain"],
    interceptors=[AuthInterceptor(credential_service=store)],
)
client = await ClientFactory.connect(TRANSLATOR_URL, client_config=config)