エージェント間通信（A2A）

大規模言語モデルが孤立したアシスタントから専門エージェントの協調ネットワークへ進化するにつれ、エージェント同士がどのように会話するかという問いは、内部でどのように推論するかと同じくらい重要になります。この節では、単一のエージェントでは単独で扱えない問題をマルチエージェントシステムが調整し、委譲し、協力して解決できるようにするプロトコル、パターン、エンジニアリング実践を扱います。

動機：エージェントが通信しなければならない理由

Tip

専門化の必然性

単一の汎用エージェントは、知識の広さと能力の深さという根本的な緊張関係に直面します。現実世界のタスク、たとえば法務文書レビュー、多段階の科学研究、エンタープライズソフトウェア開発は、その両方を求めます。エージェント間通信は、専門家のネットワークが協調できるようにすることでこの緊張関係を解消します。各エージェントが強みを提供し、弱みは委譲します。

構造化されたエージェント間通信が必要になる要因はいくつかあります。

認知負荷とコンテキストの限界

すべてのLLMは有限のコンテキストウィンドウ内で動作します。数百の文書、ツール呼び出し、推論ステップにまたがる複雑なワークフローは、単一のエージェントがメモリに保持できる量をすぐに超えます。タスクを複数のエージェントに分解すれば、各エージェントは扱いやすいコンテキスト内で動作し、オーケストレーションを担うエージェントは高レベルの状態だけを維持できます。

専門化と専門知識

異なるエージェントを特定分野向けにファインチューニングしたり、プロンプトで誘導したり、ツールを装備したりできます。たとえば、コンパイラとテストランナーにアクセスできるCodeAgent、判例データベースにアクセスできるLegalAgent、統計ライブラリを持つDataAgentです。サブタスクを適切な専門家へルーティングすることで、品質と効率の両方が向上します。

並列性とスループット

独立したサブタスクは、複数のエージェントへ同時にディスパッチできます。研究オーケストレータは、5つの専門エージェントへ文献検索を並列にファンアウトし、その結果を統合することで、実時間を大幅に短縮できます。

障害分離とレジリエンス

1つのエージェントが失敗しても、適切に設計されたマルチエージェントシステムなら、別のエージェントで再試行したり、より単純な方法へフォールバックしたり、人間へエスカレーションしたりできます。ワークフロー全体を崩壊させる必要はありません。

委譲とハンドオフ

長時間実行されるタスクでは、コンテキストの変化に応じてエージェント間でハンドオフが必要になる場合があります。最初のPlannerAgentが目標を分解し、サブタスクをExecutorAgentsへ渡し、最後にReviewerAgentが出力を検証します。各エージェントは必要なコンテキストだけを正確に受け取ります。

Important

A2A通信の中核要件

1. 発見可能性: エージェントは他のエージェントを見つけ、その能力を理解できなければなりません。

2. 相互運用性: 異なるチームやベンダーが構築したエージェントが共通プロトコルで通信できなければなりません。

3. 非同期性: 長時間タスクが呼び出し元をブロックしてはならず、結果はコールバックまたはポーリングで到着しなければなりません。

4. セキュリティ: エージェント同士が認証し、認可境界を適用しなければなりません。

5. 可観測性: デバッグと監査のため、すべてのメッセージ交換を追跡可能にしなければなりません。

Google A2Aプロトコル

2025年4月、Googleは50社を超えるテクノロジーパートナーの協力を得て、AIエージェント間の相互運用可能な通信のためのオープン仕様である Agent-to-Agent（A2A）Protocol （Google 2025）を公開しました。その後、このプロトコルは Linux Foundation へ寄贈され、2026年時点で支持組織は150を超えています。A2Aは、従来の場当たり的なアプローチと区別される一連の中核原則に基づいて設計されています。

設計思想

A2A仕様は5つの指針原則を示しています（公式仕様（Google 2025）§1.2をもとにしています）。

Important

A2Aの設計原則

不透明な実行

呼び出し元のエージェントはリモートエージェントの内部を決して検査せず、宣言されたインターフェースだけを介して相互作用します。対象がGPT-4、Gemini、ルールベースシステムのいずれであるかはプロトコルにとって重要ではなく、真に異種混在のエージェントエコシステムを可能にします。

エンタープライズ対応

認証（OAuth 2.0、APIキー、JWT）、監査ログ、規制遵守は後付けではなく、最初からプロトコルレベルに統合されています。

モダリティ非依存

1つのメッセージにテキスト、バイナリファイル、構造化JSONペイロードを組み合わせられます。これにより、画像、音声、コード、文書を扱うエージェントをプロトコル拡張なしに受け入れられます。

既存標準による単純さ

新しいトランスポートを発明するのではなく、A2AはJSON-RPC 2.0メッセージ付きのHTTP/HTTPS、ストリーミング用のServer-Sent Events（SSE）、代替バインディングとしてのgRPCを再利用します。いずれも、すべてのインフラチームがすでに運用している技術です。

非同期優先のタスクモデル

長時間実行される操作は例外ではなく、通常です。プッシュ通知とポーリングの両方が第一級の仕組みであるため、呼び出し元が何時間も接続を開いたままにする必要はありません。

エージェントカード

A2Aの発見可能性の基盤は エージェントカード です。これは既知のエンドポイント（/.well-known/agent.json）でホストされる機械可読なJSONマニフェストです。エージェントができること、認証方法、タスクの送信先を宣伝します。RESTエンドポイントではなく自律エージェント向けのOpenAPI仕様に相当します。

Note

エージェントカードの構造

# Agent Card served at https://agent.example.com/.well-known/agent.json
agent_card = {
    "name": "DataAnalysisAgent",
    "description": "Analyzes structured datasets, produces statistical summaries, "
                   "generates visualizations, and answers data questions.",
    "url": "https://agent.example.com/a2a",
    "version": "1.2.0",
    "capabilities": {
        "streaming": True,
        "pushNotifications": True,
        "stateTransitionHistory": True
    },
    "authentication": {
        "schemes": ["Bearer", "ApiKey"]
    },
    "skills": [
        {
            "id": "statistical-analysis",
            "name": "Statistical Analysis",
            "description": "Compute descriptive statistics, run hypothesis tests, "
                           "fit regression models on tabular data.",
            "tags": ["statistics", "data", "analysis", "regression"],
            "examples": [
                "What is the correlation between columns A and B?",
                "Run a t-test comparing these two groups.",
                "Fit a linear regression predicting sales from ad spend."
            ],
            "inputModes": ["text", "data"],
            "outputModes": ["text", "data", "file"]
        },
        {
            "id": "visualization",
            "name": "Data Visualization",
            "description": "Generate charts, plots, and dashboards from data.",
            "tags": ["charts", "plots", "visualization", "dashboard"],
            "examples": [
                "Create a bar chart of monthly revenue.",
                "Plot the distribution of customer ages."
            ],
            "inputModes": ["text", "data"],
            "outputModes": ["file", "text"]
        }
    ],
    "defaultInputModes": ["text"],
    "defaultOutputModes": ["text"]
}

エージェントカードは能力ベースのルーティングを可能にします。オーケストレータエージェントはレジストリからカードを取得し、サブタスクを最も適切なエージェントへ意味的にマッチさせ、ハードコードされたルーティングロジックなしにディスパッチできます。

タスクのライフサイクル

A2Aはすべての作業を タスク としてモデル化します。タスクは明確に定義された状態機械を進みます。

submitted

クライアントがタスクを送り、サーバーが受信を確認した状態です。

working

エージェントが処理中の状態です。クライアントはポーリングするか、SSEイベントを待ちます。

input-required

エージェントが処理を続ける前に、ユーザーまたは呼び出し元エージェントから追加情報を必要とする状態です（例: 明確化の質問、欠けている認証情報）。

completed

タスクが正常に完了し、応答で結果を利用できる状態です。

failed

回復不能なエラーが発生し、エラーメッセージが原因を説明する状態です。

rejected

エージェントがタスクを拒否した状態です（例: 能力の範囲外、または未認可）。A2A v1.0で追加されました。

canceled

クライアントまたはサーバーによってタスクが中断された状態です。

Server-Sent Eventsによるストリーミング

増分出力を生成するタスク（例: 長いレポートの執筆、コードファイルの生成）では、A2Aは Server-Sent Events（SSE） を使います。クライアントは永続HTTP接続を開き、JSONイベントのストリームを受け取ります。

Note

SSEイベントストリームの例

# Each SSE event carries a TaskStatusUpdateEvent or TaskArtifactUpdateEvent
# Example stream for a "write a research report" task:

# Event 1: status update
data: {
  "id": "task-abc123",
  "status": {"state": "working"},
  "final": false
}

# Event 2: partial artifact (streaming text)
data: {
  "id": "task-abc123",
  "artifact": {
    "parts": [{"type": "text", "text": "## Introduction\n\nRecent advances in..."}],
    "index": 0,
    "append": false,
    "lastChunk": false
  },
  "final": false
}

# Event 3: more text appended
data: {
  "id": "task-abc123",
  "artifact": {
    "parts": [{"type": "text", "text": " reinforcement learning have shown..."}],
    "index": 0,
    "append": true,   # append to existing artifact
    "lastChunk": false
  },
  "final": false
}

# Final event: task complete
data: {
  "id": "task-abc123",
  "status": {"state": "completed"},
  "final": true
}

長時間タスク向けのプッシュ通知

タスクに数分から数時間かかる可能性がある場合、SSE接続を開いたまま維持するのは現実的ではありません。A2Aは プッシュ通知 をサポートします。クライアントがWebhook URLを登録し、サーバーがタスクの進行に応じてステータス更新をPOSTします。

# Client registers a push notification endpoint when submitting the task
task_request = {
    "id": "task-xyz789",
    "message": {
        "role": "user",
        "parts": [{"type": "text", "text": "Analyze Q3 sales data and produce a report."}]
    },
    "pushNotification": {
        "url": "https://my-orchestrator.example.com/webhooks/a2a",
        "token": "secret-hmac-token-for-verification",
        "authentication": {
            "schemes": ["Bearer"],
            "credentials": "eyJhbGciOiJIUzI1NiJ9..."
        }
    }
}
# The server will POST TaskStatusUpdateEvent objects to the webhook URL
# as the task transitions through states.

メッセージ形式

A2Aメッセージは、 role （userまたはagent）と、型付き parts （テキスト、ファイル、構造化データ）のリストで構成されます。完全なメッセージスキーマ、マルチモーダルの例、コンテキスト受け渡しの指針は、節9.5で扱います。

認証と認可

A2Aは複数の認証方式をサポートし、エージェントカードで宣言してリクエストごとに適用します。

• Bearerトークン（JWT／OAuth 2.0）: エンタープライズ配置の標準です。トークンがスコープを持ち、呼び出し元エージェントが要求できる内容を制限します。

• APIキー: 内部環境や信頼された環境向けの、より単純な方式です。

• 相互TLS（mTLS）: 高セキュリティ配置向けの証明書ベース認証です。

• OpenID Connect: 連合アイデンティティにより、組織をまたぐエージェント通信を可能にします。

Warning

認可スコープの適用

タスクを受け取るエージェントは、誰が呼び出しているか（認証）だけでなく、何を要求することが許可されているか（認可）も検証しなければなりません。ReportingAgentは、認証済みの任意のエージェントから読み取り専用のデータ検索を受け付けつつ、書き込み操作を特定のOAuthスコープを持つエージェントに制限できます。これを適用しないと、マルチエージェントシステムで権限昇格の脆弱性が生じます。

通信パターン

マルチエージェントシステムは、タスクの性質、レイテンシ要件、関与するエージェント数に応じて、さまざまな通信パターンを使います。

リクエスト・レスポンス

最も単純なパターンです。エージェントAがエージェントBへタスクを送り、完全な応答を待ちます。結果を受け取ってから先へ進む、短く明確に定義されたサブタスクに適しています。

ストリーミング

エージェントAがSSE接続を開き、エージェントBが生成した部分結果をストリーミングします。長文生成（レポート、コード）、リアルタイム協調、段階的なUI更新に適しています。

Note

ストリーミングパターンのユースケース

オーケストレータがWritingAgentに10ページの技術文書の草稿作成を依頼するとします。完全な文書を2分待つ代わりに、オーケストレータは各節が書かれるたびにストリーミングします。これによりReviewAgentは後半の節がまだ生成中でも前半のレビューを始められ、合計レイテンシを40〜60%削減するパイプラインになります。

マルチターンインタラクション

反復的な改善が必要なタスクもあります。エージェントがinput-required状態に入り、オーケストレータが明確化を提供して、タスクが再開します。これは人間の協調ワークフロー、すなわち草稿 \(\to\) フィードバック \(\to\) 改訂を反映します。

# Multi-turn: orchestrator handles input-required state
async def run_multiturn_task(client, initial_message):
    task = await client.send_task(message=initial_message)

    while task.status.state not in ("completed", "failed", "canceled"):
        if task.status.state == "input-required":
            # Agent needs clarification
            clarification_needed = task.status.message
            print(f"Agent asks: {clarification_needed}")

            # Orchestrator generates or forwards a clarifying response
            user_reply = await get_clarification(clarification_needed)

            # Send the reply to continue the task
            task = await client.send_task(
                task_id=task.id,
                message={"role": "user",
                         "parts": [{"type": "text", "text": user_reply}]}
            )
        else:
            # Still working --- poll after a delay
            await asyncio.sleep(2)
            task = await client.get_task(task.id)

    return task

ブロードキャスト

オーケストレータが同じメッセージを複数のエージェントへ同時に送り、告知、共有コンテキストの配布、独立したワークフローの並列起動に使います。

パブリッシュ・サブスクライブ（Pub-Sub）

エージェントがイベントチャネル（例: new-document-uploaded、model-retrained）を購読します。イベントが発生すると、購読しているすべてのエージェントに通知されます。これによりプロデューサーとコンシューマーが分離され、リアクティブでイベント駆動のアーキテクチャが可能になります。

ネゴシエーション

2つのエージェントが提案と反対提案を交換し、計画、リソース配分、アプローチについて合意します。エージェントごとに異なる目的や制約があるマルチエージェント計画システムで一般的です。

Note

ネゴシエーションパターン

PlannerAgentが5段階の研究計画を提案します。ResourceAgentは、ステップ3（大規模シミュレーションの実行）が計算予算を超えると応答します。PlannerAgentは縮小版シミュレーションを反対提案し、ResourceAgentが承認します。合意された計画は実行エージェントへディスパッチされます。

オークションベースのタスク割り当て

オーケストレータが要件付きでタスクを告知し、候補エージェントが入札（推定完了時間、信頼度、コスト）を提出し、オーケストレータが落札エージェントへタスクを割り当てます。エージェントプール全体で、動的かつ市場ベースの負荷分散が可能になります。

A2A通信パターンの概要。

エージェントの発見とルーティング

エージェントが別のエージェントと通信するには、まず相手を見つける必要があります。エージェントの発見とは、特定のタスクを処理できるエージェントを探すプロセスです。

エージェントレジストリ

エージェントレジストリ は、エージェントカードをインデックス化し、検索・参照APIを提供するディレクトリサービスです。配置モデルは2つあります。

集中型レジストリ

単一の権威あるレジストリ（例: エンタープライズサービスカタログ）がすべてのエージェントをインデックス化します。運用は単純ですが、単一障害点を作り、組織をまたぐ配置にはスケールしない可能性があります。

連合型レジストリ

複数のレジストリがそれぞれ分野または組織に対して権威を持ち、レジストリ間検索プロトコルで連携します。よりレジリエントでプライバシーを保護できますが、標準化された連合プロトコルが必要です。

能力ベースのルーティング

エージェントURLをハードコードする代わりに、オーケストレータは 能力ベースのルーティング を行います。必要なスキルに一致するエージェントをレジストリへ問い合わせ、最も適したものを選択します。

class AgentRouter:
    """Routes tasks to agents based on capability matching."""

    def __init__(self, registry_url: str):
        self.registry_url = registry_url
        self._cache: dict[str, list[AgentCard]] = {}

    async def find_agents(self, required_skill: str,
                          tags: list[str] | None = None) -> list[AgentCard]:
        """Query registry for agents with the required skill."""
        params = {"skill": required_skill}
        if tags:
            params["tags"] = ",".join(tags)
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"{self.registry_url}/agents", params=params)
            return [AgentCard(**card) for card in resp.json()["agents"]]

    async def route(self, task_description: str) -> AgentCard:
        """Semantically match a task description to the best available agent."""
        # Embed the task description
        task_embedding = await embed(task_description)

        # Fetch all registered agents
        all_agents = await self.find_agents(required_skill="*")

        # Score each agent by cosine similarity of task to agent description
        scored = []
        for agent in all_agents:
            agent_embedding = await embed(agent.description)
            score = cosine_similarity(task_embedding, agent_embedding)
            scored.append((score, agent))

        # Return the highest-scoring agent
        scored.sort(key=lambda x: x[0], reverse=True)
        return scored[0][1]

同等エージェント間の負荷分散

複数のエージェントが同じ能力を提供する場合、ルーターは負荷を分散しなければなりません。一般的な戦略は次のとおりです。

• ラウンドロビン: 利用可能なすべてのエージェントにタスクを均等に分配します。

• 最小負荷: アクティブなタスクが最も少ないエージェントへルーティングします（ヘルス／メトリクスエンドポイントが必要）。

• レイテンシ考慮: 直近の応答時間が最も短いエージェントへルーティングします。

• アフィニティベース: 関連タスクを同じエージェントへルーティングし、キャッシュされたコンテキストを活用します。

バージョン管理と互換性

エージェントカードにはversionフィールドがあります。オーケストレータは最低バージョン要件を指定し、古いバージョンしか利用できない場合には適切に縮退すべきです。セマンティックバージョニング（Preston-Werner 2024）（MAJOR.MINOR.PATCH）が推奨されます。互換性を壊すインターフェース変更ではMAJORを、新しい能力の追加ではMINORを増やします。

Warning

長時間稼働システムにおけるバージョンのずれ

本番のマルチエージェントシステムでは、異なるエージェントが異なる時期に更新され、バージョンのずれが生じることがあります。Agent Card v2.1向けにコンパイルされたオーケストレータが、v1.3でまだ動作しているエージェントに遭遇する可能性があります。常に後方互換のメッセージ処理を実装し、バージョンをまたぐシナリオを明示的にテストしてください。

メッセージ形式とスキーマ

構造化メッセージ対非構造化メッセージ

A2Aは、完全に非構造化されたもの（プレーンテキスト）から完全に構造化されたもの（型付きJSONスキーマ）までの幅をサポートします。適切な選択は関与するエージェントによって異なります。

構造化A2Aメッセージと非構造化A2Aメッセージのトレードオフ。

マルチモーダルメッセージ

A2Aメッセージは、 role （userまたはagent）と型付き parts のリストとして構造化されます。

A2Aメッセージのパート種別（ワイヤーフォーマットでは"type": "text"|"file"|"data"を使います）。

現代のエージェントは、テキスト以外のモダリティを扱うことが増えています。A2AのFilePartは任意のMIMEタイプをサポートし、豊かなマルチモーダルワークフローを可能にします。

Note

マルチモーダルA2Aメッセージ：データ分析

# A message combining text instructions with a data payload and a file
message = {
    "role": "user",
    "parts": [
        {
            "type": "text",
            "text": "Analyze the attached CSV and the schema below. "
                    "Identify anomalies and produce a summary report."
        },
        {
            "type": "file",
            "mimeType": "text/csv",
            "uri": "https://storage.example.com/data/sales_q3.csv"
        },
        {
            "type": "data",
            "data": {
                "schema": {
                    "columns": ["date", "region", "product", "revenue", "units"],
                    "types":   ["date", "string", "string", "float", "int"]
                },
                "expectedRowCount": 15000,
                "anomalyThreshold": 3.0  # z-score threshold
            }
        }
    ]
}

Note

マルチモーダルA2Aメッセージ：画像分析

# Multi-modal message: text + image + structured data
multimodal_message = {
    "role": "user",
    "parts": [
        {"type": "text",
         "text": "Describe what is wrong with this chart and suggest fixes."},
        {"type": "file",
         "mimeType": "image/png",
         "bytes": base64.b64encode(chart_image_bytes).decode()},
        {"type": "data",
         "data": {
             "chartType": "bar",
             "dataSource": "Q3 Revenue by Region",
             "knownIssues": ["y-axis does not start at zero",
                             "missing error bars"]
         }}
    ]
}

コンテキストの受け渡し：共有するものと非公開にするもの

マルチエージェントシステムにおける重要な設計判断はコンテキストスコープです。会話履歴と内部状態をどの程度サブエージェントへ渡すかを決めます。

Important

コンテキストスコープの原則

最小限のコンテキスト

サブエージェントがタスクを完了するために必要なものだけを渡します。トークン使用量、レイテンシ、機密情報を漏らすリスクを減らします。

要約されたコンテキスト

生の会話履歴を渡す代わりに、目標、制約、決定事項、関連事実からなる構造化要約を渡します。

非公開状態

内部推論、中間草稿、ユーザーの個人識別情報は、明示的に必要でない限り、通常サブエージェントへ転送すべきではありません。

相関ID

常にcorrelationIdを渡し、サブエージェントのアクションをログと監査証跡で元のワークフローまで追跡できるようにします。

会話スレッドと相関ID

複雑なワークフローでは、多数のタスクが同時進行することがあります。 相関ID がエージェント間の関連タスクをつなぎます。

import uuid

class WorkflowContext:
    """Carries correlation metadata through a multi-agent workflow."""

    def __init__(self, workflow_id: str | None = None):
        self.workflow_id = workflow_id or str(uuid.uuid4())
        self.span_id = str(uuid.uuid4())
        self.parent_span_id: str | None = None

    def child_context(self) -> "WorkflowContext":
        """Create a child context for a sub-task."""
        child = WorkflowContext(workflow_id=self.workflow_id)
        child.parent_span_id = self.span_id
        return child

    def to_metadata(self) -> dict:
        return {
            "x-workflow-id": self.workflow_id,
            "x-span-id": self.span_id,
            "x-parent-span-id": self.parent_span_id
        }

# Usage: attach to every A2A task submission
ctx = WorkflowContext()
task = await client.send_task(
    message=message,
    metadata=ctx.to_metadata()
)
# Sub-tasks use child contexts for tracing
sub_ctx = ctx.child_context()

調整プロトコル

ポイントツーポイント通信に加え、マルチエージェントシステムは、集団的な意思決定と問題解決を可能にする構造化された相互作用パターンである、より高レベルの 調整プロトコル から恩恵を受けます。

Contract Netプロトコル

Contract Net Protocol（CNP） （Smith 1980）は、LLMベースのシステム向けに適応された古典的なマルチエージェント調整メカニズムです。

1. 告知: マネージャーエージェントが、タスク要件と評価基準を含むタスク告知を、すべての候補請負エージェントへブロードキャストします。

2. 入札: 請負エージェントが自分の能力に照らしてタスクを評価し、推定完了時間、信頼度、リソース要件を含む入札を提出します。

3. 落札: マネージャーが落札入札（または並列サブタスク向けの複数入札）を選び、契約を割り当てます。

4. 実行と報告: 請負エージェントがタスクを実行し、結果をマネージャーへ報告します。

Note

Contract Netプロトコルの実装

import dataclasses

class ContractNetManager:
    """Implements the Contract Net Protocol for task allocation."""

    async def allocate_task(self, task: Task,
                            candidate_agents: list[AgentCard]) -> AgentCard:
        # Phase 1: Announce task to all candidates
        announcement = {
            "type": "task-announcement",
            "task": dataclasses.asdict(task),
            "deadline": (datetime.now(timezone.utc) + timedelta(seconds=10)).isoformat(),
            "evaluationCriteria": ["confidence", "estimatedTime", "cost"]
        }

        # Phase 2: Collect bids
        bids = await asyncio.gather(*[
            self._request_bid(agent, announcement)
            for agent in candidate_agents
        ], return_exceptions=True)

        valid_bids = [(agent, bid) for agent, bid in zip(candidate_agents, bids)
                      if not isinstance(bid, Exception) and bid is not None]

        if not valid_bids:
            raise RuntimeError(f"No agents bid on task {task.id}")

        # Phase 3: Award to best bidder (highest confidence, lowest time)
        def score_bid(agent_bid):
            _, bid = agent_bid
            return bid["confidence"] - 0.1 * bid["estimatedSeconds"]

        winner_agent, winning_bid = max(valid_bids, key=score_bid)

        # Notify winner and losers
        await self._award_contract(winner_agent, task)
        await asyncio.gather(*[
            self._reject_bid(agent, task.id)
            for agent, _ in valid_bids if agent != winner_agent
        ])

        return winner_agent

    async def _request_bid(self, agent: AgentCard,
                           announcement: dict) -> dict | None:
        """Ask an agent to bid on a task."""
        try:
            result = await self.client.send_task(
                agent_url=agent.url,
                message={"role": "user",
                         "parts": [{"type": "data", "data": announcement}]}
            )
            return result.artifacts[0].parts[0]["data"]
        except Exception:
            return None

ブラックボードシステム

ブラックボードシステム （Hayes-Roth 1985）は、エージェントが部分解、観測、仮説を投稿する共有ワークスペース（「ブラックボード」）を提供します。他のエージェントはブラックボードを監視し、価値を加えられるときに貢献します。機会主義的な問題解決アプローチです。

ブラックボードシステムは、解決経路が事前に分からず、異なるエージェントが異なる段階で貢献できる問題に適しています。科学的仮説の生成、複雑なデバッグ、複数ソースのインテリジェンス分析などが例です。

合意形成プロトコル

複数のエージェントが意思決定（例: どの計画を実行するか、結果が正しいか）について合意しなければならない場合、 合意形成プロトコル が構造化された投票メカニズムを提供します。

単純多数決

各エージェントが投票し、\(> 50%\) の票を得た選択肢が勝ちます。高速ですが、エージェントが同じベースモデルを共有すると相関した誤りに弱くなります。

重み付き投票

投票にエージェントの信頼度や過去の正確性に基づく重みを付けます。より堅牢ですが、較正された信頼度推定が必要です。

定足数ベース

意思決定には、\(k\) エージェント、すなわち \(n\) エージェントのうち少なくともその数の合意が必要です。最大 \(n-k\) エージェントが失敗または不同意でもブロックしない、フォールトトレランスを提供します。

デルファイ法

エージェントが投票し、匿名化された結果を見て投票を修正し、収束するまで繰り返します。アンカリングバイアスを減らし、本物の熟議を促します。

async def quorum_vote(agents: list[AgentCard], question: str,
                      options: list[str], quorum: int) -> str | None:
    """Run a quorum vote across agents. Returns winning option or None."""
    votes = await asyncio.gather(*[
        ask_agent_to_vote(agent, question, options)
        for agent in agents
    ])

    counts: dict[str, int] = {}
    for vote in votes:
        if vote in options:
            counts[vote] = counts.get(vote, 0) + 1

    # Return first option that reaches quorum
    for option, count in sorted(counts.items(), key=lambda x: -x[1]):
        if count >= quorum:
            return option
    return None  # No quorum reached

リーダー選出

動的なマルチエージェントシステムでは、実行時に リーダー （オーケストレータ）を選出する必要がある場合があります。たとえば、元のオーケストレータが失敗した場合や、事前にコーディネータを割り当てずにエージェントが自己組織化する場合です。古典的な分散システムアルゴリズム（Bully、Ring）をエージェントネットワーク向けに適応し、エージェントが能力スコアや優先度トークンを交換して、利用可能な最も能力の高いエージェントをリーダーに選出できます。

A2AとMCP：補完的なプロトコル

A2Aと Model Context Protocol（MCP） （Anthropic 2024b）の関係は、よく混乱を招きます。これらのプロトコルは競合するのではなく、補完的です。

Important

中核的な違い

• MCP は垂直プロトコルです。エージェントをデータベース、API、ファイルシステム、コード実行器の世界へ下向きに拡張します。推論するのはエージェントだけで、MCPエンドポイントは決定論的なサービスです。

• A2A は水平プロトコルです。1つの推論エージェントを別の推論エージェントへ接続します。両側が推論、計画、ツール利用のできる知的主体です。

どちらを使うか

• リモートエンドポイントが決定論的な関数の場合は MCP を使います。データベース検索、API呼び出し、コード実行サンドボックスなどです。エージェントが相互作用を完全に制御します。

• リモートエンドポイントが要求について推論する必要がある場合は A2A を使います。曖昧な指示の解釈、判断、独自ツールの利用、マルチターン対話などです。

• 同じシステムで 両方 を使います。オーケストレータエージェントがA2Aで専門エージェントへ委譲し、各専門エージェントがMCPでツールへアクセスします。

統合アーキテクチャ

本番のマルチエージェントシステムでは、A2AとMCPが異なる層で協調します。 A2A はエージェント間の委譲と調整（ピア間の水平通信）を扱い、 MCP は各エージェントとツール・データソースの接続（能力との垂直統合）を扱います。この関心の分離が、スケール可能なエージェントアーキテクチャを構築する鍵です。

• 委譲のためのA2A: エージェントが持たない能力を必要とするとき、A2Aタスクメッセージを介して別のエージェントへ委譲します。各エージェントは独自のエージェントカードを持つ自己完結型サービスです。

• ツールアクセスのためのMCP: 各エージェントはMCPサーバーを通じてツールへ接続します。つまり、ツールが他のエージェントへ直接公開されることはなく、所有エージェントのインターフェースを通じてのみ公開されます。

• 信頼境界の分離: オーケストレータは専門エージェントを信頼します（A2A認証で検証）。各専門エージェントは自分のMCPサーバー（ローカルまたは認証済み）を信頼します。推移的なツールアクセスはありません。

• 独立したスケーリング: コード中心のワークロードはCodeAgentインスタンスを、データワークロードはDataAgentをスケールできます。オーケストレータは軽量なままです。

マルチエージェントシステムのセキュリティと信頼

マルチエージェントシステムには固有のセキュリティ課題があります。エージェントAがエージェントBへ委譲し、BがエージェントCへ委譲する場合、信頼の連鎖を慎重に管理しなければなりません。

エージェントの身元検証

各エージェントは検証可能な身元を持たなければなりません。選択肢には次があります。

• 信頼されたIDプロバイダーが署名した JWTトークン （Jones et al. 2015）。エージェントID、発行者、有効期限を持ち、受信エージェントがプロバイダーの公開鍵で検証します。

• 内部CAが発行する mTLS証明書 （Campbell et al. 2020）。認証とトランスポート暗号化の両方を提供します。

• 単一の信頼された権威が存在しない組織横断シナリオ向けの 分散型識別子（DID） （Consortium 2022）。

メッセージ完全性と暗号化

• 盗聴や中間者攻撃を防ぐため、すべてのA2A通信を TLS 1.3 （Rescorla 2018）上で行うべきです。

• 機密ペイロードには エンドツーエンド暗号化 （例: JWE）を使い、中間インフラ（ロードバランサー、プロキシ）がメッセージ内容を読めないようにします。

• メッセージ署名 （JWS）は否認防止を提供します。受信エージェントは、特定のメッセージが特定の送信者から来たことを証明できます。

認可スコープ

すべてのエージェントが、他のすべてのエージェントへ何でも依頼できるべきではありません。OAuth 2.0認可スコープ（Hardt 2012）が境界を定義します。

# Example OAuth 2.0 scopes for a DataAgent
SCOPES = {
    "data:read":        "Read data from connected databases",
    "data:write":       "Write or modify data in connected databases",
    "data:export":      "Export data to external systems",
    "analysis:run":     "Execute statistical analyses",
    "analysis:schedule":"Schedule recurring analyses",
    "admin:config":     "Modify agent configuration"
}

# A ReportingAgent might hold only: data:read, analysis:run
# An ETL pipeline agent might hold: data:read, data:write, data:export
# Only a human admin holds: admin:config

class A2AServer:
    def verify_authorization(self, token: str, required_scope: str) -> bool:
        """Verify that the calling agent holds the required scope."""
        claims = jwt.decode(token, self.public_key, algorithms=["RS256"])
        granted_scopes = claims.get("scope", "").split()
        if required_scope not in granted_scopes:
            raise PermissionError(
                f"Caller lacks required scope '{required_scope}'. "
                f"Granted: {granted_scopes}"
            )
        return True

監査証跡と説明責任

Warning

説明責任の空白

エージェント委譲の連鎖では、アクションの責任者が誰か不明確になることがあります。エージェントAがエージェントBにファイル削除を依頼し、Bが実行した場合、誰が説明責任を負うのでしょうか。すべてのA2A相互作用に、呼び出し元エージェントの身元、タスク説明、使用した認可トークン、タイムスタンプ、結果を記録しなければなりません。この監査証跡は、インシデント対応、コンプライアンス、デバッグに不可欠です。

すべてのA2Aサーバーは構造化監査ログを出力すべきです。

@dataclass
class A2AAuditEvent:
    timestamp: str          # ISO 8601
    workflow_id: str        # Correlation ID for the top-level workflow
    span_id: str            # This task's span
    parent_span_id: str     # Calling task's span (for delegation chains)
    caller_agent_id: str    # Verified identity of the calling agent
    callee_agent_id: str    # This agent's identity
    task_id: str
    skill_invoked: str
    authorization_scopes: list[str]
    outcome: str            # "completed" | "failed" | "rejected"
    duration_ms: int
    error_code: str | None

実装例：マルチエージェント研究ワークフロー

次の例は、A2Aを使った完全なマルチエージェント研究ワークフローを示します。OrchestratorAgentが研究課題を分解し、専門エージェントへ委譲し、その結果を統合します。

"""
Multi-agent research workflow using A2A protocol.
Demonstrates: Agent Cards, A2A client/server, task lifecycle,
multi-turn interaction, and agent handoffs.
"""

import asyncio
import json
import uuid
from collections.abc import AsyncIterator
from datetime import datetime, timedelta, timezone

import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

# -- Data Models --------------------------------------------------------------

class Part(BaseModel):
    type: str           # "text" | "file" | "data"
    text: str | None = None
    data: dict | None = None
    mimeType: str | None = None
    uri: str | None = None

class Message(BaseModel):
    role: str           # "user" | "agent"
    parts: list[Part]

class TaskStatus(BaseModel):
    state: str          # submitted | working | input-required | completed | failed
    message: str | None = None
    timestamp: str = Field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )

class Artifact(BaseModel):
    parts: list[Part]
    index: int = 0
    append: bool = False
    lastChunk: bool = True

class Task(BaseModel):
    id: str
    status: TaskStatus
    messages: list[Message] = []
    artifacts: list[Artifact] = []
    metadata: dict = {}

# -- A2A Client (HTTP/REST binding) --------------------------------------------
# Note: A2A v1.0 defines three protocol bindings: JSON-RPC 2.0, gRPC, and
# HTTP+JSON/REST. This example uses the REST binding for readability.

class A2AClient:
    """Client for sending tasks to A2A-compliant agents."""

    def __init__(self, agent_url: str, auth_token: str):
        self.agent_url = agent_url.rstrip("/")
        self.headers = {
            "Authorization": f"Bearer {auth_token}",
            "Content-Type": "application/json"
        }

    async def get_agent_card(self) -> dict:
        """Fetch the agent's capability card."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.agent_url}/.well-known/agent.json",
                headers=self.headers
            )
            resp.raise_for_status()
            return resp.json()

    async def send_task(self, message: Message,
                        task_id: str | None = None,
                        metadata: dict | None = None) -> Task:
        """Submit a task and return the initial task object."""
        payload = {
            "id": task_id or str(uuid.uuid4()),
            "message": message.model_dump(),
            "metadata": metadata or {}
        }
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.agent_url}/tasks/send",
                json=payload,
                headers=self.headers,
                timeout=30.0
            )
            resp.raise_for_status()
            return Task(**resp.json())

    async def stream_task(self, message: Message,
                          metadata: dict | None = None) -> AsyncIterator[dict]:
        """Submit a task and stream SSE events."""
        payload = {
            "id": str(uuid.uuid4()),
            "message": message.model_dump(),
            "metadata": metadata or {}
        }
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                f"{self.agent_url}/tasks/sendSubscribe",
                json=payload,
                headers={**self.headers, "Accept": "text/event-stream"},
                timeout=300.0
            ) as response:
                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        event_data = json.loads(line[6:])
                        yield event_data
                        if event_data.get("final"):
                            break

    async def get_task(self, task_id: str) -> Task:
        """Poll for task status."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.agent_url}/tasks/{task_id}",
                headers=self.headers
            )
            resp.raise_for_status()
            return Task(**resp.json())

    async def wait_for_completion(self, task: Task,
                                  poll_interval: float = 2.0) -> Task:
        """Poll until task reaches a terminal state."""
        terminal_states = {"completed", "failed", "canceled"}
        while task.status.state not in terminal_states:
            await asyncio.sleep(poll_interval)
            task = await self.get_task(task.id)
        return task

# -- A2A Server (FastAPI) -----------------------------------------------------

class ResearchAgent:
    """
    A specialist research agent that searches literature and
    summarizes findings on a given topic.
    """

    AGENT_CARD = {
        "name": "ResearchAgent",
        "description": "Searches academic literature and synthesizes research findings.",
        "url": "https://research-agent.example.com/a2a",
        "version": "1.0.0",
        "capabilities": {
            "streaming": True,
            "pushNotifications": False,
            "stateTransitionHistory": True
        },
        "authentication": {"schemes": ["Bearer"]},
        "skills": [{
            "id": "literature-search",
            "name": "Literature Search",
            "description": "Search and summarize academic papers on a topic.",
            "tags": ["research", "literature", "academic", "papers"],
            "examples": [
                "Summarize recent papers on transformer attention mechanisms.",
                "What does the literature say about RLHF for code generation?"
            ],
            "inputModes": ["text"],
            "outputModes": ["text", "data"]
        }]
    }

    def __init__(self):
        self.tasks: dict[str, Task] = {}
        self.app = FastAPI(title="ResearchAgent A2A Server")
        self._register_routes()

    def _register_routes(self):
        @self.app.get("/.well-known/agent.json")
        async def agent_card():
            return self.AGENT_CARD

        @self.app.post("/tasks/send")
        async def send_task(request: Request):
            body = await request.json()
            task = await self._create_and_run_task(body)
            return task.model_dump()

        @self.app.post("/tasks/sendSubscribe")
        async def send_subscribe(request: Request):
            body = await request.json()
            return StreamingResponse(
                self._stream_task(body),
                media_type="text/event-stream"
            )

        @self.app.get("/tasks/{task_id}")
        async def get_task(task_id: str):
            if task_id not in self.tasks:
                raise HTTPException(status_code=404, detail="Task not found")
            return self.tasks[task_id].model_dump()

    async def _create_and_run_task(self, body: dict) -> Task:
        task_id = body.get("id", str(uuid.uuid4()))
        message = Message(**body["message"])

        task = Task(
            id=task_id,
            status=TaskStatus(state="submitted"),
            messages=[message],
            metadata=body.get("metadata", {})
        )
        self.tasks[task_id] = task

        # Run asynchronously
        asyncio.create_task(self._execute_task(task_id))
        return task

    async def _execute_task(self, task_id: str):
        task = self.tasks[task_id]
        task.status = TaskStatus(state="working")

        try:
            # Extract the research question from the message
            question = task.messages[0].parts[0].text

            # Simulate literature search (replace with real search tool)
            await asyncio.sleep(1)  # Simulated latency
            findings = await self._search_literature(question)

            # Produce artifact
            task.artifacts = [Artifact(parts=[
                Part(type="text", text=findings["summary"]),
                Part(type="data", data={"papers": findings["papers"],
                                        "query": question})
            ])]
            task.status = TaskStatus(state="completed")

        except Exception as e:
            task.status = TaskStatus(state="failed", message=str(e))

        self.tasks[task_id] = task

    async def _search_literature(self, question: str) -> dict:
        """Placeholder: in production, calls a real search API."""
        return {
            "summary": f"Based on a search of recent literature regarding "
                       f"'{question}', key findings include: ...",
            "papers": [
                {"title": "Attention Is All You Need", "year": 2017,
                 "relevance": 0.95},
                {"title": "RLHF: Training Language Models to Follow Instructions",
                 "year": 2022, "relevance": 0.88}
            ]
        }

    async def _stream_task(self, body: dict) -> AsyncIterator[str]:
        task = await self._create_and_run_task(body)

        # Stream status updates
        yield f"data: {json.dumps({'id': task.id, 'status': {'state': 'submitted'}, 'final': False})}\n\n"
        yield f"data: {json.dumps({'id': task.id, 'status': {'state': 'working'}, 'final': False})}\n\n"

        # Wait for completion
        while task.status.state not in ("completed", "failed", "canceled"):
            await asyncio.sleep(0.5)
            task = self.tasks[task.id]

        # Stream the artifact
        if task.artifacts:
            for part in task.artifacts[0].parts:
                event = {
                    "id": task.id,
                    "artifact": {
                        "parts": [part.model_dump()],
                        "index": 0,
                        "append": False,
                        "lastChunk": True
                    },
                    "final": False
                }
                yield f"data: {json.dumps(event)}\n\n"

        # Final status
        yield f"data: {json.dumps({'id': task.id, 'status': task.status.model_dump(), 'final': True})}\n\n"

# -- Orchestrator: Multi-Agent Workflow ----------------------------------------

class ResearchOrchestrator:
    """
    Orchestrates a multi-agent research workflow:
    1. Decomposes the research question into sub-questions
    2. Dispatches each sub-question to a ResearchAgent
    3. Synthesizes results into a final report
    """

    def __init__(self, research_agent_url: str, auth_token: str):
        self.research_client = A2AClient(research_agent_url, auth_token)
        self.workflow_id = str(uuid.uuid4())

    async def run(self, research_question: str) -> str:
        print(f"[Orchestrator] Starting workflow {self.workflow_id}")
        print(f"[Orchestrator] Question: {research_question}")

        # Step 1: Decompose into sub-questions
        sub_questions = self._decompose(research_question)
        print(f"[Orchestrator] Decomposed into {len(sub_questions)} sub-questions")

        # Step 2: Dispatch sub-questions in parallel
        tasks = await asyncio.gather(*[
            self.research_client.send_task(
                message=Message(role="user", parts=[Part(type="text", text=q)]),
                metadata={"workflowId": self.workflow_id, "subQuestion": i}
            )
            for i, q in enumerate(sub_questions)
        ])

        # Step 3: Wait for all tasks to complete
        completed_tasks = await asyncio.gather(*[
            self.research_client.wait_for_completion(task)
            for task in tasks
        ])

        # Step 4: Check for failures
        failed = [t for t in completed_tasks if t.status.state == "failed"]
        if failed:
            print(f"[Orchestrator] Warning: {len(failed)} sub-tasks failed")

        # Step 5: Synthesize results
        findings = []
        for task, question in zip(completed_tasks, sub_questions):
            if task.status.state == "completed" and task.artifacts:
                summary = task.artifacts[0].parts[0].text
                findings.append(f"### {question}\n{summary}")

        report = self._synthesize(research_question, findings)
        print(f"[Orchestrator] Workflow complete. Report: {len(report)} chars")
        return report

    def _decompose(self, question: str) -> list[str]:
        """Decompose a complex question into focused sub-questions."""
        # In production: use an LLM to decompose
        return [
            f"What are the foundational methods for: {question}?",
            f"What are the most recent advances in: {question}?",
            f"What are the open challenges and limitations in: {question}?"
        ]

    def _synthesize(self, question: str, findings: list[str]) -> str:
        """Synthesize sub-findings into a coherent report."""
        # In production: use an LLM to synthesize
        sections = "\n\n".join(findings)
        return f"# Research Report: {question}\n\n{sections}"

# -- Entry Point ---------------------------------------------------------------

async def main():
    orchestrator = ResearchOrchestrator(
        research_agent_url="https://research-agent.example.com/a2a",
        auth_token="eyJhbGciOiJSUzI1NiJ9..."
    )
    report = await orchestrator.run(
        "Reinforcement learning from human feedback for large language models"
    )
    print(report)

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

まとめ

Important

要点：エージェント間通信

1. A2Aは大規模な専門化を可能にします。 タスクを専門エージェントへルーティングすることで、マルチエージェントシステムは深さと広さを同時に実現します（章10ではマルチエージェントアーキテクチャを詳しく扱います）。

2. GoogleのA2A Protocol は、エージェントカード、タスクライフサイクル管理、SSEストリーミング、エンタープライズ認証を備えた、本番対応のオープンな相互運用標準を提供します。

3. 通信パターン は、単純なリクエスト・レスポンスから複雑なネゴシエーションやオークションベースの割り当てまで広がります。タスクの複雑さとレイテンシの要件に基づいて選択します。

4. A2AとMCPは補完的です。 A2Aはエージェント同士を接続し、MCPはエージェントをツールへ接続します。本番システムの多くは両方を使います。

5. セキュリティは譲れません。 エージェントの身元検証、認可スコープ、監査証跡は、あらゆるマルチエージェント配置に不可欠です。

6. 調整プロトコル （Contract Net、ブラックボード、合意形成）は、単純な委譲を超えた集団意思決定のための構造化メカニズムを提供します。

7. 相関IDによる可観測性 は、多数のエージェントとツールにまたがる複雑なマルチエージェントワークフローのデバッグと監査に不可欠です。

Note

A2Aにおける未解決の研究課題

• 階層内の複数のオーケストレータから受け取る矛盾した指示を、エージェントはどのように処理すべきでしょうか。どのような競合解決メカニズムが最も効果的でしょうか。

• 静的な能力宣言に頼るのではなく、エージェントは経験を通じてよりよいルーティングや委譲戦略を学習できるでしょうか。

• 悪意のあるエージェントがメッセージに敵対的な指示を埋め込み、下流エージェントを操作するプロンプトインジェクション攻撃を、どのように防げばよいでしょうか。

• コンテキスト受け渡しにおける適切なプライバシー境界はどこでしょうか。サブエージェントにどの程度の会話履歴を見せるべきで、技術的に境界をどう適用すべきでしょうか。

• エージェントネットワークが数百、数千のエージェントへ拡大したとき、ボトルネックや一貫性違反を生じさせずに整合したグローバル状態をどのように維持すればよいでしょうか。