A2A

一行サマリ

app = agent.to_a2a() の 1 行 で PydanticAI Agent を Agent-to-Agent (A2A) プロトコル準拠の Starlette ASGI アプリ として公開でき、uvicorn でそのまま起動できる。他フレームワーク (LangChain / Crew / 他言語) や他社の Agent から 標準プロトコル経由で呼び出してもらえる ようになる。

ヒーロー: Agent を「外部公開する標準形」

社内 Agent をマイクロサービス化したい、自社の Agent を他社 Agent から呼べるようにしたい ── そのたびに 独自 REST API を設計 していたら相互運用は永遠に成立しません。A2A (Agent-to-Agent) プロトコル は、Agent 間通信のための オープンな標準仕様 で、Pydantic の FastA2A がこれを実装しています。

概念: A2A の中核 3 概念

クライアント側は Agent Card を読んで「何ができる Agent か」を判断し、Task を投げて結果を受け取る、というシンプルなライフサイクル。

コード: 3 つのパターン

パターン 1: Agent を A2A サーバーとして公開

# agent_to_a2a.py
from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel
 
agent = Agent(
    GoogleModel('gemini-3-flash-preview'),
    name='joke_agent',
    instructions='与えられたお題で短い日本語のジョークを 1 つ作ってください。',
)
 
# これだけで Starlette ASGI アプリになる
app = agent.to_a2a()

起動コマンド:

uvicorn agent_to_a2a:app --host 0.0.0.0 --port 8000

ポイント:

• agent.to_a2a() の戻り値は Starlette アプリ。FastAPI / Uvicorn / Gunicorn で動かせる
• name='joke_agent' は Agent Card に出る識別子
• 必要パッケージ: pip install 'pydantic-ai-slim[a2a]'

パターン 2: 既存の Agent (tool 付き) もそのまま公開

from pydantic_ai import Agent, RunContext
from pydantic_ai.models.google import GoogleModel
 
agent = Agent(
    GoogleModel('gemini-3-flash-preview'),
    name='weather_agent',
    instructions='都市名を受け取り get_weather で天気を返してください。',
)
 
@agent.tool
async def get_weather(ctx: RunContext, city: str) -> str:
    """指定都市の現在の天気を返す。"""
    table = {'Tokyo': '晴れ', 'Osaka': '曇り'}
    return table.get(city, '不明')
 
app = agent.to_a2a()

tool 付き Agent も 何も変更せず 公開可能。クライアント側は Task を投げるだけで Agent が必要な tool を内部で呼んでくれます。

パターン 3: A2A クライアントから呼ぶ (擬似)

# Pydantic AI 側のクライアント API は将来拡充予定。現状の使い方として:
import httpx
 
client = httpx.Client(base_url='http://localhost:8000')
 
# Agent Card 取得
card = client.get('/.well-known/agent.json').json()
print(card)
# {'name': 'joke_agent', 'description': '...', 'capabilities': [...]}
 
# Task 投入 (A2A 仕様に従った payload)
res = client.post('/tasks', json={
    'message': {'role': 'user', 'parts': [{'text': '猫'}]},
})
task = res.json()
 
# Task 完了を polling で待つ (実装によっては SSE で push される)
while task['status'] != 'completed':
    task = client.get(f'/tasks/{task["id"]}').json()
 
# Artifact から結果取得
print(task['artifacts'][0]['parts'][0]['text'])

A2A クライアントは 言語非依存。TypeScript / Go / Rust 等、HTTP が話せる環境なら同じプロトコルで呼び出せます。

観察: ストレージ・ブローカー・ワーカー

FastA2A は内部で 3 部品を必要とします:

• Storage: Task / Context を永続化 (デフォルトはインメモリ。本番では Redis / Postgres)
• Broker: Task の実行スケジューラ
• Worker: 実際に Agent を回す実行体

agent.to_a2a() のデフォルトはインメモリで動くのでローカル開発には十分。本番では Storage を外部 KV/RDB に切り替える のが定石です。

まとめ

• app = agent.to_a2a() で Starlette ASGI アプリ として A2A サーバー化
• Agent Card / Task / Context の 3 概念で他フレームワーク・他言語の Agent と相互運用
• tool 付き Agent もそのまま公開できる
• 本番では Storage を Redis / Postgres に外部化 + 認証 (mTLS / OIDC / API Key) 必須
• 長時間ストリームは A2A ではなく SSE / WebSocket と組み合わせ

🎉 Ch10 完結 (Production) + Production グループ完結 + 全 11 章完走 🎊

• ✅ L01 Durable Execution (Temporal) / L02 UI Event Streams / L03 A2A
• ✅ Production グループ (Ch9-Ch10) 全 6 レッスン完走
• ✅ 全 11 章 34 レッスン完走 — Foundation (9) + Core (10) + Advanced (9) + Production (6)

PydanticAI ビジュアルガイドのカリキュラムはここで完成です。最初の Agent('google-gla:gemini-3-flash-preview', ...).run_sync('Hello') から、本番運用で必要な 観測・評価・耐久実行・UI 連携・Agent 間相互運用 までを 1 本の動線で学んできました。次は実際のプロジェクトに適用し、独自の tool / 構造化出力 / マルチエージェントを設計してみてください。