A2A（Agent-to-Agent）プロトコルは、独立して動作するエージェント同士が相互に依頼・応答・進捗共有できるようにするための共通仕様です。Google が2025年後半から本格的に提唱を始め、Antigravity でも対応が進んでいます。1つの大きなエージェントを作るより、役割分担した複数の小さなエージェントを連携させた方が、保守も拡張も楽になることが多いのですが、その「連携」を標準化したのが A2A です。
ここではAntigravity 上で A2A を使ったエージェント間通信の実装を、最小サンプルから本番運用レベルまで段階的に組み上げていきます。私自身がプロトタイプ開発で何度も書き直してたどり着いた「動かしやすい構成」を共有します。
なぜ単一エージェントではなく A2A なのか
「全てを1つのエージェントに任せる」のは、最初は速いです。プロンプトを長く書いてツールを増やせばたいてい何でもこなせます。しかし規模が大きくなると、次の問題が次々に出てきます。
第一に、コンテキストウィンドウの飽和です。1つのエージェントに全責任を負わせると、過去の会話・ツールの定義・ドキュメント・例示が膨れ上がり、肝心の判断のためのトークン余裕がなくなります。
第二に、評価が難しくなります。失敗したとき「どの判断ステップが間違ったか」を後から追えません。専門エージェントごとに切り出してあれば、ログから「ここで失敗した」が即座に分かります。
第三に、ツール権限の設計が雑になります。ファイル書き込み、外部 API 呼び出し、決済処理など、同じエージェントに集約していると、思わぬ場面で危険なツールが呼ばれます。
A2A はこれらを解消するために、エージェント同士を「サービス」として扱います。各エージェントは公開された Capability（できること）と Endpoint（受信口）を持ち、リクエストとレスポンスを構造化されたメッセージで交換します。
A2A メッセージの基本構造
A2A は基本的に JSON ベースのリクエスト/レスポンス形式です。Antigravity の Manager Agent から Surface Agent を呼ぶ場合の最小構造は次のような形になります。
{
"jsonrpc": "2.0",
"id": "req-001",
"method": "agent.invoke",
"params": {
"capability": "code_review",
"input": {
"diff": "...",
"language": "typescript"
},
"metadata": {
"trace_id": "trace-abc-123",
"deadline_ms": 30000
}
}
}JSON-RPC 2.0 がベースになっていて、method は通常 agent.invoke または agent.stream、params の中に呼び出したい Capability と入力を入れます。
レスポンスは次のような形です。
{
"jsonrpc": "2.0",
"id": "req-001",
"result": {
"status": "completed",
"output": {
"issues": [
{"severity": "warning", "line": 42, "message": "..."}
],
"summary": "..."
},
"metadata": {
"tokens_used": 1820,
"elapsed_ms": 8421
}
}
}ここで重要なのは、metadata です。トレース ID とトークン使用量を返すのは A2A 仕様で強く推奨されており、これがあるかないかで本番運用での観測可能性が大きく変わります。
Antigravity でのエージェント定義
Antigravity 上で A2A 対応エージェントを定義するには、エージェントの設定ファイルに capabilities セクションを追加します。
# .antigravity/agents/code-reviewer.yaml
name: code-reviewer
description: "TypeScript / Python のコードレビュー専門エージェント"
model: gemini-3-flash
a2a:
enabled: true
capabilities:
- id: code_review
description: "差分コードのレビュー"
input_schema:
type: object
properties:
diff: {type: string}
language: {type: string, enum: [typescript, python, go]}
required: [diff, language]
output_schema:
type: object
properties:
issues: {type: array}
summary: {type: string}
auth:
method: api_key
header: X-A2A-Tokena2a.enabled: true で受信エンドポイントが有効化され、capabilities で公開する機能を JSON Schema 形式で定義します。スキーマがあると呼び出し側が間違った入力を渡したときに即座にエラーで弾けるので、デバッグが格段に楽になります。
呼び出し側エージェントの実装
呼び出す側のエージェント（たとえば Manager Agent）は、A2A クライアントを使ってメッセージを送ります。Antigravity SDK を使った実装例です。
from antigravity.a2a import A2AClient
client = A2AClient(
target_agent="code-reviewer",
auth={"api_key": os.environ["CODE_REVIEWER_TOKEN"]}
)
result = await client.invoke(
capability="code_review",
input={
"diff": diff_content,
"language": "typescript"
},
metadata={
"trace_id": trace_id,
"deadline_ms": 30000
}
)
if result.status == "completed":
issues = result.output["issues"]
# 後続処理
elif result.status == "failed":
logger.error(f"Code review failed: {result.error}")deadline_ms は重要なパラメータです。これを設定しないと、呼ばれたエージェントが無限ループに入ったときに呼び出し側が永遠に待つことになります。本番では必ず妥当な値（私は基本30秒、長い処理でも60秒）を設定してください。
ストリーミング応答の扱い
長時間かかるエージェント呼び出しでは、進捗を返したいことがあります。A2A では agent.stream メソッドで部分応答を返すことができます。
async for chunk in client.stream(
capability="long_analysis",
input=large_dataset
):
if chunk.type == "progress":
print(f"進捗: {chunk.percentage}%")
elif chunk.type == "partial_result":
# 中間結果を UI に反映
update_ui(chunk.data)
elif chunk.type == "completed":
final_result = chunk.data
breakストリーミングは特にユーザー対面のアプリケーションで効果的です。「3分待たされて結果が返ってくる」よりも「30秒ごとに途中経過が見える」方が体感が大きく違います。
認証と権限スコープ
A2A では認証は実装側に委ねられていますが、最低限 API キーまたは OIDC トークンを使うべきです。私が実運用で使っている構成は次のようなものです。
auth:
method: oidc
issuer: https://auth.example.com
audience: a2a://code-reviewer
scopes:
- capability:code_review:invoke
- metadata:trace:writeスコープを capability 単位で切ると、たとえば「外部公開エージェントは code_review だけ呼べる、内部エージェントは code_review と auto_fix を呼べる」という権限分離が綺麗にできます。
これを最初からやっておくと、後でエージェントを社外公開するときに認証回りで悩まずに済みます。
エラーハンドリングの設計
A2A のエラーは JSON-RPC のエラーコード体系に従います。私が本番で対応している主なエラーは次の通りです。
try:
result = await client.invoke(...)
except A2AError as e:
if e.code == -32000:
# Server error: エージェント側の内部エラー（再試行可能）
await retry_with_backoff()
elif e.code == -32602:
# Invalid params: 入力が schema に合わない（即時失敗）
logger.error(f"Bad input: {e.data}")
raise
elif e.code == -32603:
# Internal error: フレームワーク側のバグ（運用に通知）
notify_oncall(e)
elif e.code == 408:
# Timeout: deadline_ms を超えた
# → 部分結果が返っているか確認、必要なら再試行
if e.data.get("partial_result"):
return e.data["partial_result"]
await retry_with_longer_deadline()特に Timeout（408）の扱いは丁寧にやるのがおすすめです。部分結果が返っている場合があるので、そのまま捨てると無駄なコストになります。
メッセージのトレーシング
複数のエージェントが連鎖して呼ばれると、どこで何が起きたか追えなくなりがちです。trace_id を最初のリクエストで作って、後続の全 A2A 呼び出しに引き回すのが基本です。
async def handle_user_request(user_input):
trace_id = generate_trace_id()
plan = await planner.invoke(
capability="plan",
input={"goal": user_input},
metadata={"trace_id": trace_id}
)
for step in plan.output["steps"]:
result = await executor.invoke(
capability="execute_step",
input=step,
metadata={"trace_id": trace_id} # 同じ trace_id を引き継ぐ
)これで OpenTelemetry や独自のトレースバックエンドに集約すれば、「ユーザーの1回のリクエストが内部で何個のエージェントを呼び、それぞれ何ミリ秒かかったか」が1つのタイムラインで見えます。
失敗の局所化と再試行
複数エージェントが連鎖していると、1つが失敗したときの影響範囲が問題になります。私が採用している方針は「失敗は局所化する」です。
呼び出し側は、呼んだエージェントが失敗しても、それを上位に投げ返す前に「このエージェントが失敗したことで全体が失敗する必要があるか」を判断します。たとえばコードレビューが失敗してもデプロイは続行できることが多いので、ログを残してスキップします。
try:
review_result = await reviewer.invoke(...)
findings = review_result.output["issues"]
except A2AError as e:
logger.warning(f"Review skipped due to: {e}")
findings = [] # 空リストで続行逆に、計画立てるエージェントが失敗したときは続行不可能なので、即座に上位に伝える。これを各エージェントごとに整理して、ハンドブックに書いておくと、運用時の判断が早くなります。
本番投入前のチェックリスト
A2A を本番に投入する前に、最低限次は確認しています。
入力スキーマと出力スキーマが定義され、CI で検証されているか。deadline_ms が全呼び出しに設定されているか。認証が API キー以上の強度で実装されているか。trace_id が全エージェント間で引き継がれているか。エラー時の上位への伝播ポリシーが文書化されているか。
これらが揃っていれば、エージェント間の連携が壊れたときも、原因を素早く特定できます。
実装の出発点
最初から大規模に組もうとせず、まず2つのエージェントを A2A で繋いでみてください。Manager Agent が Reviewer Agent を呼ぶ、というシンプルな構成で十分です。これだけでもメッセージのフォーマット、認証、エラー処理、トレーシングの基本が体得できます。
そこから3つ目、4つ目と増やしていくときに、この記事のチェックリストを横に置いて参照していただけると、つまずきが減ると思います。マルチエージェント設計は楽しい領域ですが、A2A のような共通プロトコルがないと急速に混沌に向かいます。最初にきちんと足場を組んでから走り出すのがおすすめです。