マルチエージェント設計の議論で必ず出てくるのが「エージェント同士はどうやって会話するのか」という問いです。Antigravity には A2A（Agent-to-Agent）プロトコルという仕組みが用意されていて、これを使うと、あるエージェントが別のエージェントに作業を依頼し、結果を受け取る流れを型として書けます。
ただ、公式ドキュメントだけを読むと「どういうときに使うのか」がつかみにくいので、ここでは実務で有効なパターンを 3 つ、動くサンプルを添えて紹介します。
A2A プロトコルの 1 分サマリー
A2A プロトコルは、エージェントを 1 つの HTTP サービスとして扱い、別のエージェントがそのサービスに対して構造化されたリクエストを送ることで会話を成立させます。中心概念は 3 つです。
- Agent Card: エージェントが自己申告する「何ができるか」のスキーマ。公開 URL・対応タスク・認証方式を含みます
- Task: 依頼を送る側が作成するオブジェクト。入力メッセージとメタデータを持ちます
- Message Stream: タスクの進行中に発生するメッセージ列。途中経過・確認の問い返し・最終結果が流れます
図にすると、「依頼する側（クライアント・エージェント）」「引き受ける側（リモート・エージェント）」「両者をつなぐタスク」の三角関係です。
最小実装 — 2 エージェントのタスク委譲
まず最小構成で動かしてみます。エージェント A が「要約してほしい記事テキスト」をエージェント B に渡し、B が要約を返す、という単純なケースです。
# agent_b_server.py ― 要約を担当するエージェント
from antigravity.a2a import AgentServer, AgentCard, TaskHandler
card = AgentCard(
name="summarizer",
description="Summarize long-form text in 3 bullet points.",
supported_tasks=["text/summarize"],
endpoint="http://localhost:8081",
)
handler = TaskHandler(card=card)
@handler.on_task("text/summarize")
async def summarize(task):
text = task.input["text"]
result = await task.call_llm(
prompt=f"Summarize this text in 3 bullet points:\n\n{text}",
)
return {"summary": result}
if __name__ == "__main__":
AgentServer(handler).run(port=8081)# agent_a_client.py ― 依頼する側
from antigravity.a2a import AgentClient
async def main():
client = AgentClient(remote="http://localhost:8081")
result = await client.submit_task(
kind="text/summarize",
input={"text": open("article.md").read()},
)
print(result["summary"])ポイントは 2 つあります。1 つは Agent Card で「何ができるか」を宣言的に記述している点。もう 1 つは、タスクの種類（kind）を文字列 ID で揃えている点です。この 2 つが決まっていれば、エージェント A は B の実装を知らなくても委譲できます。
パターン 1: 単方向委譲（Fire-and-Forget 型）
上のサンプルのように、A から B にタスクを投げ、B が結果を返す一方向の流れです。最も単純で、最も多用する形です。
適しているケースとしては、次のような条件が揃うときです。
- タスクの入出力が明確に定義されている（要約、翻訳、分類、データ整形）
- 中間で人間の確認が不要
- 結果が得られたら A 側の処理は完結する
この形は REST API 呼び出しと本質的に同じですので、「A2A を使う意義はあるのか」と感じられるかもしれません。答えは、エージェントを差し替え可能なコンポーネントとして扱える点にあります。同じ text/summarize タスクを受け付ける別実装（性能の違う LLM・別言語特化・無料版）に切り替えるときに、クライアント側のコードを変えずに済みます。
パターン 2: 双方向対話（途中で確認を取る型）
タスクの途中でエージェント B が A に確認したくなるケースもあります。たとえば B が資料生成中に「タイトルはこれで良いですか？」と聞き返したい、といった状況です。A2A は Message Stream で双方向のメッセージをやり取りできるため、これを型として書けます。
@handler.on_task("doc/generate")
async def generate_doc(task):
draft_title = await task.call_llm(prompt="Propose a title for this report: ...")
# A にタイトル候補を確認する
response = await task.ask(
message=f"Is this title OK? '{draft_title}'. Reply 'yes' or a new title.",
)
final_title = draft_title if response.strip().lower() == "yes" else response.strip()
# ... 以下、本文生成 ...
return {"title": final_title, "body": body}このパターンは、完全自動化したいけれども重要な分岐点では人間的判断を挟みたい、というニーズに合います。A 側の実装によっては、確認要求が来たときに別エージェント（権限を持つ上位エージェント）にさらに委譲することも可能です。
パターン 3: 並列分散（Scatter-Gather 型）
A が同時に複数の B1・B2・B3 にタスクを投げ、結果を集約するパターンです。大量の類似タスクを高速化したいときに効きます。
import asyncio
async def process_articles(article_list):
clients = [AgentClient(remote=url) for url in worker_endpoints]
tasks = [
clients[i % len(clients)].submit_task(
kind="text/summarize",
input={"text": text},
)
for i, text in enumerate(article_list)
]
results = await asyncio.gather(*tasks)
return [r["summary"] for r in results]シンプルなキュー＋ワーカーと思えるかもしれませんが、A2A 経由で実装しておく利点は、ワーカーごとに異なる能力（日本語特化・英語特化・コード特化）を持たせられる点です。tasks を構築する段階で「日本語記事は日本語特化ワーカーに」「コード記事はコード特化ワーカーに」と振り分けるロジックを挟むと、単純な並列処理以上の価値が出てきます。
実装で気をつけたい 3 つの落とし穴
認証とアクセス制御: A2A エンドポイントをそのままインターネットに公開すると、任意のクライアントからタスクを受け取ってしまいます。Agent Card に auth 要件を書き、Bearer トークンまたは mTLS で絞ることをおすすめします。社内ネットワーク内でも、VPN 越しのアクセスを想定するなら最低限のトークン認証は必須です。
タイムアウトと再送: リモートエージェントが応答しないとき、クライアント側で無限に待たないようタイムアウトを設定してください。加えて、同じタスクを 2 回送っても副作用が重複しないように、タスクに idempotency_key を付与する設計が安全です。
エラー情報の粒度: リモートエージェント内で発生した例外をそのまま A に返すと、内部実装が漏れます。TaskError(code="llm_timeout", message="...", retry_after=30) のような形でコード化し、受け取った側が意味のあるフォールバックを取れるようにするとよいです。
どこから試すと良いか
本記事のサンプルは、Antigravity の Python SDK をインストールして 2 つのターミナルで動かすだけで体験できます。まずは単方向委譲（パターン 1）を動かしてみて、感覚がつかめたら双方向対話に進むのが無理のない順番です。
複雑なマルチエージェント構成を設計する段階に入ったら、Antigravity マルチエージェント・オーケストレーション入門 も合わせて読んでみてください。A2A をどこで使い、どこで避けるかの判断基準が得られるはずです。