Google A2Aプロトコルとは？MCPとの違いと実装方法

なぜ今「エージェント間通信」が必要なのか

2025年から2026年にかけて、AIエージェントの開発と導入は爆発的に加速しています。Gartnerの予測では、2028年までにエンタープライズソフトウェアの33%にAIエージェント機能が組み込まれるとされています。しかし、ここで大きな課題が浮上しました。異なるフレームワークやベンダーで構築されたエージェント同士が、直接コミュニケーションを取れないという問題です。

例えば、社内に「顧客対応エージェント（Salesforce Agentforce上）」と「在庫管理エージェント（SAP Joule上）」と「データ分析エージェント（自社開発）」が存在するケースを考えてみましょう。顧客から「在庫があれば注文したい」と問い合わせがあった場合、顧客対応エージェントは在庫管理エージェントに問い合わせ、結果に応じてデータ分析エージェントに需要予測を依頼する必要があります。しかし、それぞれのエージェントが異なるプラットフォーム上にあるため、個別のAPI連携を手作業で構築するしかないのが現状でした。

この「AIエージェントのサイロ化」問題を解決するために、Googleが2025年4月に発表したのがA2A（Agent-to-Agent）プロトコルです。A2Aは、AIエージェントが互いの能力を発見し、タスクを委任し、結果を受け取るための共通プロトコルを定義します。いわば「AIエージェント版HTTP」とも言える存在です。

A2Aの全体アーキテクチャ

基本構成: クライアントとリモートエージェント

A2Aプロトコルは、大きく分けて2つの役割で構成されます。

┌─────────────────┐          JSON-RPC 2.0 / HTTP(S)          ┌─────────────────┐
│                 │  ──── Agent Card 発見 ──────────────────▶ │                 │
│  Client Agent   │  ──── SendMessage（タスク送信）──────────▶ │  Remote Agent   │
│  （依頼する側）  │  ◀─── TaskStatusUpdate（進捗通知）──────  │  （実行する側）  │
│                 │  ◀─── TaskArtifactUpdate（成果物）──────  │                 │
└─────────────────┘                                           └─────────────────┘
        │                                                              │
        │  ユーザーからの                                               │  タスクを処理し
        │  リクエストを受け取り、                                        │  結果を返す
        │  最適なエージェントに委任                                      │
        ▼                                                              ▼
   ┌──────────┐                                                  ┌──────────┐
   │  ユーザー  │                                                  │  ツール群  │
   └──────────┘                                                  └──────────┘

• Client Agent（クライアント）: ユーザーからのリクエストを受け取り、適切なリモートエージェントを発見して、タスクを委任する側
• Remote Agent（リモート / A2Aサーバー）: タスクを受け取り、処理を実行して結果を返す側

通信の流れ

A2Aの通信は、以下の4ステップで行われます。

1. 発見（Discovery）: クライアントがリモートエージェントの /.well-known/agent-card.json を取得し、能力・スキル・認証要件を確認
2. タスク送信（SendMessage）: JSON-RPC 2.0形式でタスクをリモートエージェントに送信
3. 処理と更新（Working + Updates）: リモートエージェントがタスクを処理しながら、ステータス更新やストリーミングで進捗を通知
4. 完了（Completed / Failed）: 成果物（Artifact）とともにタスク完了を通知。必要に応じて追加入力を要求（input-required）

プロトコルの設計原則

Googleは、A2Aの設計に際して5つの原則を掲げています。

技術的な仕組み: Agent Card・Task・Streaming

Agent Card: エージェントの「名刺」

Agent Cardは、A2Aプロトコルの最も重要なコンポーネントです。各エージェントが自分の能力を外部に公開するためのJSON形式のメタデータドキュメントで、/.well-known/agent-card.json というパスでホストされます。

Agent Cardには以下の情報が含まれます。

• name / description: エージェントの名前と説明
• url: A2Aサービスのエンドポイント
• version: プロトコルバージョン
• capabilities: ストリーミング対応、プッシュ通知対応などの機能フラグ
• skills: エージェントが提供するスキル一覧（ID、名前、説明、タグ、入出力モード）
• defaultInputModes / defaultOutputModes: 対応するデータ形式（text, image, audio など）
• authentication: 認証要件（OpenAPI Auth構造に準拠）
• provider: サービス提供者情報

以下は、Agent Cardの具体例です。

{
  "name": "在庫管理エージェント",
  "description": "商品の在庫確認、入出庫管理、需要予測を行うエージェント",
  "url": "https://inventory.example.com/a2a",
  "version": "1.0.0",
  "defaultInputModes": ["text"],
  "defaultOutputModes": ["text"],
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "check_stock",
      "name": "在庫確認",
      "description": "指定商品の現在庫数と入荷予定を返す",
      "tags": ["inventory", "stock", "availability"],
      "examples": ["商品Aの在庫を確認して", "SKU-12345の在庫数は？"]
    },
    {
      "id": "demand_forecast",
      "name": "需要予測",
      "description": "過去データに基づく需要予測を生成",
      "tags": ["forecast", "analytics", "demand"],
      "examples": ["来月の商品Aの需要予測を出して"]
    }
  ],
  "authentication": {
    "schemes": ["bearer"],
    "credentials": "OAuth 2.0 Bearer Token required"
  },
  "provider": {
    "organization": "Example Corp",
    "url": "https://example.com"
  }
}

Task ライフサイクル: タスクの一生

A2Aでは、すべての作業がTask（タスク）オブジェクトとして管理されます。タスクには一意のIDが割り当てられ、定義された状態遷移に従って進行します。

                    ┌──────────┐
                    │ submitted │
                    └─────┬────┘
                          │
                    ┌─────▼────┐     ┌──────────────┐
              ┌────▶│ working   │────▶│ input-required│──┐
              │     └─────┬────┘     └──────────────┘  │
              │           │                             │
              │     ┌─────▼────┐                        │
              └─────┤  (再処理)  │◀───────────────────────┘
                    └─────┬────┘
                          │
          ┌───────────────┼───────────────┐
          ▼               ▼               ▼
   ┌───────────┐  ┌───────────┐  ┌───────────┐
   │ completed  │  │  failed   │  │ canceled  │
   └───────────┘  └───────────┘  └───────────┘
     （終了状態）    （終了状態）    （終了状態）

各状態の意味は以下の通りです。

Streaming: リアルタイム更新

A2Aは、長時間実行タスクのリアルタイム進捗通知をサポートしています。ストリーミングでは、2種類のイベントが送信されます。

• TaskStatusUpdateEvent: タスクの状態変化を通知（working → input-required → completed など）
• TaskArtifactUpdateEvent: 成果物のチャンク（部分データ）を逐次送信

ストリーミングはSSE（Server-Sent Events）を使用して実装されており、message/stream エンドポイントで利用できます。クライアントは AsyncGenerator として受信し、各チャンクを逐次処理できます。

JSON-RPC メソッド一覧

A2Aプロトコルは、以下のJSON-RPCメソッドを定義しています。

MCPとの違い・使い分け

A2Aと並んで語られることが多いのが、Anthropicが提唱するMCP（Model Context Protocol）です。「A2AとMCPのどちらを使うべきか？」という質問をよく見かけますが、結論から言えば両者は競合ではなく補完関係にあります。Google自身もA2AはMCPと補完的に使うことを意図していると明言しています。

比較表: A2A vs MCP

いつどちらを使うか？

シンプルに考えると、以下の使い分けになります。

• 「エージェントに新しいツールを使わせたい」→ MCP: データベースへのアクセス、外部APIの呼び出し、ファイル操作など
• 「エージェント同士を連携させたい」→ A2A: 異なるチーム・ベンダーが開発したエージェント間の協調作業
• 両方必要なケースが多い: 各エージェントがMCP経由でツールにアクセスしつつ、A2A経由で他のエージェントと連携する構成が一般的

ハンズオン: Python SDKで動かしてみる

ここからは、A2A公式Python SDK（a2a-sdk）を使って、実際にA2A対応エージェントを構築してみましょう。

環境構築

# Python 3.10以上が必要
pip install "a2a-sdk[all]"

# もしくは最小構成でインストール
pip install a2a-sdk

コード例1: リモートエージェント（サーバー側）の実装

まず、A2Aサーバーとして動作するリモートエージェントを実装します。このエージェントは、為替レートを返す簡単なスキルを持っています。

# agent_executor.py
from a2a.server.agent_execution import AgentExecutor
from a2a.server.events import EventQueue
from a2a.server.context import RequestContext
from a2a.utils import new_agent_text_message


class CurrencyAgent:
    """為替レートを返すエージェント"""

    RATES = {
        "USD_JPY": 149.50,
        "EUR_JPY": 162.30,
        "GBP_JPY": 189.80,
    }

    async def invoke(self, query: str) -> str:
        query_upper = query.upper()
        for pair, rate in self.RATES.items():
            src, dst = pair.split("_")
            if src in query_upper and dst in query_upper:
                return f"現在の{src}/{dst}レートは {rate} です（※デモ用固定値）"
        return "対応する通貨ペアが見つかりませんでした。USD/JPY, EUR/JPY, GBP/JPY に対応しています。"


class CurrencyAgentExecutor(AgentExecutor):
    """A2Aプロトコル対応のAgentExecutor実装"""

    def __init__(self):
        self.agent = CurrencyAgent()

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        # ユーザーメッセージからテキストを抽出
        user_message = context.message
        query = ""
        for part in user_message.parts:
            if part.kind == "text":
                query = part.text
                break

        # エージェントのビジネスロジックを実行
        result = await self.agent.invoke(query)

        # 結果をイベントキューに送信
        await event_queue.enqueue_event(
            new_agent_text_message(result)
        )

    async def cancel(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        raise Exception("キャンセルは未対応です")

コード例2: サーバーの起動設定

次に、Agent CardとA2Aサーバーのセットアップを行う __main__.py を作成します。

# __main__.py
import uvicorn

from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)
from agent_executor import CurrencyAgentExecutor


if __name__ == "__main__":
    # 1. スキルを定義
    skill = AgentSkill(
        id="currency_rate",
        name="為替レート照会",
        description="主要通貨ペアの為替レートを返します",
        tags=["currency", "forex", "exchange rate"],
        examples=[
            "USD/JPYのレートを教えて",
            "ユーロ円はいくら？",
        ],
    )

    # 2. Agent Cardを作成
    agent_card = AgentCard(
        name="為替レートエージェント",
        description="主要通貨ペアの最新レートを提供するエージェント",
        url="http://localhost:9999/",
        version="1.0.0",
        default_input_modes=["text"],
        default_output_modes=["text"],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],
    )

    # 3. リクエストハンドラーを設定
    request_handler = DefaultRequestHandler(
        agent_executor=CurrencyAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    # 4. A2Aサーバーを構築・起動
    server = A2AStarletteApplication(
        agent_card=agent_card,
        http_handler=request_handler,
    )

    # ポート9999でサーバーを起動
    # Agent Cardは /.well-known/agent-card.json で自動公開される
    uvicorn.run(server.build(), host="0.0.0.0", port=9999)

コード例3: クライアントからの呼び出し

最後に、別のエージェント（またはテストスクリプト）からA2Aサーバーにリクエストを送信するクライアント実装です。

# test_client.py
import asyncio
from uuid import uuid4

import httpx
from a2a.client import A2AClient, A2ACardResolver
from a2a.types import (
    MessageSendParams,
    SendMessageRequest,
    SendStreamingMessageRequest,
)


async def main():
    base_url = "http://localhost:9999"

    async with httpx.AsyncClient() as httpx_client:
        # 1. Agent Cardを取得してクライアントを初期化
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=base_url,
        )
        agent_card = await resolver.get_agent_card()
        print(f"接続先エージェント: {agent_card.name}")
        print(f"スキル: {[s.name for s in agent_card.skills]}")

        client = A2AClient(
            httpx_client=httpx_client,
            agent_card=agent_card,
        )

        # 2. 同期リクエスト（message/send）
        payload = {
            "message": {
                "role": "user",
                "parts": [
                    {"kind": "text", "text": "USD/JPYのレートを教えて"}
                ],
                "messageId": uuid4().hex,
            },
        }

        request = SendMessageRequest(
            id=str(uuid4()),
            params=MessageSendParams(payload),
        )

        response = await client.send_message(request)
        print("同期レスポンス:", response.model_dump(
            mode="json", exclude_none=True
        ))

        # 3. ストリーミングリクエスト（message/stream）
        stream_request = SendStreamingMessageRequest(
            id=str(uuid4()),
            params=MessageSendParams(payload),
        )

        print("nストリーミングレスポンス:")
        stream = client.send_message_streaming(stream_request)
        async for chunk in stream:
            print("  chunk:", chunk.model_dump(
                mode="json", exclude_none=True
            ))


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

動作確認

以下の手順で動作を確認できます。

# ターミナル1: サーバーを起動
python __main__.py
# → INFO:     Uvicorn running on http://0.0.0.0:9999

# ターミナル2: Agent Cardを確認
curl http://localhost:9999/.well-known/agent-card.json | python -m json.tool

# ターミナル2: クライアントを実行
python test_client.py
# → 接続先エージェント: 為替レートエージェント
# → スキル: ['為替レート照会']
# → 同期レスポンス: {"result": {"id": "...", "status": {"state": "completed"}, ...}}

Google ADKとの連携

Google ADK（Agent Development Kit）を使えば、既存のADKエージェントをA2A対応に変換することも簡単です。ADKは to_a2a() メソッドを提供しており、数行のコードでA2Aサーバーとして公開できます。

逆に、他のA2Aサーバーを呼び出す側では RemoteA2aAgent クラスを使います。これにより、ネットワーク通信・認証・データ形式変換がすべて抽象化され、ローカルのツールを呼び出すのと同じ感覚でリモートエージェントを利用できます。

パフォーマンスと制限事項

パフォーマンス特性

• 通信オーバーヘッド: HTTP(S) + JSON-RPCベースのため、ローカル関数呼び出しと比較して数ms〜数十msのレイテンシが追加される。v0.3ではgRPCもサポートされ、高頻度通信のケースではさらに低レイテンシが実現可能
• ストリーミング: SSEベースのストリーミングにより、長時間タスクでもクライアントにリアルタイムで進捗通知が可能。ただし、SSEは単方向（サーバー → クライアント）であり、双方向リアルタイム通信には向かない
• スケーラビリティ: 各エージェントが独立したHTTPサーバーとして動作するため、水平スケーリングは容易。ロードバランサー配下での運用も標準的なWebサービスと同様に可能

現在の制限事項

• プロトコルの成熟度: 2026年3月時点の最新バージョンはv0.3。仕様はまだ安定化の途上にあり、マイナーバージョン間でbreaking changeが発生する可能性がある
• エージェント発見のスケール: Agent Cardは特定のURLから取得する仕組みであり、「世界中のA2Aエージェントを検索する」ようなグローバルディレクトリは公式には存在しない（企業内レジストリを構築する必要がある）
• 認証の複雑さ: エンタープライズ環境ではOAuth 2.0フローの実装が必要で、複数ベンダー間の認証連携は設計・運用の両面で複雑になる
• デバッグの難しさ: 分散システム特有の問題として、複数エージェント間のリクエストチェーンのデバッグは容易ではない。OpenTelemetryへの対応が進んでいるが、ツールチェーンはまだ発展途上

ハマりやすいポイント

1. Agent Cardのパス設定

# Agent Cardを /api/agent-card など独自パスで公開
# → 他のA2Aクライアントが発見できない

# A2A仕様に従い、必ず /.well-known/agent-card.json で公開する
# A2AStarletteApplication を使えば自動的にこのパスで公開される
server = A2AStarletteApplication(
    agent_card=agent_card,
    http_handler=request_handler,
)
# → http://localhost:9999/.well-known/agent-card.json で自動公開

2. TaskStoreの選択

# 本番環境で InMemoryTaskStore を使い続ける
request_handler = DefaultRequestHandler(
    agent_executor=my_executor,
    task_store=InMemoryTaskStore(),  # サーバー再起動でタスク情報が消失！
)

# 本番ではデータベースバックエンドを使用
# a2a-sdk[sql] でPostgreSQL/MySQL/SQLiteに対応
pip install "a2a-sdk[sql]"

# SQLベースのTaskStoreを使用（永続化される）
from a2a.server.tasks import DatabaseTaskStore
task_store = DatabaseTaskStore(database_url="postgresql://...")

3. input-required 状態の処理漏れ

# タスクの結果だけチェックして、input-requiredを無視
response = await client.send_message(request)
if response.result.status.state == "completed":
    print("完了！", response.result.artifacts)
# → input-required の場合、何も表示されずにハングする

# すべてのTaskStateを適切にハンドリング
response = await client.send_message(request)
state = response.result.status.state

if state == "completed":
    print("完了！", response.result.artifacts)
elif state == "input-required":
    # エージェントからの質問に回答して再送信
    print("追加情報が必要:", response.result.status.message)
    # → 必要な情報を含めて再度 send_message する
elif state == "failed":
    print("エラー:", response.result.status.message)
elif state == "rejected":
    print("タスクが拒否されました")

4. MCPとA2Aの混同

# データベースアクセスにA2Aを使おうとする
# → A2Aは「エージェント間通信」用。ツール接続はMCPの役割

# または、MCPでエージェント間の協調を実装しようとする
# → MCPはステートレス。長時間タスクの状態管理には向かない

# レイヤーに応じて使い分ける
# ツール接続（DB、API、ファイル） → MCP
# エージェント間の協調・委任     → A2A

# 典型的な構成:
# [ユーザー] → [Client Agent (MCP経由でDB接続)]
#                    ↓ A2A
#              [Remote Agent (MCP経由で外部API接続)]

参考・出典

1. Google Developers Blog, “Announcing the Agent2Agent Protocol (A2A)“, 2025年4月
2. A2A Protocol, “Agent2Agent (A2A) Protocol Specification“, v0.3
3. GitHub, “a2aproject/a2a-python: Official Python SDK“
4. Google Cloud Blog, “Agent2Agent protocol (A2A) is getting an upgrade“, 2025年7月
5. Linux Foundation, “Linux Foundation Launches the Agent2Agent Protocol Project“, 2025年6月
6. IBM, “What Is Agent2Agent (A2A) Protocol?“
7. Google ADK Docs, “ADK with Agent2Agent (A2A) Protocol“

まとめ: 今日からできる3つのアクション

A2Aプロトコルは、AIエージェントのサイロ化問題を解決するための重要な標準規格です。MCPがエージェントの「手足」を拡張するプロトコルなら、A2Aはエージェント同士の「会話」を可能にするプロトコルです。