A2A

Mark as read

A2A (Agent2Agent) 技術概要 — エージェント間通信プロトコルの全容

はじめに — A2Aとは何か、なぜ重要なのか
プロトコルアーキテクチャと設計原則
コアコンセプト: Agent Card, Task, Message, Artifact, Streaming
通信パターンとデータフロー
エージェント探索とケイパビリティネゴシエーション
タスクライフサイクル管理
セキュリティモデルと認証
既存エージェントフレームワークとの統合
MCP (Model Context Protocol) との比較
実装例と詳細コード
実世界のユースケース
将来のロードマップとエコシステム

1. はじめに — A2Aとは何か、なぜ重要なのか

1.1 AI エージェント時代の到来

2024年から2025年にかけて、AI技術は単なるチャットボットや質問応答システムから、自律的にタスクを実行する「AIエージェント」へと急速に進化してきた。AIエージェントとは、与えられた目標に対して自ら計画を立て、ツールを使い、判断を下し、結果を返すことができる自律的なAIシステムである。

LangChain、CrewAI、AutoGen、Google ADK（Agent Development Kit）といったフレームワークが登場し、単独のエージェントを構築することは比較的容易になった。しかし、現実のビジネスプロセスは一つのエージェントだけで完結することは稀である。カスタマーサポート、サプライチェーン管理、金融取引処理など、多くの業務は複数の専門領域にまたがり、異なるシステム、異なるベンダーのAIエージェントが協調して動作する必要がある。

ここで根本的な課題が浮上する。異なるフレームワーク、異なるベンダーで構築されたAIエージェント同士がどのように通信し、協調作業を行うのか？ これが A2A（Agent2Agent）プロトコルが解決しようとする核心的な問題である。

1.2 A2A プロトコルの概要

A2A（Agent2Agent Protocol）は、独立したAIエージェント間のシームレスな通信とコラボレーションを可能にするために設計されたオープンスタンダードである。2025年4月にGoogleによって最初に発表され、その後 Linux Foundation に寄贈された。

A2Aの基本的な考え方は明快である。エージェントは他のエージェントの内部実装（メモリ、ツール、プロプライエタリなロジック）にアクセスする必要なく、標準化されたプロトコルを介して相互に通信できるべきだという原則に基づいている。これは、Webブラウザが異なるWebサーバーとHTTPプロトコルを通じてシームレスに通信できるのと同じ考え方である。

1.3 なぜ A2A が必要なのか

A2Aが必要とされる理由を、具体的なシナリオで説明しよう。

シナリオ: エンタープライズ採用プロセス

大企業の人事部門を考えてみよう。採用プロセスには以下の専門エージェントが関与する可能性がある。

候補者スクリーニングエージェント: 応募書類を分析し、要件に合致する候補者をフィルタリングする
面接スケジューリングエージェント: 候補者と面接官のカレンダーを調整し、最適な面接日程を設定する
バックグラウンドチェックエージェント: 候補者の経歴確認を実施する
オファーレタージェネレーターエージェント: 給与計算、福利厚生パッケージを含むオファーレターを作成する
オンボーディングエージェント: 入社手続きを自動化する

これらのエージェントは異なるベンダーによって開発され、異なるフレームワーク上で動作し、異なるクラウドプラットフォームにデプロイされている可能性がある。A2Aがなければ、各エージェント間のカスタム統合を個別に構築する必要があり、エージェントの数が増えるにつれて統合の複雑さは指数関数的に増大する。

A2Aプロトコルは、この問題を以下の方法で解決する。

標準化されたエージェント探索: Agent Card を通じて、各エージェントが自身の能力を宣言し、他のエージェントがそれを発見できる
統一されたタスク管理: 標準化されたタスクライフサイクルにより、エージェント間でのタスク委任と追跡が可能になる
フレームワーク非依存: 特定のAIフレームワークやベンダーに依存しない通信が可能になる
セキュアな通信: 標準化された認証・認可メカニズムにより、安全なエージェント間通信が保証される

1.4 A2A の歴史と現在

A2Aプロトコルの主要なマイルストーンは以下の通りである。

時期	イベント
2025年4月	Google が A2A プロトコルを初めて発表
2025年中頃	初期仕様の公開とコミュニティフィードバック
2025年後半	Linux Foundation への寄贈
2025年-2026年	SDK の拡充（Python, JavaScript, Java, C#/.NET, Go）
2026年	エコシステムの拡大と企業採用の進展

現在、A2Aは Python、JavaScript、Java、C#/.NET、Golang の公式SDKを提供しており、LangGraph、CrewAI、Semantic Kernel、Google ADK などの主要エージェントフレームワークとの統合がサポートされている。

1.5 本記事の構成

本記事では、A2Aプロトコルのすべての側面を詳細に解説する。プロトコルアーキテクチャの設計思想から始まり、コアコンセプトの深い理解、具体的な実装パターン、セキュリティモデル、他のプロトコルとの比較、そして実践的なコード例までを網羅する。A2Aを実際のプロジェクトに導入しようとするエンジニアにとって、包括的なリファレンスとなることを目指している。

2. プロトコルアーキテクチャと設計原則

2.1 設計思想

A2Aプロトコルの設計は、以下の5つの基本原則に基づいている。

2.1.1 不透明性（Opaqueness）

A2Aにおいて最も重要な設計原則の一つは、エージェント間の「不透明性」である。これは、あるエージェントが別のエージェントと通信する際に、相手の内部構造（使用しているLLMモデル、プロンプト、ツール、メモリ構造など）を知る必要がないことを意味する。

この原則により、各エージェントは完全にカプセル化された状態で動作できる。エージェントの開発者は、内部ロジックを公開することなく、標準化されたインターフェースのみを通じてサービスを提供できる。これは、マイクロサービスアーキテクチャにおけるAPIファーストの考え方と類似している。

エージェントA                    エージェントB
+-----------+                   +-----------+
| 内部ロジック |                   | 内部ロジック |
| LLM       | <--- A2A ----->  | LLM       |
| ツール     |    プロトコル      | ツール     |
| メモリ     |                   | メモリ     |
+-----------+                   +-----------+
  公開されない                     公開されない

2.1.2 相互運用性（Interoperability）

A2Aは特定のフレームワーク、モデル、ベンダーに依存しない設計となっている。LangChain で構築されたエージェントが CrewAI で構築されたエージェントと通信でき、Google の Gemini を使うエージェントが OpenAI の GPT を使うエージェントと協調作業できる。

この相互運用性は、Web標準（HTTP、JSON-RPC、gRPC）に基づくプロトコルバインディングによって実現されている。

2.1.3 ケイパビリティ駆動（Capability-Driven）

エージェントは Agent Card を通じて自身の能力を宣言し、クライアントはその能力に基づいて適切なエージェントを選択し、適切な方法で通信できる。ストリーミング対応、プッシュ通知対応、対応するコンテンツタイプなど、すべての能力が明示的に宣言される。

2.1.4 タスク中心（Task-Centric）

A2Aのすべての通信はタスクを中心に組織されている。メッセージの交換、成果物の生成、状態の管理はすべてタスクというコンテキストの中で行われる。これにより、非同期的な処理、長時間実行されるタスク、マルチターンの対話が自然にモデル化できる。

2.1.5 セキュリティファースト（Security-First）

エンタープライズでの使用を前提として、A2Aは最初からセキュリティを組み込んでいる。OAuth 2.0、OpenID Connect、Mutual TLS など、業界標準の認証・認可メカニズムをサポートし、Agent Card の署名検証や拡張Agent Card による段階的な情報開示も提供している。

2.2 レイヤードアーキテクチャ

A2Aプロトコルは、関心の分離を実現するために3層のレイヤードアーキテクチャを採用している。

+-----------------------------------------------+
|         Layer 3: Protocol Bindings             |
|   JSON-RPC 2.0 | gRPC | HTTP/REST             |
+-----------------------------------------------+
|         Layer 2: Abstract Operations           |
|   SendMessage | GetTask | CancelTask | ...     |
+-----------------------------------------------+
|         Layer 1: Canonical Data Model          |
|   Protocol Buffers (.proto) 定義               |
|   Task, Message, Part, Artifact, AgentCard     |
+-----------------------------------------------+

Layer 1: 正規データモデル（Canonical Data Model）

最下層は Protocol Buffers（protobuf）で定義された正規データモデルである。spec/a2a.proto ファイルが権威ある定義であり、すべてのSDKバインディングやJSON成果物はこのソースから生成される。

このアプローチにより、以下の利点が得られる。

型安全性: 厳密に型付けされたデータ構造により、ランタイムエラーを防止
言語中立性: protobuf から各言語のコードを自動生成
バージョン管理: protobuf の互換性ルールに基づいた安全な進化
効率性: バイナリシリアライゼーションによる効率的なデータ転送（gRPCバインディングの場合）

Layer 2: 抽象オペレーション（Abstract Operations）

中間層は、プロトコルバインディングに依存しない抽象的な操作を定義する。これらの操作は、エージェントが提供すべき機能的なインターフェースを記述する。

主要な抽象オペレーションは以下の通りである。

オペレーション	説明
Send Message	エージェントとの対話を開始し、Task または Message を返す
Send Streaming Message	リアルタイム更新を伴うメッセージ送信
Get Task	タスクの現在の状態とアーティファクトを取得
List Tasks	フィルタリングとページネーションによるタスク一覧の取得
Cancel Task	タスクのキャンセルを要求
Subscribe to Task	リアルタイムのタスク更新をストリーミング
Push Notification Config	Webhookベースの非同期更新の管理
Get Extended Agent Card	認証済みクライアント向けの詳細な Agent Card を取得

Layer 3: プロトコルバインディング（Protocol Bindings）

最上層は、抽象オペレーションを具体的な通信プロトコルにマッピングする。A2Aは3つのプロトコルバインディングをサポートしている。

JSON-RPC 2.0: 最も広く採用されているバインディング。メソッドベースの呼び出しと名前付きパラメータを使用する。Web APIに馴染みのある開発者にとって最も理解しやすい。

{
  "jsonrpc": "2.0",
  "method": "message/send",
  "id": "req-001",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "東京の明日の天気を教えてください"
        }
      ]
    }
  }
}

gRPC: Protocol Buffers メッセージを使用するサービスベースのバインディング。高パフォーマンスが必要な場合や、既存のgRPCインフラストラクチャとの統合に適している。

HTTP/REST: 標準的なRESTエンドポイントとJSONペイロードを使用するバインディング。既存のHTTPインフラストラクチャとの互換性が最も高い。

2.3 クライアント-サーバーモデル

A2Aの基本的な通信モデルは、クライアント-サーバーアーキテクチャに基づいている。

+------------------+                    +------------------+
|   A2A Client     |                    |   A2A Server     |
|  (呼び出し側の    | --- Request --->   |  (タスクを実行    |
|   エージェント)   |                    |   するエージェント) |
|                  | <-- Response ---   |                  |
+------------------+                    +------------------+

A2A Client（クライアント）: タスクを依頼する側のエージェント。他のエージェントの Agent Card を発見し、メッセージを送信し、タスクの進行状況を監視する。
A2A Server（サーバー）: タスクを実行する側のエージェント。Agent Card を公開し、受信したメッセージに基づいてタスクを処理し、結果を返す。

重要な点として、一つのエージェントが同時にクライアントとサーバーの両方の役割を果たすことができる。例えば、オーケストレーターエージェントは、ユーザーからの要求を受け取るサーバーとして動作しながら、専門エージェントに対してはクライアントとして動作する。

ユーザー --> [オーケストレーター] --> [専門エージェントA]
               (Server & Client)     (Server)
                    |
                    +--------------> [専門エージェントB]
                                      (Server)

3. コアコンセプト: Agent Card, Task, Message, Artifact, Streaming

A2Aプロトコルを理解するためには、5つのコアコンセプトを深く理解する必要がある。これらはプロトコルの基盤であり、すべての通信パターンはこれらの概念の組み合わせで構成される。

3.1 Agent Card — エージェントの名刺

Agent Card は、エージェントの身分証明書とカタログを兼ねたJSON形式のメタデータドキュメントである。人間がWebサイトの「About」ページを見てサービス内容を理解するように、AIエージェントは Agent Card を読んで相手エージェントの能力と通信方法を理解する。

Agent Card の構造

Agent Card は以下の主要セクションで構成される。

{
  "name": "Weather Information Agent",
  "description": "世界中の都市の天気情報を提供するエージェント",
  "version": "1.2.0",
  "provider": {
    "organization": "Weather Corp",
    "url": "https://weathercorp.example.com"
  },
  "url": "https://api.weathercorp.example.com/a2a",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "extendedAgentCard": true
  },
  "skills": [
    {
      "id": "current-weather",
      "name": "Current Weather",
      "description": "指定された都市の現在の天気情報を返します",
      "tags": ["weather", "temperature", "forecast"],
      "examples": [
        "東京の現在の天気を教えてください",
        "What is the weather in New York?"
      ]
    },
    {
      "id": "weather-forecast",
      "name": "Weather Forecast",
      "description": "指定された都市の天気予報を最大7日間提供します",
      "tags": ["weather", "forecast", "prediction"]
    }
  ],
  "interfaces": [
    {
      "protocol": "jsonrpc",
      "url": "https://api.weathercorp.example.com/a2a/jsonrpc"
    },
    {
      "protocol": "rest",
      "url": "https://api.weathercorp.example.com/a2a/rest"
    }
  ],
  "securitySchemes": {
    "bearer": {
      "type": "http",
      "scheme": "bearer",
      "bearerFormat": "JWT"
    },
    "oauth2": {
      "type": "oauth2",
      "flows": {
        "clientCredentials": {
          "tokenUrl": "https://auth.weathercorp.example.com/token",
          "scopes": {
            "weather:read": "天気情報の読み取り"
          }
        }
      }
    }
  },
  "security": [
    { "bearer": [] }
  ],
  "defaultInputModes": ["text"],
  "defaultOutputModes": ["text", "application/json"]
}

各フィールドの詳細

基本情報フィールド

name: エージェントの名前。人間が読みやすい短い名前
description: エージェントの機能の説明。他のエージェントがこのエージェントに何を依頼できるかを判断するために使用
version: セマンティックバージョニング（SemVer）に従うバージョン文字列
provider: エージェントを提供する組織の情報
url: エージェントのベースURL

ケイパビリティ（capabilities）

capabilities オブジェクトは、エージェントがサポートする機能的な能力を宣言する。

streaming: true の場合、エージェントは Server-Sent Events (SSE) を使用したストリーミングレスポンスをサポートする
pushNotifications: true の場合、エージェントはWebhookを使用したプッシュ通知をサポートする。長時間実行されるタスクの完了通知に有用
extendedAgentCard: true の場合、認証済みクライアントに対してより詳細な Agent Card を提供する

スキル（skills）

スキルは、エージェントが実行できる具体的な能力を記述する。各スキルには一意のID、名前、説明、タグ、および入力例が含まれる。クライアントエージェントはスキル情報を使用して、特定のタスクに最適なエージェントを選択できる。

インターフェース（interfaces）

エージェントがサポートするプロトコルバインディングを宣言する。クライアントは利用可能なインターフェースから互換性のあるものを選択して通信する。

セキュリティスキーム（securitySchemes）

エージェントが受け入れる認証メカニズムを定義する。OpenAPI仕様のセキュリティスキームに類似した構造を採用している。

Public Agent Card と Extended Agent Card

A2Aは二段階の情報開示をサポートしている。

Public Agent Card: 認証不要でアクセスできる公開情報。エージェントの基本的な能力と接続方法を含む。ただし、機密性の高いスキルや内部的なエンドポイントは含まない場合がある。

Extended Agent Card: 認証済みのクライアントのみがアクセスできる拡張情報。追加のスキル、詳細な設定パラメータ、内部向けのケイパビリティなど、公開 Agent Card には含まれない情報を提供する。

認証なし --> Public Agent Card（基本情報のみ）
認証済み --> Extended Agent Card（全情報）

Agent Card の署名と検証

Agent Card はデジタル署名を付与でき、受信者はその署名を検証してカードの真正性と完全性を確認できる。これは、なりすまし攻撃や改ざん攻撃からの保護に不可欠である。

Agent Card のキャッシュ

Agent Card のキャッシュは HTTP キャッシュヘッダーを通じて管理される。サーバーは適切な Cache-Control ヘッダーを設定し、クライアントはそれを尊重してキャッシュの有効期限内は再取得を行わない。バージョンが変更された場合は再検証が必要となる。

3.2 Task — 作業の基本単位

Task は A2A プロトコルにおける最も重要なデータ構造であり、エージェント間で交換される作業の基本単位である。すべてのエージェント間通信は、最終的にタスクの作成、更新、完了として表現される。

Task のデータ構造

{
  "id": "task-abc123",
  "contextId": "ctx-session-789",
  "status": {
    "state": "WORKING",
    "message": {
      "role": "agent",
      "parts": [
        {
          "kind": "text",
          "text": "天気情報を取得中です..."
        }
      ]
    },
    "timestamp": "2026-04-08T10:30:00Z"
  },
  "artifacts": [],
  "history": [
    {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "東京の天気を教えてください"
        }
      ]
    }
  ],
  "metadata": {}
}

Task の主要フィールド

id: タスクの一意識別子。サーバーによって生成される。クライアントはこのIDを使用してタスクの状態を照会、更新、キャンセルする
contextId: 関連するタスクやメッセージをグループ化するコンテキストID。マルチターンの会話において、同一セッション内のタスクを関連付けるために使用される
status: タスクの現在のステータス。状態（state）、オプションのメッセージ、タイムスタンプを含む
artifacts: タスクの実行によって生成された成果物のコレクション
history: タスクに関連するメッセージの履歴
metadata: 任意の追加メタデータ

Task のライフサイクル状態

Task は以下の状態を遷移する（詳細は第6章で解説）。

PENDING --> WORKING --> COMPLETED
                   |-> INPUT_REQUIRED --> WORKING
                   |-> AUTH_REQUIRED --> WORKING
                   |-> FAILED
                   |-> CANCELED
                   |-> REJECTED

3.3 Message — 通信の原子単位

Message は、エージェント間で交換される通信の原子単位である。各メッセージにはロール（送信者の種別）とパーツ（コンテンツ）が含まれる。

Message のデータ構造

{
  "role": "user",
  "parts": [
    {
      "kind": "text",
      "text": "この画像に写っている建物を特定してください"
    },
    {
      "kind": "file",
      "file": {
        "mimeType": "image/jpeg",
        "uri": "https://example.com/images/building.jpg"
      }
    }
  ],
  "metadata": {
    "language": "ja",
    "priority": "high"
  }
}

ロール（Role）

Message のロールは2種類ある。

user: クライアント（要求元）からのメッセージ。タスクの開始や追加情報の提供に使用される
agent: サーバー（実行側エージェント）からのメッセージ。進捗報告、質問、結果の返信に使用される

Part — コンテンツの構成要素

Part はメッセージの実際のコンテンツを表す。A2Aは3種類のパーツをサポートしている。

TextPart: プレーンテキストコンテンツ

{
  "kind": "text",
  "text": "分析結果: 売上は前月比15%増加しました"
}

FilePart: ファイル参照またはインラインファイルデータ

{
  "kind": "file",
  "file": {
    "name": "report.pdf",
    "mimeType": "application/pdf",
    "uri": "https://storage.example.com/reports/2026-q1.pdf"
  }
}

または、Base64エンコードされたインラインデータとして:

{
  "kind": "file",
  "file": {
    "name": "chart.png",
    "mimeType": "image/png",
    "bytes": "iVBORw0KGgoAAAANSUhEUgAA..."
  }
}

DataPart: 構造化データ（JSON）

{
  "kind": "data",
  "data": {
    "temperature": 22.5,
    "humidity": 65,
    "condition": "晴れ",
    "city": "東京",
    "unit": "celsius"
  }
}

DataPart は、エージェント間で構造化された情報を正確に交換する場合に特に有用である。テキストとは異なり、パースの曖昧さがなく、受信側のエージェントがデータを直接プログラム的に処理できる。

3.4 Artifact — タスクの成果物

Artifact は、タスクの実行によって生成される成果物を表す。メッセージが通信手段であるのに対し、アーティファクトはタスクの最終的な出力物である。

Artifact のデータ構造

{
  "name": "weather-report",
  "description": "東京の天気レポート",
  "parts": [
    {
      "kind": "text",
      "text": "## 東京の天気レポート\n\n現在の気温: 22.5°C\n天候: 晴れ\n湿度: 65%"
    },
    {
      "kind": "data",
      "data": {
        "temperature": 22.5,
        "humidity": 65,
        "condition": "sunny",
        "windSpeed": 12,
        "windDirection": "NW"
      }
    }
  ],
  "metadata": {
    "generatedAt": "2026-04-08T10:30:00Z",
    "source": "Japan Meteorological Agency"
  }
}

Message と Artifact の違い

この区別は重要である。

特性	Message	Artifact
目的	通信・対話	最終的な成果物
方向	双方向（user/agent）	エージェントからのみ
持続性	会話履歴として保存	タスクの出力として保存
用途	質問、指示、進捗報告	レポート、データ、ファイル

例えば、翻訳タスクにおいて：

Message: 「この文書を日本語に翻訳してください」（user）→「翻訳を開始します」（agent）→「いくつか確認があります」（agent）
Artifact: 翻訳された文書全体

3.5 Streaming — リアルタイム通信

A2Aプロトコルは、リアルタイムのタスク更新を提供するためのストリーミングメカニズムをサポートしている。

ストリーミングの仕組み

ストリーミングは Server-Sent Events (SSE) プロトコルに基づいている。クライアントが Send Streaming Message オペレーションを呼び出すと、サーバーはHTTP接続を維持したまま、タスクの進行状況をイベントのストリームとして送信する。

クライアント                     サーバー
    |                              |
    |-- Send Streaming Message --> |
    |                              |-- タスク処理開始
    |<-- SSE: status=WORKING ------|
    |<-- SSE: partial result ------|
    |<-- SSE: partial result ------|
    |<-- SSE: status=COMPLETED ----|
    |<-- SSE: artifact ------------|
    |                              |

ストリーミングイベントの種類

ストリーミングで受信できるイベントには以下の種類がある。

TaskStatusUpdateEvent: タスクの状態変更を通知する
TaskArtifactUpdateEvent: 新しいアーティファクトの生成またはアーティファクトの更新を通知する

ストリーミングの利点

低レイテンシ: ポーリングと比較して、ほぼリアルタイムの更新が可能
帯域幅効率: 必要な更新のみが送信される
ユーザー体験の向上: LLMの出力がトークンごとに表示される「タイピング効果」を実現可能
長時間タスクの監視: タスクの進行状況をリアルタイムで追跡可能

3つの更新配信メカニズムの比較

A2Aは3つの更新配信メカニズムを提供しており、ユースケースに応じて適切なものを選択できる。

メカニズム	方式	レイテンシ	適用場面
ポーリング	クライアントが定期的に Get Task を呼び出す	高	シンプルな実装、低頻度の更新
ストリーミング（SSE）	持続的HTTP接続でイベントを配信	低	リアルタイム更新が必要な場合
プッシュ通知	サーバーがクライアントのWebhookにPOST	中	非同期処理、サーバー間通信

4. 通信パターンとデータフロー

4.1 基本的な通信フロー

A2Aの最も基本的な通信パターンは、リクエスト-レスポンス型の同期通信である。

1. クライアントがサーバーの Agent Card を取得
2. クライアントが Send Message を呼び出す
3. サーバーがタスクを作成し処理を開始
4. サーバーが結果（Task）を返す

シーケンス図: 基本的なリクエスト-レスポンス

Client                          Server
  |                               |
  |--- GET /.well-known/agent ---> |  (1) Agent Card 取得
  |<-- Agent Card (JSON) ---------|
  |                               |
  |--- message/send ------------> |  (2) メッセージ送信
  |                               |  (3) タスク処理
  |<-- Task (COMPLETED) ----------|  (4) 結果返却
  |                               |

4.2 マルチターン通信

多くの現実的なシナリオでは、タスクの完了に複数回の対話が必要になる。例えば、エージェントが追加情報を要求する場合がある。

Client                          Server
  |                               |
  |--- message/send ------------> |  "ホテルを予約して"
  |<-- Task (INPUT_REQUIRED) -----|  "どの都市ですか？"
  |                               |
  |--- message/send (taskId) ---> |  "東京です"
  |<-- Task (INPUT_REQUIRED) -----|  "日程はいつですか？"
  |                               |
  |--- message/send (taskId) ---> |  "4月15日から3泊"
  |<-- Task (WORKING) ------------|  "検索中です..."
  |                               |
  |--- task/get (taskId) -------> |  (ポーリングで状態確認)
  |<-- Task (COMPLETED) ----------|  予約完了
  |                               |

マルチターンの通信では、taskId を使用して既存のタスクを参照する。これにより、サーバーはタスクのコンテキスト（これまでの会話履歴）を維持できる。

4.3 ストリーミング通信

ストリーミングを使用した通信パターンでは、サーバーは処理の進行中にリアルタイムで更新を送信する。

Client                          Server
  |                               |
  |--- message/stream ---------> |  "この論文を要約して"
  |<== SSE: TaskStatusUpdate =====|  state: WORKING
  |<== SSE: TaskArtifactUpdate ===|  部分的な要約...
  |<== SSE: TaskArtifactUpdate ===|  要約の続き...
  |<== SSE: TaskStatusUpdate =====|  state: COMPLETED
  |                               |

4.4 プッシュ通知パターン

長時間実行されるタスクの場合、クライアントはWebhook URLを登録して、タスクの状態変更時にプッシュ通知を受け取ることができる。

Client                          Server
  |                               |
  |--- pushNotification/set ----> |  Webhook URL を登録
  |<-- 設定完了 ------------------|
  |                               |
  |--- message/send ------------> |  "大規模データを分析して"
  |<-- Task (PENDING) ------------|
  |                               |
  |  (クライアントは他の処理を実行)  |
  |                               |  (サーバーがバックグラウンドで処理)
  |                               |
  |<-- POST /webhook -------------|  state: WORKING (50%完了)
  |                               |
  |<-- POST /webhook -------------|  state: COMPLETED + artifacts
  |                               |

4.5 エージェント間の委任パターン

実際のマルチエージェントシステムでは、オーケストレーターエージェントが複数の専門エージェントにタスクを委任するパターンが一般的である。

ユーザー     オーケストレーター   旅行エージェント  ホテルエージェント
  |               |                   |                  |
  |-- "旅行計画" ->|                   |                  |
  |               |--- message/send -->|                  |
  |               |                   |-- フライト検索 -->|
  |               |<-- Task(flights) --|                  |
  |               |                   |                  |
  |               |--- message/send ---|----------------->|
  |               |                   |                  |-- ホテル検索
  |               |<-- Task(hotels) --|------------------|
  |               |                   |                  |
  |               |-- 結果統合 --------|                  |
  |<-- 旅行計画 ---|                   |                  |
  |               |                   |                  |

4.6 Context ID による会話の継続

A2Aでは、contextId を使用して関連する複数のタスクをグループ化できる。これにより、一連の対話が論理的なセッションとして管理される。

// 最初のメッセージ（contextId はサーバーが生成）
{
  "method": "message/send",
  "params": {
    "message": {
      "role": "user",
      "parts": [{"kind": "text", "text": "新しいプロジェクトを始めたい"}]
    }
  }
}

// レスポンスに contextId が含まれる
{
  "id": "task-001",
  "contextId": "ctx-abc123",
  "status": {"state": "COMPLETED", ...}
}

// 同じコンテキストで続ける
{
  "method": "message/send",
  "params": {
    "message": {
      "role": "user",
      "parts": [{"kind": "text", "text": "予算はどのくらい必要？"}],
      "contextId": "ctx-abc123"
    }
  }
}

5. エージェント探索とケイパビリティネゴシエーション

5.1 エージェント探索の仕組み

A2Aプロトコルにおけるエージェント探索は、クライアントエージェントが利用可能なサーバーエージェントを見つけ出すプロセスである。これはWebの世界における DNS やサービスディスカバリに相当する概念である。

Well-Known URL パターン

A2Aでは、エージェントの Agent Card は Well-Known URL パターンに従って公開される。具体的には、エージェントのベースURLに /.well-known/agent.json を付加したURLで Agent Card にアクセスできる。

https://agent.example.com/.well-known/agent.json

このパターンは RFC 8615 の Well-Known URI 仕様に準拠しており、Webインフラストラクチャとの互換性が高い。

探索フロー

1. クライアントが既知のURL（またはレジストリ）からエージェントを発見
2. Agent Card を取得（GET /.well-known/agent.json）
3. Agent Card のケイパビリティを分析
4. 必要に応じて Extended Agent Card を取得（認証後）
5. 適切なインターフェースとセキュリティスキームを選択
6. 通信を開始

5.2 ケイパビリティネゴシエーション

ケイパビリティネゴシエーションは、クライアントとサーバーが通信に使用するプロトコル、認証方式、コンテンツタイプなどを合意するプロセスである。

ステップ 1: ケイパビリティの確認

クライアントは Agent Card の capabilities フィールドを調べ、サーバーがサポートする機能を確認する。

{
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "extendedAgentCard": true
  }
}

ステップ 2: インターフェースの選択

Agent Card に複数のインターフェースが宣言されている場合、クライアントは自身がサポートするプロトコルの中から最適なものを選択する。

{
  "interfaces": [
    {"protocol": "jsonrpc", "url": "https://agent.example.com/jsonrpc"},
    {"protocol": "grpc", "url": "grpc://agent.example.com:50051"},
    {"protocol": "rest", "url": "https://agent.example.com/api/v1"}
  ]
}

例えば、高パフォーマンスが必要な場合は gRPC を選択し、既存のHTTPインフラストラクチャとの互換性を重視する場合は JSON-RPC または REST を選択する。

ステップ 3: 認証の設定

securitySchemes と security フィールドに基づいて、適切な認証方式を設定する。

{
  "securitySchemes": {
    "oauth2": {
      "type": "oauth2",
      "flows": {
        "clientCredentials": {
          "tokenUrl": "https://auth.example.com/token",
          "scopes": {"agent:execute": "タスク実行"}
        }
      }
    }
  },
  "security": [{"oauth2": ["agent:execute"]}]
}

ステップ 4: コンテンツタイプの確認

defaultInputModes と defaultOutputModes を確認し、交換可能なコンテンツタイプを理解する。

5.3 動的なエージェント選択

高度なマルチエージェントシステムでは、オーケストレーターエージェントが複数のエージェントの Agent Card を比較分析し、特定のタスクに最適なエージェントを動的に選択する。

# 擬似コード: 動的エージェント選択
def select_best_agent(task_description, available_agents):
    """タスクの説明に基づいて最適なエージェントを選択する"""
    best_match = None
    best_score = 0

    for agent_card in available_agents:
        for skill in agent_card.skills:
            score = calculate_relevance(task_description, skill)
            if score > best_score:
                best_score = score
                best_match = agent_card

    return best_match

このパターンにより、新しいエージェントが追加された場合でも、オーケストレーターは Agent Card を通じて自動的にそのエージェントを発見し、適切なタスクに割り当てることができる。

5.4 エージェントレジストリ

大規模なエコシステムでは、個々のエージェントのURLを手動で管理するのは非現実的である。エージェントレジストリは、利用可能なエージェントのカタログを提供するサービスであり、スキル、タグ、カテゴリに基づいた検索が可能である。

A2A仕様自体にはレジストリの標準的な実装は含まれていないが、Agent Card の標準化されたフォーマットにより、レジストリの構築は比較的容易である。

[エージェントレジストリ]
  |
  |-- search("weather") --> [Weather Agent Card]
  |-- search("translation") --> [Translation Agent Card]
  |-- search("data-analysis") --> [Analysis Agent Card]
  |
  クライアントはレジストリに問い合わせて
  必要なエージェントを発見する

6. タスクライフサイクル管理

6.1 タスク状態の定義

A2Aプロトコルは、タスクのライフサイクルを以下の8つの状態で管理する。

状態	説明	遷移先
`PENDING`	タスクが受け付けられ、処理待ち	WORKING, REJECTED, CANCELED
`WORKING`	タスクが処理中	COMPLETED, FAILED, INPUT_REQUIRED, AUTH_REQUIRED, CANCELED
`INPUT_REQUIRED`	エージェントがユーザーからの追加入力を待っている	WORKING, CANCELED
`AUTH_REQUIRED`	エージェントが追加の認証を要求している	WORKING, CANCELED
`COMPLETED`	タスクが正常に完了した	（終了状態）
`FAILED`	タスクの処理が失敗した	（終了状態）
`CANCELED`	ユーザーがタスクをキャンセルした	（終了状態）
`REJECTED`	エージェントがタスクを拒否した	（終了状態）

6.2 状態遷移図

                         +----------+
                    +--->| REJECTED |
                    |    +----------+
                    |
+--------+    +---------+    +-----------+
| PENDING|--->| WORKING |--->| COMPLETED |
+--------+    +---------+    +-----------+
                    |
                    +--->+-----------+
                    |    | FAILED    |
                    |    +-----------+
                    |
                    +--->+----------------+
                    |    | INPUT_REQUIRED |---+
                    |    +----------------+   |
                    |         ^               |
                    |         |  (入力提供)     |
                    |         +---------------+
                    |
                    +--->+---------------+
                    |    | AUTH_REQUIRED |---+
                    |    +---------------+   |
                    |         ^              |
                    |         |  (認証完了)    |
                    |         +--------------+
                    |
                    +--->+----------+
                         | CANCELED |
                         +----------+

6.3 各状態の詳細

PENDING（保留）

タスクがサーバーによって受け付けられたが、まだ処理が開始されていない状態。キューに入っている場合や、リソースの確保を待っている場合にこの状態となる。

{
  "id": "task-001",
  "status": {
    "state": "PENDING",
    "timestamp": "2026-04-08T10:00:00Z"
  }
}

WORKING（処理中）

エージェントがタスクを積極的に処理している状態。ストリーミング接続の場合、この状態中に部分的な結果や進捗情報が送信される。

{
  "id": "task-001",
  "status": {
    "state": "WORKING",
    "message": {
      "role": "agent",
      "parts": [{"kind": "text", "text": "データを分析中です（50%完了）"}]
    },
    "timestamp": "2026-04-08T10:05:00Z"
  }
}

INPUT_REQUIRED（入力待ち）

エージェントがタスクを続行するために追加の情報を必要としている状態。クライアントは必要な情報を含むメッセージを送信して、タスクを WORKING 状態に戻す。

{
  "id": "task-001",
  "status": {
    "state": "INPUT_REQUIRED",
    "message": {
      "role": "agent",
      "parts": [{
        "kind": "text",
        "text": "分析の期間を指定してください（例: 2025年1月〜2026年3月）"
      }]
    },
    "timestamp": "2026-04-08T10:03:00Z"
  }
}

AUTH_REQUIRED（認証待ち）

エージェントがタスクを続行するために追加の認証を要求している状態。例えば、外部サービスへのアクセスに追加のOAuthトークンが必要な場合に使用される。

COMPLETED（完了）

タスクが正常に完了した終了状態。成果物（artifacts）が含まれる。

{
  "id": "task-001",
  "status": {
    "state": "COMPLETED",
    "timestamp": "2026-04-08T10:10:00Z"
  },
  "artifacts": [
    {
      "name": "analysis-report",
      "parts": [{
        "kind": "text",
        "text": "分析結果: ..."
      }]
    }
  ]
}

FAILED（失敗）

タスクの処理中にエラーが発生し、完了できなかった状態。

CANCELED（キャンセル）

クライアントの要求によりタスクがキャンセルされた状態。task/cancel オペレーションで発生する。

REJECTED（拒否）

エージェントがタスクの実行を拒否した状態。エージェントの能力外の要求、ポリシー違反、リソース不足などの理由で発生する。

6.4 タスクのキャンセル

クライアントは task/cancel オペレーションを使用してタスクのキャンセルを要求できる。ただし、キャンセルは即座に反映されるとは限らない。エージェントは進行中の処理を安全に停止するために時間を要する場合がある。

// キャンセル要求
{
  "method": "task/cancel",
  "params": {
    "taskId": "task-001"
  }
}

// キャンセル応答
{
  "id": "task-001",
  "status": {
    "state": "CANCELED",
    "message": {
      "role": "agent",
      "parts": [{"kind": "text", "text": "タスクがキャンセルされました"}]
    }
  }
}

6.5 タスクの一覧取得とフィルタリング

task/list オペレーションにより、タスクの一覧を取得できる。フィルタリングとページネーションがサポートされている。

{
  "method": "task/list",
  "params": {
    "contextId": "ctx-abc123",
    "status": ["WORKING", "PENDING"],
    "limit": 10,
    "offset": 0
  }
}

7. セキュリティモデルと認証

7.1 セキュリティの設計思想

A2Aのセキュリティモデルは、エンタープライズグレードのセキュリティ要件を満たすように設計されている。以下の原則に基づいている。

多層防御（Defense in Depth）: 認証、認可、暗号化の複数のレイヤーで保護
最小権限（Least Privilege）: クライアントは必要最小限のリソースにのみアクセス可能
情報の段階的開示: Extended Agent Card により、認証レベルに応じた情報開示
標準準拠: 業界標準の認証・認可プロトコルを採用

7.2 サポートされる認証スキーム

A2Aは以下の認証スキームをサポートしている。

API Key 認証

最もシンプルな認証方式。ヘッダーまたはクエリパラメータでAPIキーを送信する。

{
  "type": "apiKey",
  "name": "X-API-Key",
  "in": "header"
}

GET /a2a/task/task-001 HTTP/1.1
Host: agent.example.com
X-API-Key: sk-your-api-key-here

使用場面: 開発環境、シンプルなサーバー間通信 注意点: 本番環境では他の方式との併用を推奨

HTTP Basic/Bearer 認証

HTTP標準の認証ヘッダーを使用する方式。

{
  "type": "http",
  "scheme": "bearer",
  "bearerFormat": "JWT"
}

GET /a2a/task/task-001 HTTP/1.1
Host: agent.example.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIs...

使用場面: JWT ベースの認証、マイクロサービス間通信

OAuth 2.0

最も広く採用されている認可フレームワーク。複数のフロー（Grant Type）をサポートする。

{
  "type": "oauth2",
  "flows": {
    "clientCredentials": {
      "tokenUrl": "https://auth.example.com/oauth2/token",
      "scopes": {
        "agent:read": "タスクの読み取り",
        "agent:write": "タスクの作成・更新",
        "agent:admin": "管理操作"
      }
    },
    "authorizationCode": {
      "authorizationUrl": "https://auth.example.com/oauth2/authorize",
      "tokenUrl": "https://auth.example.com/oauth2/token",
      "scopes": {
        "agent:read": "タスクの読み取り"
      }
    }
  }
}

Client Credentials フロー: エージェント間のサーバー対サーバー認証に最適。人間の介入なしにトークンを取得できる。

Authorization Code フロー: ユーザーの認可が必要な場合に使用。ユーザーが特定のエージェントにアクセス権を委任する場合に適している。

OpenID Connect

OAuth 2.0の上に構築された認証レイヤー。ユーザーのアイデンティティ確認が必要な場合に使用する。

{
  "type": "openIdConnect",
  "openIdConnectUrl": "https://auth.example.com/.well-known/openid-configuration"
}

Mutual TLS（相互TLS）

クライアントとサーバーの双方がX.509証明書を提示して相互に認証する方式。最も強力な認証メカニズムの一つ。

{
  "type": "mutualTLS"
}

使用場面: 金融、医療、政府機関など、高セキュリティが要求される環境

7.3 認可モデル

A2Aの認可モデルは以下の原則に基づいている。

タスクレベルの認可: クライアントは自身が作成したタスク、または明示的にアクセス権を付与されたタスクにのみアクセスできる
リソース存在の非開示: サーバーは、クライアントがアクセス権を持たないリソースの存在を開示してはならない。認可されていないタスクIDに対するリクエストは、403（Forbidden）ではなく 404（Not Found）を返すべきである
スコープベースの権限: OAuth 2.0 スコープを使用して、操作レベルの細かい権限制御が可能

7.4 Agent Card の署名と検証

Agent Card のデジタル署名は、なりすまし攻撃からの保護に不可欠である。

1. サーバーが Agent Card を正規化（Canonicalization）
2. 正規化されたデータに秘密鍵で署名
3. 署名を Agent Card の AgentCardSignature フィールドに格納
4. クライアントが署名を公開鍵で検証

7.5 通信の暗号化

A2Aプロトコルは、すべての通信に TLS（Transport Layer Security）の使用を要求する。これにより、通信内容の盗聴や改ざんが防止される。

本番環境では TLS 1.2 以上の使用が推奨され、TLS 1.3 がベストプラクティスとされている。

7.6 セキュリティベストプラクティス

A2Aを本番環境で運用する際のセキュリティベストプラクティスを以下にまとめる。

認証: OAuth 2.0 Client Credentials フローまたは Mutual TLS を使用する
暗号化: TLS 1.3 を使用し、すべての通信を暗号化する
トークン管理: アクセストークンの有効期間を短く設定し、リフレッシュトークンを使用する
レート制限: DoS攻撃を防ぐためにリクエストレート制限を実装する
監査ログ: すべてのエージェント間通信を監査ログに記録する
Agent Card の検証: 受信した Agent Card の署名を常に検証する
Extended Agent Card の使用: 機密性の高いスキルや設定は Extended Agent Card でのみ公開する
入力検証: 受信したメッセージやデータの内容を検証し、インジェクション攻撃を防ぐ

8. 既存エージェントフレームワークとの統合

8.1 A2A とエージェントフレームワークの関係

A2Aプロトコルは特定のエージェントフレームワークに依存しない設計となっているが、主要なフレームワークとの統合が公式にサポートされている。A2Aはフレームワーク間の「共通語」として機能し、異なるフレームワークで構築されたエージェントが標準化されたインターフェースを通じて通信できるようにする。

+------------------+     +---------+     +------------------+
| LangChain Agent  |<--->|         |<--->| CrewAI Agent     |
+------------------+     |         |     +------------------+
                          |  A2A   |
+------------------+     |Protocol |     +------------------+
| Google ADK Agent |<--->|         |<--->| AutoGen Agent    |
+------------------+     |         |     +------------------+
                          |         |
+------------------+     |         |     +------------------+
| Semantic Kernel  |<--->|         |<--->| Custom Agent     |
+------------------+     +---------+     +------------------+

8.2 Google ADK（Agent Development Kit）との統合

Google ADK は A2A プロトコルの最もネイティブな統合を提供するフレームワークである。ADKはGoogleが開発したエージェント開発キットであり、A2Aサポートが組み込まれている。

# Google ADK を使用した A2A サーバーの例
from google.adk.agents import Agent
from google.adk.a2a import A2AServer

class WeatherAgent(Agent):
    """天気情報を提供するエージェント"""

    def __init__(self):
        super().__init__(
            name="Weather Agent",
            description="世界中の都市の天気情報を提供します",
            skills=[
                {
                    "id": "current-weather",
                    "name": "Current Weather",
                    "description": "指定された都市の現在の天気を返します",
                    "tags": ["weather", "temperature"]
                }
            ]
        )

    async def handle_message(self, message):
        # メッセージの処理ロジック
        city = self.extract_city(message)
        weather_data = await self.fetch_weather(city)
        return self.create_response(weather_data)

# A2A サーバーとして起動
server = A2AServer(agent=WeatherAgent())
server.run(host="0.0.0.0", port=8080)

8.3 LangChain / LangGraph との統合

LangChainは最も広く使用されているLLMアプリケーションフレームワークの一つであり、LangGraphはその上に構築されたステートフルなエージェントワークフローフレームワークである。

A2Aとの統合は、LangChainエージェントをA2Aサーバーとしてラップすることで実現できる。

# LangChain エージェントを A2A サーバーとしてラップする例
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from a2a.server import A2AServer, AgentCard, AgentSkill

# LangChain エージェントの構築
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [search_tool, calculator_tool, weather_tool]
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# A2A Agent Card の定義
agent_card = AgentCard(
    name="Research Assistant",
    description="調査、計算、天気情報の提供ができる研究アシスタント",
    version="1.0.0",
    skills=[
        AgentSkill(
            id="research",
            name="Web Research",
            description="Webを検索して情報を収集します",
            tags=["search", "research"]
        ),
        AgentSkill(
            id="calculate",
            name="Calculator",
            description="数学的な計算を実行します",
            tags=["math", "calculation"]
        )
    ],
    capabilities={
        "streaming": True,
        "pushNotifications": False
    }
)

# A2A サーバーのハンドラー
class LangChainA2AHandler:
    def __init__(self, executor):
        self.executor = executor

    async def handle_message(self, message, task):
        """A2A メッセージを LangChain の入力に変換して処理"""
        user_input = self.extract_text(message)

        # LangChain エージェントを実行
        result = await self.executor.ainvoke({"input": user_input})

        # 結果を A2A の Artifact に変換
        return {
            "status": "COMPLETED",
            "artifacts": [{
                "name": "research-result",
                "parts": [{"kind": "text", "text": result["output"]}]
            }]
        }

# A2A サーバーの起動
handler = LangChainA2AHandler(agent_executor)
server = A2AServer(agent_card=agent_card, handler=handler)
server.start(port=8080)

8.4 CrewAI との統合

CrewAIは、複数のエージェントがチームとして協調動作するフレームワークである。A2Aとの統合により、CrewAIのクルーを外部エージェントに公開できる。

# CrewAI のクルーを A2A サーバーとして公開する例
from crewai import Agent, Task, Crew
from a2a.server import A2AServer, AgentCard

# CrewAI エージェントの定義
researcher = Agent(
    role="Research Analyst",
    goal="包括的な市場調査を実施する",
    backstory="経験豊富な市場調査アナリスト"
)

writer = Agent(
    role="Content Writer",
    goal="調査結果を明確なレポートにまとめる",
    backstory="テクニカルライター"
)

# CrewAI タスクの定義
research_task = Task(
    description="指定されたトピックについて調査する: {topic}",
    agent=researcher
)

writing_task = Task(
    description="調査結果をレポートにまとめる",
    agent=writer,
    context=[research_task]
)

# クルーの定義
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task]
)

# A2A ハンドラー
class CrewAIA2AHandler:
    def __init__(self, crew):
        self.crew = crew

    async def handle_message(self, message, task):
        topic = self.extract_topic(message)
        result = self.crew.kickoff(inputs={"topic": topic})

        return {
            "status": "COMPLETED",
            "artifacts": [{
                "name": "research-report",
                "parts": [{"kind": "text", "text": str(result)}]
            }]
        }

# A2A サーバーとして起動
agent_card = AgentCard(
    name="Market Research Crew",
    description="市場調査と報告を行うエージェントチーム",
    skills=[{
        "id": "market-research",
        "name": "Market Research",
        "description": "指定されたトピックの市場調査レポートを作成"
    }]
)

server = A2AServer(agent_card=agent_card, handler=CrewAIA2AHandler(crew))
server.start(port=8081)

8.5 Microsoft Semantic Kernel との統合

Semantic Kernelは、MicrosoftのAIオーケストレーションSDKであり、C#/.NETエコシステムで広く採用されている。

// Semantic Kernel エージェントを A2A サーバーとして公開する例（C#）
using Microsoft.SemanticKernel;
using A2A.Server;

public class SemanticKernelA2AAgent : IA2AHandler
{
    private readonly Kernel _kernel;

    public SemanticKernelA2AAgent()
    {
        _kernel = Kernel.CreateBuilder()
            .AddOpenAIChatCompletion("gpt-4o", apiKey)
            .Build();

        // プラグインの追加
        _kernel.ImportPluginFromType<WeatherPlugin>();
        _kernel.ImportPluginFromType<TranslationPlugin>();
    }

    public async Task<A2AResponse> HandleMessage(
        A2AMessage message, A2ATask task)
    {
        var userInput = message.GetTextContent();

        var result = await _kernel.InvokePromptAsync(userInput);

        return new A2AResponse
        {
            Status = TaskStatus.Completed,
            Artifacts = new[]
            {
                new Artifact
                {
                    Name = "result",
                    Parts = new[] { new TextPart(result.ToString()) }
                }
            }
        };
    }
}

// サーバーの起動
var agentCard = new AgentCard
{
    Name = "SK Assistant",
    Description = "Semantic Kernel ベースの多機能アシスタント",
    Skills = new[] { /* ... */ }
};

var server = new A2AServer(agentCard, new SemanticKernelA2AAgent());
await server.StartAsync(port: 8082);

8.6 統合パターンのまとめ

フレームワーク	言語	統合の容易さ	特徴
Google ADK	Python	ネイティブサポート	A2A が組み込まれている
LangChain/LangGraph	Python	SDKによる統合	豊富なツールエコシステム
CrewAI	Python	SDKによる統合	マルチエージェントチームの公開
Semantic Kernel	C#, Python	SDKによる統合	エンタープライズ向け
AutoGen	Python	カスタム統合	会話型エージェント

9. MCP (Model Context Protocol) との比較

9.1 根本的な違い

A2AとMCPは、AIエージェントエコシステムにおける異なる問題を解決するために設計されたプロトコルである。両者は競合するものではなく、相互に補完する関係にある。

MCP（Model Context Protocol）: エージェントがツールやリソース（データベース、API、ファイルシステムなど）にアクセスするためのプロトコル。エージェントと外部機能との接続を標準化する。

A2A（Agent2Agent Protocol）: 独立したエージェント同士が協調してタスクを実行するためのプロトコル。エージェント間の通信を標準化する。

+------------------------------------------------------+
|                  AIエージェントシステム                  |
|                                                        |
|  +-----------+   A2A   +-----------+                   |
|  | Agent A   |<------->| Agent B   |                   |
|  |           |         |           |                   |
|  | +-------+ |         | +-------+ |                   |
|  | | MCP   | |         | | MCP   | |                   |
|  | +---+---+ |         | +---+---+ |                   |
|  +-----|-----+         +-----|-----+                   |
|        |                     |                         |
|  +-----v-----+         +----v------+                   |
|  |  Tools    |         |  Tools    |                   |
|  | DB, API   |         | Files,    |                   |
|  | Search    |         | Services  |                   |
|  +-----------+         +-----------+                   |
+------------------------------------------------------+

9.2 対象ドメインの違い

観点	MCP	A2A
対話の相手	ツール、リソース	他のエージェント
相手の特性	明確な入出力を持つプリミティブ	自律的に推論・計画する存在
ステートフルネス	主にステートレス	ステートフル（タスクの状態管理）
対話のパターン	関数呼び出し	マルチターンの対話
不確実性	低い（定義された入出力）	高い（エージェントの判断に依存）

9.3 具体的な比較

MCP のアプローチ

MCPでは、エージェントはツールの能力記述を読み、構造化された入力を提供し、構造化された出力を受け取る。

// MCP: ツールの定義
{
  "name": "get_weather",
  "description": "指定された都市の天気を取得",
  "parameters": {
    "type": "object",
    "properties": {
      "city": {"type": "string"},
      "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
    },
    "required": ["city"]
  }
}

// MCP: ツールの呼び出し
{
  "name": "get_weather",
  "arguments": {"city": "Tokyo", "unit": "celsius"}
}

// MCP: ツールの応答
{
  "temperature": 22.5,
  "condition": "sunny",
  "humidity": 65
}

A2A のアプローチ

A2Aでは、エージェントは自然言語や構造化データでタスクを依頼し、相手エージェントが自律的に処理する。

// A2A: メッセージの送信
{
  "method": "message/send",
  "params": {
    "message": {
      "role": "user",
      "parts": [{
        "kind": "text",
        "text": "東京の天気レポートを作成してください。降水確率、紫外線指数、花粉情報も含めて"
      }]
    }
  }
}

// A2A: 追加入力の要求
{
  "status": {"state": "INPUT_REQUIRED"},
  "message": {
    "role": "agent",
    "parts": [{
      "kind": "text",
      "text": "どの期間のレポートが必要ですか？本日のみ、または週間予報も含めますか？"
    }]
  }
}

// A2A: 最終結果
{
  "status": {"state": "COMPLETED"},
  "artifacts": [{
    "name": "weather-report",
    "parts": [
      {"kind": "text", "text": "# 東京天気レポート\n\n...詳細な分析..."},
      {"kind": "data", "data": {"temperature": 22.5, ...}}
    ]
  }]
}

9.4 補完的な使い方

実際のシステムでは、A2A と MCP は同時に使用されることが多い。

具体例: カスタマーサービスシステム

[ユーザー] --> [カスタマーサービスエージェント]
                  |
                  |-- A2A --> [請求エージェント]
                  |              |-- MCP --> [データベース]
                  |              |-- MCP --> [請求API]
                  |
                  |-- A2A --> [技術サポートエージェント]
                  |              |-- MCP --> [ナレッジベース]
                  |              |-- MCP --> [チケットシステム]
                  |
                  |-- A2A --> [配送エージェント]
                                 |-- MCP --> [配送追跡API]
                                 |-- MCP --> [在庫システム]

この例では：

A2A はエージェント間のコラボレーションに使用（カスタマーサービスエージェントが専門エージェントにタスクを委任）
MCP は各エージェントが内部的にツールやリソースにアクセスするために使用

9.5 選択のガイドライン

シナリオ	推奨プロトコル	理由
データベースクエリの実行	MCP	明確な入出力、ステートレス
API の呼び出し	MCP	構造化された操作
別チームが管理するサービスの利用	A2A	自律的な処理、不透明性
マルチターンの調査依頼	A2A	対話的な処理が必要
ファイルの読み書き	MCP	シンプルなリソースアクセス
複雑な分析タスクの委任	A2A	推論と計画が必要
外部ベンダーのAIサービスとの連携	A2A	フレームワーク非依存の通信

9.6 他のプロトコルとの比較

A2A と MCP 以外にも、エージェント関連のプロトコルや規格が存在する。

プロトコル	目的	対象
A2A	エージェント間通信	エージェント ↔ エージェント
MCP	エージェント-ツール連携	エージェント ↔ ツール
OpenAI Function Calling	LLMのツール使用	LLM ↔ 関数
AutoGen Protocol	会話型マルチエージェント	エージェント ↔ エージェント（AutoGen内）
FIPA ACL	エージェント通信言語	エージェント ↔ エージェント（学術的）

A2Aの差別化要因は、フレームワーク非依存であること、Web標準に基づいていること、そして実用的なエンタープライズユースケースに焦点を当てていることである。

10. 実装例と詳細コード

本章では、A2Aプロトコルの実装例を Python SDK を使用して詳しく解説する。基本的なエコーサーバーから始まり、実用的なマルチエージェントシステムまでを段階的に構築する。

10.1 環境セットアップ

必要なパッケージのインストール

# A2A Python SDK のインストール
pip install a2a-sdk

# 追加の依存関係
pip install uvicorn fastapi httpx

プロジェクト構造

my-a2a-project/
├── agents/
│   ├── __init__.py
│   ├── echo_agent.py          # エコーエージェント（基本例）
│   ├── weather_agent.py       # 天気エージェント
│   └── orchestrator_agent.py  # オーケストレーターエージェント
├── client/
│   ├── __init__.py
│   └── a2a_client.py          # A2A クライアント
├── config/
│   └── agent_cards.json       # Agent Card 設定
├── tests/
│   ├── test_echo.py
│   └── test_multi_agent.py
├── requirements.txt
└── README.md

10.2 基本例: エコーエージェント

最もシンプルなA2Aサーバーとして、受信したメッセージをそのまま返すエコーエージェントを実装する。

# agents/echo_agent.py
"""
A2A エコーエージェント - 受信メッセージをエコーバックする最もシンプルなA2Aサーバー
"""

import uuid
from datetime import datetime, timezone
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# Agent Card の定義
AGENT_CARD = {
    "name": "Echo Agent",
    "description": "受信したメッセージをそのまま返すシンプルなエージェント",
    "version": "1.0.0",
    "provider": {
        "organization": "Example Corp"
    },
    "url": "http://localhost:8080",
    "capabilities": {
        "streaming": False,
        "pushNotifications": False
    },
    "skills": [
        {
            "id": "echo",
            "name": "Echo",
            "description": "受信したメッセージをそのまま返します",
            "tags": ["echo", "test"],
            "examples": ["Hello, World!"]
        }
    ],
    "defaultInputModes": ["text"],
    "defaultOutputModes": ["text"]
}

# タスクストア（メモリ内）
tasks = {}


@app.get("/.well-known/agent.json")
async def get_agent_card():
    """Agent Card を返すエンドポイント"""
    return JSONResponse(content=AGENT_CARD)


@app.post("/a2a")
async def handle_jsonrpc(request: Request):
    """JSON-RPC リクエストを処理するメインエンドポイント"""
    body = await request.json()
    method = body.get("method")
    params = body.get("params", {})
    request_id = body.get("id")

    if method == "message/send":
        return await handle_send_message(params, request_id)
    elif method == "task/get":
        return await handle_get_task(params, request_id)
    elif method == "task/cancel":
        return await handle_cancel_task(params, request_id)
    else:
        return JSONResponse(content={
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {
                "code": -32601,
                "message": f"Method not found: {method}"
            }
        })


async def handle_send_message(params, request_id):
    """メッセージ送信の処理"""
    message = params.get("message", {})
    task_id = params.get("taskId") or str(uuid.uuid4())

    # メッセージからテキストを抽出
    text_parts = [
        part["text"] for part in message.get("parts", [])
        if part.get("kind") == "text"
    ]
    user_text = " ".join(text_parts)

    # エコーレスポンスの作成
    task = {
        "id": task_id,
        "status": {
            "state": "COMPLETED",
            "timestamp": datetime.now(timezone.utc).isoformat()
        },
        "artifacts": [
            {
                "name": "echo-response",
                "parts": [
                    {
                        "kind": "text",
                        "text": f"Echo: {user_text}"
                    }
                ]
            }
        ],
        "history": [message]
    }

    tasks[task_id] = task

    return JSONResponse(content={
        "jsonrpc": "2.0",
        "id": request_id,
        "result": task
    })


async def handle_get_task(params, request_id):
    """タスク取得の処理"""
    task_id = params.get("taskId")
    task = tasks.get(task_id)

    if not task:
        return JSONResponse(content={
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {
                "code": -32602,
                "message": f"Task not found: {task_id}"
            }
        })

    return JSONResponse(content={
        "jsonrpc": "2.0",
        "id": request_id,
        "result": task
    })


async def handle_cancel_task(params, request_id):
    """タスクキャンセルの処理"""
    task_id = params.get("taskId")
    task = tasks.get(task_id)

    if not task:
        return JSONResponse(content={
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {"code": -32602, "message": "Task not found"}
        })

    task["status"] = {
        "state": "CANCELED",
        "timestamp": datetime.now(timezone.utc).isoformat()
    }

    return JSONResponse(content={
        "jsonrpc": "2.0",
        "id": request_id,
        "result": task
    })


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

10.3 A2A クライアントの実装

# client/a2a_client.py
"""
A2A クライアント - A2A サーバーと通信するためのクライアントライブラリ
"""

import httpx
import uuid
from typing import Optional, Dict, Any, List


class A2AClient:
    """A2A プロトコルクライアント"""

    def __init__(self, base_url: str, auth_token: Optional[str] = None):
        self.base_url = base_url.rstrip("/")
        self.headers = {"Content-Type": "application/json"}
        if auth_token:
            self.headers["Authorization"] = f"Bearer {auth_token}"
        self._client = httpx.AsyncClient(headers=self.headers)
        self._agent_card = None

    async def discover(self) -> Dict[str, Any]:
        """Agent Card を取得してエージェントのケイパビリティを発見する"""
        response = await self._client.get(
            f"{self.base_url}/.well-known/agent.json"
        )
        response.raise_for_status()
        self._agent_card = response.json()
        return self._agent_card

    async def send_message(
        self,
        text: str,
        task_id: Optional[str] = None,
        context_id: Optional[str] = None,
        data: Optional[Dict] = None
    ) -> Dict[str, Any]:
        """メッセージを送信してタスクを作成または更新する"""
        parts = [{"kind": "text", "text": text}]
        if data:
            parts.append({"kind": "data", "data": data})

        params = {
            "message": {
                "role": "user",
                "parts": parts
            }
        }

        if task_id:
            params["taskId"] = task_id
        if context_id:
            params["message"]["contextId"] = context_id

        return await self._jsonrpc_call("message/send", params)

    async def get_task(self, task_id: str) -> Dict[str, Any]:
        """タスクの現在の状態を取得する"""
        return await self._jsonrpc_call("task/get", {"taskId": task_id})

    async def cancel_task(self, task_id: str) -> Dict[str, Any]:
        """タスクをキャンセルする"""
        return await self._jsonrpc_call("task/cancel", {"taskId": task_id})

    async def list_tasks(
        self,
        context_id: Optional[str] = None,
        status: Optional[List[str]] = None,
        limit: int = 10,
        offset: int = 0
    ) -> Dict[str, Any]:
        """タスクの一覧を取得する"""
        params = {"limit": limit, "offset": offset}
        if context_id:
            params["contextId"] = context_id
        if status:
            params["status"] = status
        return await self._jsonrpc_call("task/list", params)

    async def _jsonrpc_call(
        self, method: str, params: Dict[str, Any]
    ) -> Dict[str, Any]:
        """JSON-RPC 呼び出しを実行する"""
        request_body = {
            "jsonrpc": "2.0",
            "method": method,
            "id": str(uuid.uuid4()),
            "params": params
        }

        response = await self._client.post(
            f"{self.base_url}/a2a",
            json=request_body
        )
        response.raise_for_status()
        result = response.json()

        if "error" in result:
            raise A2AError(
                result["error"]["code"],
                result["error"]["message"]
            )

        return result.get("result")

    async def close(self):
        """クライアントを閉じる"""
        await self._client.aclose()

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()


class A2AError(Exception):
    """A2A プロトコルエラー"""
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message
        super().__init__(f"A2A Error [{code}]: {message}")

10.4 クライアントの使用例

# examples/basic_usage.py
"""A2A クライアントの基本的な使用例"""

import asyncio
from client.a2a_client import A2AClient


async def main():
    async with A2AClient("http://localhost:8080") as client:
        # 1. エージェントの発見
        agent_card = await client.discover()
        print(f"エージェント名: {agent_card['name']}")
        print(f"スキル: {[s['name'] for s in agent_card['skills']]}")

        # 2. メッセージの送信
        result = await client.send_message("こんにちは、A2Aの世界！")
        print(f"タスクID: {result['id']}")
        print(f"状態: {result['status']['state']}")

        # 3. アーティファクトの取得
        for artifact in result.get("artifacts", []):
            for part in artifact.get("parts", []):
                if part["kind"] == "text":
                    print(f"応答: {part['text']}")

        # 4. タスクの再取得
        task = await client.get_task(result["id"])
        print(f"再取得した状態: {task['status']['state']}")


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

10.5 ストリーミング対応エージェント

# agents/streaming_agent.py
"""ストリーミング対応の A2A エージェント"""

import asyncio
import json
import uuid
from datetime import datetime, timezone
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, JSONResponse

app = FastAPI()

AGENT_CARD = {
    "name": "Streaming Analysis Agent",
    "description": "テキストの段階的な分析結果をストリーミングで返すエージェント",
    "version": "1.0.0",
    "capabilities": {
        "streaming": True,
        "pushNotifications": False
    },
    "skills": [
        {
            "id": "text-analysis",
            "name": "Text Analysis",
            "description": "テキストの多面的な分析をリアルタイムで提供",
            "tags": ["analysis", "nlp"]
        }
    ]
}


@app.get("/.well-known/agent.json")
async def get_agent_card():
    return JSONResponse(content=AGENT_CARD)


@app.post("/a2a")
async def handle_request(request: Request):
    body = await request.json()
    method = body.get("method")

    if method == "message/stream":
        return await handle_streaming_message(body)
    elif method == "message/send":
        return await handle_send_message(body)


async def handle_streaming_message(body):
    """ストリーミングメッセージの処理"""
    params = body.get("params", {})
    message = params.get("message", {})
    task_id = str(uuid.uuid4())

    async def generate_events():
        # タスク開始イベント
        yield format_sse_event({
            "type": "TaskStatusUpdateEvent",
            "taskId": task_id,
            "status": {
                "state": "WORKING",
                "message": {
                    "role": "agent",
                    "parts": [{"kind": "text", "text": "分析を開始します..."}]
                },
                "timestamp": datetime.now(timezone.utc).isoformat()
            }
        })

        # 分析の各段階をストリーミング
        analysis_steps = [
            "テキストの構造を解析中...",
            "キーワードを抽出中...",
            "感情分析を実行中...",
            "要約を生成中..."
        ]

        for step in analysis_steps:
            await asyncio.sleep(1)  # 処理のシミュレーション
            yield format_sse_event({
                "type": "TaskStatusUpdateEvent",
                "taskId": task_id,
                "status": {
                    "state": "WORKING",
                    "message": {
                        "role": "agent",
                        "parts": [{"kind": "text", "text": step}]
                    },
                    "timestamp": datetime.now(timezone.utc).isoformat()
                }
            })

        # アーティファクトの送信
        yield format_sse_event({
            "type": "TaskArtifactUpdateEvent",
            "taskId": task_id,
            "artifact": {
                "name": "analysis-result",
                "parts": [
                    {
                        "kind": "text",
                        "text": "## 分析結果\n\n"
                                "- キーワード: AI, プロトコル, エージェント\n"
                                "- 感情: ニュートラル\n"
                                "- カテゴリ: テクノロジー\n"
                    },
                    {
                        "kind": "data",
                        "data": {
                            "keywords": ["AI", "プロトコル", "エージェント"],
                            "sentiment": "neutral",
                            "confidence": 0.92,
                            "category": "technology"
                        }
                    }
                ]
            }
        })

        # 完了イベント
        yield format_sse_event({
            "type": "TaskStatusUpdateEvent",
            "taskId": task_id,
            "status": {
                "state": "COMPLETED",
                "timestamp": datetime.now(timezone.utc).isoformat()
            }
        })

    return StreamingResponse(
        generate_events(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive"
        }
    )


def format_sse_event(data: dict) -> str:
    """Server-Sent Events 形式にフォーマットする"""
    return f"data: {json.dumps(data)}\n\n"


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

10.6 マルチエージェント: オーケストレーターの実装

# agents/orchestrator_agent.py
"""
オーケストレーターエージェント
複数の専門エージェントにタスクを委任し、結果を統合する
"""

import asyncio
from typing import Dict, List, Any
from client.a2a_client import A2AClient

class OrchestratorAgent:
    """複数のA2Aエージェントをオーケストレーションするエージェント"""

    def __init__(self):
        self.agents: Dict[str, A2AClient] = {}
        self.agent_cards: Dict[str, Dict] = {}

    async def register_agent(self, name: str, url: str):
        """エージェントを登録し、Agent Card を取得する"""
        client = A2AClient(url)
        card = await client.discover()
        self.agents[name] = client
        self.agent_cards[name] = card
        print(f"エージェント '{card['name']}' を登録しました")
        print(f"  スキル: {[s['name'] for s in card.get('skills', [])]}")

    def find_agent_for_task(self, task_description: str) -> str:
        """タスクの説明に基づいて最適なエージェントを見つける"""
        best_match = None
        best_score = 0

        for name, card in self.agent_cards.items():
            for skill in card.get("skills", []):
                # スキルのタグとタスクの説明を照合
                tags = skill.get("tags", [])
                description = skill.get("description", "").lower()
                score = sum(
                    1 for tag in tags
                    if tag.lower() in task_description.lower()
                )
                if any(
                    word in task_description.lower()
                    for word in description.split()
                ):
                    score += 2

                if score > best_score:
                    best_score = score
                    best_match = name

        return best_match

    async def delegate_task(
        self, agent_name: str, message: str
    ) -> Dict[str, Any]:
        """特定のエージェントにタスクを委任する"""
        client = self.agents.get(agent_name)
        if not client:
            raise ValueError(f"Agent '{agent_name}' not found")

        result = await client.send_message(message)

        # INPUT_REQUIRED の場合はマルチターン処理
        while result.get("status", {}).get("state") == "INPUT_REQUIRED":
            # エージェントからの質問を取得
            agent_question = self._extract_text(
                result["status"].get("message", {})
            )
            print(f"  [{agent_name}] 追加入力が必要: {agent_question}")

            # 自律的に回答を生成（実際にはLLMで処理）
            answer = await self._generate_answer(agent_question)
            result = await client.send_message(
                answer, task_id=result["id"]
            )

        return result

    async def execute_parallel_tasks(
        self, tasks: List[Dict[str, str]]
    ) -> List[Dict[str, Any]]:
        """複数のタスクを並列に実行する"""
        coroutines = [
            self.delegate_task(task["agent"], task["message"])
            for task in tasks
        ]
        results = await asyncio.gather(*coroutines, return_exceptions=True)
        return results

    async def execute_workflow(self, user_request: str) -> Dict[str, Any]:
        """
        ユーザーのリクエストを分析し、適切なエージェントに
        タスクを委任するワークフローを実行する
        """
        print(f"\n{'='*60}")
        print(f"ワークフロー開始: {user_request}")
        print(f"{'='*60}")

        # ステップ 1: リクエストの分析と計画
        plan = await self._create_plan(user_request)
        print(f"\n実行計画:")
        for i, step in enumerate(plan, 1):
            print(f"  {i}. [{step['agent']}] {step['task']}")

        # ステップ 2: タスクの実行
        results = []
        for step in plan:
            print(f"\n--- {step['agent']} にタスクを委任 ---")
            result = await self.delegate_task(
                step["agent"], step["task"]
            )
            results.append({
                "agent": step["agent"],
                "task": step["task"],
                "result": result
            })

            state = result.get("status", {}).get("state")
            print(f"  結果: {state}")

            if state == "FAILED":
                print(f"  エラー: タスクが失敗しました")

        # ステップ 3: 結果の統合
        final_report = self._compile_results(results)
        print(f"\n{'='*60}")
        print(f"ワークフロー完了")
        print(f"{'='*60}")

        return final_report

    def _extract_text(self, message: Dict) -> str:
        """メッセージからテキストを抽出する"""
        parts = message.get("parts", [])
        texts = [p["text"] for p in parts if p.get("kind") == "text"]
        return " ".join(texts)

    async def _create_plan(self, request: str) -> List[Dict]:
        """リクエストを分析して実行計画を作成する（簡略版）"""
        # 実際にはLLMを使用してリクエストを分析する
        return [
            {"agent": "weather", "task": f"Weather info for: {request}"},
            {"agent": "research", "task": f"Research on: {request}"}
        ]

    async def _generate_answer(self, question: str) -> str:
        """質問に対する回答を生成する（簡略版）"""
        return f"Auto-generated answer for: {question}"

    def _compile_results(
        self, results: List[Dict]
    ) -> Dict[str, Any]:
        """複数のエージェントの結果を統合する"""
        compiled_artifacts = []
        for r in results:
            result_data = r.get("result", {})
            for artifact in result_data.get("artifacts", []):
                compiled_artifacts.append({
                    "source_agent": r["agent"],
                    "artifact": artifact
                })

        return {
            "status": "COMPLETED",
            "compiled_results": compiled_artifacts
        }

    async def close(self):
        """すべてのクライアント接続を閉じる"""
        for client in self.agents.values():
            await client.close()


# 使用例
async def main():
    orchestrator = OrchestratorAgent()

    # エージェントの登録
    await orchestrator.register_agent(
        "weather", "http://localhost:8081"
    )
    await orchestrator.register_agent(
        "research", "http://localhost:8082"
    )

    # ワークフローの実行
    result = await orchestrator.execute_workflow(
        "東京への出張を計画してください。天気と観光情報が必要です"
    )

    print(f"\n最終結果: {result}")
    await orchestrator.close()


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

10.7 テストコード

# tests/test_echo.py
"""エコーエージェントのテスト"""

import pytest
import asyncio
from client.a2a_client import A2AClient


@pytest.mark.asyncio
async def test_agent_card_discovery():
    """Agent Card の取得をテスト"""
    async with A2AClient("http://localhost:8080") as client:
        card = await client.discover()
        assert card["name"] == "Echo Agent"
        assert len(card["skills"]) > 0
        assert card["skills"][0]["id"] == "echo"


@pytest.mark.asyncio
async def test_echo_message():
    """エコーメッセージの送受信をテスト"""
    async with A2AClient("http://localhost:8080") as client:
        result = await client.send_message("Hello, A2A!")
        assert result["status"]["state"] == "COMPLETED"
        assert len(result["artifacts"]) > 0

        text = result["artifacts"][0]["parts"][0]["text"]
        assert "Hello, A2A!" in text


@pytest.mark.asyncio
async def test_task_retrieval():
    """タスクの再取得をテスト"""
    async with A2AClient("http://localhost:8080") as client:
        result = await client.send_message("Test message")
        task_id = result["id"]

        retrieved = await client.get_task(task_id)
        assert retrieved["id"] == task_id
        assert retrieved["status"]["state"] == "COMPLETED"


@pytest.mark.asyncio
async def test_task_cancellation():
    """タスクのキャンセルをテスト"""
    async with A2AClient("http://localhost:8080") as client:
        result = await client.send_message("To be canceled")
        task_id = result["id"]

        canceled = await client.cancel_task(task_id)
        assert canceled["status"]["state"] == "CANCELED"

11. 実世界のユースケース

11.1 エンタープライズカスタマーサポート

シナリオ

大規模なEコマース企業のカスタマーサポートシステムを考える。顧客からの問い合わせは、注文状況、返品、技術的な問題、請求に関する質問など多岐にわたる。

アーキテクチャ

[顧客チャット]
     |
     v
[フロントデスクエージェント] --- A2A 探索 ---> [エージェントレジストリ]
     |
     |--- A2A ---> [注文管理エージェント]
     |                  |--- MCP ---> 注文DB
     |                  |--- MCP ---> 配送API
     |
     |--- A2A ---> [返品・返金エージェント]
     |                  |--- MCP ---> 返品ポリシーDB
     |                  |--- MCP ---> 決済API
     |
     |--- A2A ---> [技術サポートエージェント]
     |                  |--- MCP ---> ナレッジベース
     |                  |--- MCP ---> デバイス診断ツール
     |
     |--- A2A ---> [請求エージェント]
                       |--- MCP ---> 請求システム
                       |--- MCP ---> 税計算API

フロントデスクエージェントの実装概要

class CustomerSupportOrchestrator:
    """カスタマーサポートのフロントデスクエージェント"""

    async def handle_customer_inquiry(self, inquiry: str):
        # 1. 問い合わせの分類
        category = await self.classify_inquiry(inquiry)

        # 2. 適切な専門エージェントを選択
        agent = self.select_specialist(category)

        # 3. タスクを委任
        result = await self.delegate_to_agent(agent, inquiry)

        # 4. INPUT_REQUIRED の場合、顧客に追加情報を要求
        while result.status.state == "INPUT_REQUIRED":
            customer_input = await self.ask_customer(
                result.status.message
            )
            result = await agent.send_message(
                customer_input, task_id=result.id
            )

        # 5. 結果を顧客に返す
        return self.format_response(result)

ビジネス上の効果

応答時間の短縮: 自動ルーティングにより、最初から正しい専門エージェントに接続
スケーラビリティ: 新しい専門エージェントの追加が容易（Agent Card を公開するだけ）
ベンダー柔軟性: 各専門エージェントを異なるベンダーのソリューションに置き換え可能
品質の向上: 各エージェントが専門分野に特化することで、回答精度が向上

11.2 サプライチェーン最適化

シナリオ

グローバルな製造業企業のサプライチェーン管理において、需要予測、在庫管理、調達、物流の各機能を専門エージェントが担当する。

エージェント間のワークフロー

[需要予測エージェント]
     |
     | (1) 需要予測データ
     v
[在庫管理エージェント] <--- (3) 納期回答 --- [調達エージェント]
     |                                           |
     | (2) 発注要求                               | (4) 発注
     v                                           v
[物流エージェント]                            [サプライヤーシステム]
     |
     | (5) 配送手配
     v
[倉庫管理システム]

各エージェントはA2Aを介して通信し、以下のような対話が行われる。

需要予測 -> 在庫管理: "来月の製品Aの需要は15,000個と予測"
在庫管理 -> 調達: "製品Aの部品を10,000個発注してください"
調達 -> 在庫管理: "納期は2週間後、単価は$2.50です（確認しますか？）"
在庫管理 -> 調達: "確認します。発注を進めてください"
調達 -> 在庫管理: "発注完了。注文番号: PO-2026-04-001"
在庫管理 -> 物流: "2週間後に10,000個の部品が到着。倉庫Aに搬入手配を"

11.3 金融サービス: コンプライアンス自動化

シナリオ

銀行のコンプライアンス部門で、取引モニタリング、KYC（本人確認）、規制報告の各プロセスを自動化する。

マルチエージェントアーキテクチャ

[取引モニタリングエージェント]
     | 不審な取引を検出
     v
[調査エージェント]
     |--- A2A ---> [KYCエージェント]      (顧客情報の確認)
     |--- A2A ---> [リスク評価エージェント] (リスクスコアの計算)
     |--- A2A ---> [規制報告エージェント]   (必要な報告書の作成)
     |
     v
[コンプライアンスオフィサー]（人間の最終判断）

このユースケースの特徴は、AUTH_REQUIRED 状態の活用である。KYCエージェントが外部の本人確認サービスにアクセスする際に追加の認証が必要になる場合がある。また、コンプライアンスの意思決定には人間の承認が必要な場合があり、INPUT_REQUIRED 状態でコンプライアンスオフィサーの判断を待つことができる。

11.4 ヘルスケア: 患者ケアコーディネーション

シナリオ

病院システムにおいて、患者の診察予約、検査オーダー、薬剤管理、保険請求を各専門エージェントが処理する。

[患者ポータルエージェント]
     |
     |--- A2A ---> [予約管理エージェント]
     |                  |--- 医師のスケジュール確認
     |                  |--- 検査室の空き確認
     |
     |--- A2A ---> [検査オーダーエージェント]
     |                  |--- 検査項目の確認
     |                  |--- 結果の通知
     |
     |--- A2A ---> [薬剤管理エージェント]
     |                  |--- 処方箋の確認
     |                  |--- 薬物相互作用チェック
     |
     |--- A2A ---> [保険請求エージェント]
                       |--- 保険適用の確認
                       |--- 請求書の作成

このユースケースでは、Mutual TLS による厳格な認証が必要であり、患者データの保護のために各エージェント間の通信は完全に暗号化される。また、Extended Agent Card を使用して、認証されたエージェントのみが患者データにアクセスできるスキルを利用できるようにする。

11.5 ソフトウェア開発: CI/CD パイプライン

シナリオ

ソフトウェア開発チームのCI/CDパイプラインにおいて、コードレビュー、テスト、デプロイ、モニタリングを各エージェントが自律的に実行する。

[開発者] --- コミット ---> [Git]
                             |
                             v
                    [CI/CDオーケストレーターエージェント]
                             |
                             |--- A2A ---> [コードレビューエージェント]
                             |                  |--- 静的解析
                             |                  |--- セキュリティスキャン
                             |                  |--- コーディング規約チェック
                             |
                             |--- A2A ---> [テストエージェント]
                             |                  |--- 単体テスト実行
                             |                  |--- 統合テスト実行
                             |                  |--- パフォーマンステスト
                             |
                             |--- A2A ---> [デプロイエージェント]
                             |                  |--- ステージング環境
                             |                  |--- 本番環境
                             |
                             |--- A2A ---> [モニタリングエージェント]
                                                |--- エラー率監視
                                                |--- パフォーマンス監視
                                                |--- ロールバック判断

ストリーミング機能を活用して、長時間実行されるテストの進捗をリアルタイムで報告することが可能である。また、テストが失敗した場合、テストエージェントは FAILED 状態を返し、オーケストレーターはデプロイを中止してコードレビューエージェントに失敗原因の分析を依頼する。

11.6 教育: パーソナライズド学習

シナリオ

オンライン学習プラットフォームにおいて、学習者のレベルに合わせたコンテンツ提供、進捗管理、質問応答を行う。

[学習者] --> [学習コーディネーターエージェント]
                  |
                  |--- A2A ---> [能力評価エージェント]
                  |--- A2A ---> [コンテンツ推薦エージェント]
                  |--- A2A ---> [チュータリングエージェント]
                  |--- A2A ---> [進捗追跡エージェント]

マルチターンの対話機能を活用して、チュータリングエージェントは学習者と段階的な質疑応答セッションを実施できる。INPUT_REQUIRED 状態で学習者の回答を待ち、その回答に基づいて次の質問や説明を提供する。

12. 将来のロードマップとエコシステム

12.1 プロトコルの進化

A2Aプロトコルは、Linux Foundation への寄贈を経て、オープンなガバナンスモデルの下で進化を続けている。今後の主要な開発方向性は以下の通りである。

プロトコルの拡張

マルチモーダル対応の強化: 画像、音声、動画などのマルチモーダルデータの交換がより自然にサポートされる予定。現在のFilePart によるファイル交換に加えて、ストリーミングでのマルチモーダルデータ配信が検討されている。
エージェントグループ（群）のサポート: 現在のA2Aは主にクライアント-サーバー間の1対1通信をモデル化しているが、エージェントの群（グループ）間での協調メカニズムが将来的に追加される可能性がある。
拡張メカニズムの充実: Agent Card の extensions フィールドを通じたプロトコル拡張のためのフレームワークがさらに充実し、ドメイン固有の拡張（金融、医療、製造など）が標準化される。
トレーサビリティとオブザーバビリティ: マルチエージェントシステムのデバッグと監視のために、分散トレーシング（OpenTelemetry統合など）のネイティブサポートが検討されている。

12.2 SDK とツールの拡充

現在提供されている SDK（Python, JavaScript, Java, C#/.NET, Go）に加えて、以下の拡充が進んでいる。

Rust SDK: パフォーマンスクリティカルなエッジデバイスでのエージェント実行向け
Swift SDK: Apple プラットフォームでのネイティブエージェント開発向け
開発者ツール: エージェントのテスト、デバッグ、モニタリングのための専用ツール
Agent Card ジェネレーター: OpenAPI仕様やgRPC定義からAgent Cardを自動生成するツール

12.3 エコシステムの拡大

エージェントマーケットプレイス

Agent Card の標準化により、エージェントのマーケットプレイスの構築が可能になる。開発者はA2A互換のエージェントを公開し、他の開発者やシステムがそれを発見して利用できる。

[エージェントマーケットプレイス]
     |
     |--- カテゴリ: ビジネス
     |       |--- 請求エージェント (Vendor A)
     |       |--- CRMエージェント (Vendor B)
     |       |--- 人事エージェント (Vendor C)
     |
     |--- カテゴリ: データ分析
     |       |--- BIエージェント (Vendor D)
     |       |--- 予測分析エージェント (Vendor E)
     |
     |--- カテゴリ: コミュニケーション
             |--- 翻訳エージェント (Vendor F)
             |--- メールエージェント (Vendor G)

業界標準化

A2Aが特定の業界で標準プロトコルとして採用されることで、エージェント間の相互運用性が飛躍的に向上する可能性がある。特に、金融（FINOS経由）、ヘルスケア（HL7/FHIR統合）、製造（Industry 4.0）などの業界での標準化が期待されている。

12.4 技術的な課題と展望

スケーラビリティ

数百、数千のエージェントが協調するシステムにおけるスケーラビリティの課題。エージェント探索、タスクルーティング、負荷分散のための効率的なメカニズムが必要となる。

セキュリティの進化

ゼロトラストアーキテクチャ: エージェント間のすべての通信が検証されるゼロトラストモデル
Verifiable Credentials: W3C Verifiable Credentials との統合による、エージェントの身元証明
Confidential Computing: 機密データを暗号化したまま処理するコンフィデンシャルコンピューティングとの統合

AI ガバナンス

監査証跡: エージェントの意思決定プロセスの完全な記録
説明可能性: エージェントがなぜ特定の判断を下したかの説明能力
バイアス検出: エージェントの出力におけるバイアスの検出と是正

12.5 まとめ

A2A（Agent2Agent Protocol）は、AI エージェントの相互運用性を実現するための重要な一歩である。Web が HTTP によって異なるシステム間の情報交換を標準化したように、A2A は AI エージェント間の通信を標準化し、真のマルチエージェントエコシステムの基盤を提供する。

本記事で解説した通り、A2Aは以下の特徴を持つ。

不透明性: エージェントの内部実装を公開せずに通信できる
標準化: Protocol Buffers に基づく正規データモデルと複数のプロトコルバインディング
柔軟性: 同期、ストリーミング、プッシュ通知など多様な通信パターンをサポート
セキュリティ: エンタープライズグレードの認証・認可メカニズム
エコシステム: 主要なエージェントフレームワークとの統合サポート

A2Aの採用は始まったばかりだが、Google をはじめとする主要テクノロジー企業のサポート、Linux Foundation によるオープンガバナンス、そして急速に成長するAIエージェントエコシステムにより、今後数年で広範な普及が期待される。

AIエージェントの開発者にとって、A2Aプロトコルの理解と実装は、将来のマルチエージェントシステム構築における重要なスキルとなるだろう。

参考文献

A2A Protocol Official Website: https://a2a-protocol.org
A2A Protocol Specification: https://a2a-protocol.org/latest/specification/
A2A GitHub Repository: https://github.com/a2aproject
Google Agent Development Kit (ADK): https://google.github.io/adk-docs/
Model Context Protocol (MCP): https://modelcontextprotocol.io
LangChain Documentation: https://python.langchain.com
CrewAI Documentation: https://docs.crewai.com
Microsoft Semantic Kernel: https://learn.microsoft.com/semantic-kernel/
JSON-RPC 2.0 Specification: https://www.jsonrpc.org/specification
gRPC Documentation: https://grpc.io/docs/
Protocol Buffers Documentation: https://protobuf.dev
OAuth 2.0 Framework (RFC 6749): https://datatracker.ietf.org/doc/html/rfc6749
OpenID Connect: https://openid.net/connect/
Linux Foundation AI & Data: https://lfaidata.foundation
Server-Sent Events (SSE): https://html.spec.whatwg.org/multipage/server-sent-events.html

A2A

A2A (Agent2Agent) 技術概要 — エージェント間通信プロトコルの全容

目次

1. はじめに — A2Aとは何か、なぜ重要なのか

1.1 AI エージェント時代の到来

1.2 A2A プロトコルの概要

1.3 なぜ A2A が必要なのか

1.4 A2A の歴史と現在

1.5 本記事の構成

2. プロトコルアーキテクチャと設計原則

2.1 設計思想

2.1.1 不透明性（Opaqueness）

2.1.2 相互運用性（Interoperability）

2.1.3 ケイパビリティ駆動（Capability-Driven）

2.1.4 タスク中心（Task-Centric）

2.1.5 セキュリティファースト（Security-First）

2.2 レイヤードアーキテクチャ

Layer 1: 正規データモデル（Canonical Data Model）

Layer 2: 抽象オペレーション（Abstract Operations）

Layer 3: プロトコルバインディング（Protocol Bindings）

2.3 クライアント-サーバーモデル

3. コアコンセプト: Agent Card, Task, Message, Artifact, Streaming

3.1 Agent Card — エージェントの名刺

Agent Card の構造

各フィールドの詳細

Public Agent Card と Extended Agent Card

Agent Card の署名と検証

Agent Card のキャッシュ

3.2 Task — 作業の基本単位

Task のデータ構造

Task の主要フィールド

Task のライフサイクル状態

3.3 Message — 通信の原子単位

Message のデータ構造

ロール（Role）

Part — コンテンツの構成要素

3.4 Artifact — タスクの成果物

Artifact のデータ構造

Message と Artifact の違い

3.5 Streaming — リアルタイム通信

ストリーミングの仕組み

ストリーミングイベントの種類

ストリーミングの利点

3つの更新配信メカニズムの比較

4. 通信パターンとデータフロー

4.1 基本的な通信フロー

シーケンス図: 基本的なリクエスト-レスポンス

4.2 マルチターン通信

4.3 ストリーミング通信

4.4 プッシュ通知パターン

4.5 エージェント間の委任パターン

4.6 Context ID による会話の継続

5. エージェント探索とケイパビリティネゴシエーション

5.1 エージェント探索の仕組み

Well-Known URL パターン

探索フロー

5.2 ケイパビリティネゴシエーション

ステップ 1: ケイパビリティの確認

ステップ 2: インターフェースの選択

ステップ 3: 認証の設定

ステップ 4: コンテンツタイプの確認

5.3 動的なエージェント選択

5.4 エージェントレジストリ

6. タスクライフサイクル管理

6.1 タスク状態の定義

6.2 状態遷移図

6.3 各状態の詳細

PENDING（保留）

WORKING（処理中）

INPUT_REQUIRED（入力待ち）

AUTH_REQUIRED（認証待ち）

COMPLETED（完了）

FAILED（失敗）

CANCELED（キャンセル）

REJECTED（拒否）

6.4 タスクのキャンセル

6.5 タスクの一覧取得とフィルタリング

7. セキュリティモデルと認証

7.1 セキュリティの設計思想

7.2 サポートされる認証スキーム