A2A over MQTT の仕組み

このページでは、A2A over MQTT の基本概念について説明します。エージェント、クライアント、ブローカーの構成、エージェント同士の識別と検出方法、エージェントカードの内容、エージェント間の通信パターンなどを解説します。これらの概念を理解することは、EMQX の A2A レジストリを活用するための基礎となります。

アーキテクチャ

A2A over MQTT は、ブローカー中心のモデルで以下の3者が参加します。

• エージェント（レスポンダー）: 自身のエージェントカードを検出トピックにリテインドメッセージとしてパブリッシュし、自身のリクエストトピックをサブスクライブして、受信したタスクリクエストに応答します。
• クライアントエージェント（リクエスター）: 検出トピックをサブスクライブして利用可能なエージェントを探索し、タスクリクエストを送信し、返信を受信します。
• MQTT ブローカー（EMQX）: すべてのメッセージをルーティングし、A2A レジストリにエージェントカードを記録し、認証と認可を実施し、検出メッセージにライブネスメタデータを付与します。

エージェント識別

各エージェントは3階層の階層構造で識別されます。

{org_id} / {unit_id} / {agent_id}

• org_id: エージェントが所属する組織（例: com.example）。
• unit_id: 組織内の部門やチーム、デプロイ環境などの区分（例: factory-a）。
• agent_id: 組織とユニット内で一意のエージェント識別子（例: iot-ops-agent-001）。

3つのセグメントすべては ^[A-Za-z0-9_.-]+$ にマッチし、/、+、#、空白文字を含んではいけません。エージェントの MQTT クライアントIDは、これらを組み合わせた {org_id}/{unit_id}/{agent_id} 形式を使用します。

トピックモデル

トピック	用途
$a2a/v1/discovery/{org_id}/{unit_id}/{agent_id}	エージェント登録および検出（リテインド）
$a2a/v1/request/{org_id}/{unit_id}/{agent_id}	特定エージェントへのタスクリクエスト受信
$a2a/v1/reply/{org_id}/{unit_id}/{agent_id}/{suffix}	推奨される返信トピックパターン（下記注記参照）
$a2a/v1/event/{org_id}/{unit_id}/{agent_id}	要求なしのイベントパブリッシュ
$a2a/v1/request/{org_id}/{unit_id}/pool/{pool_id}	ロードバランスされたディスパッチ用の共有プールトピック

注記

返信トピックは固定のプロトコルトピックではありません。リクエスターは任意のトピックを返信トピックとして使用できます。レスポンダーは、MQTT v5 の Response Topic プロパティから返信先を知ります。上記のパターンは一貫性を保ち、ACL設定を簡素化するための推奨例です。

検出サブスクリプションはワイルドカードを使って範囲を限定します。

$a2a/v1/discovery/com.example/+/+     # 組織内のすべてのエージェント
$a2a/v1/discovery/com.example/factory-a/+  # ユニット内のすべてのエージェント

エージェントカード

エージェントカードは、エージェントが自身の検出トピックにパブリッシュする JSON ドキュメントです。エージェントの識別情報、機能、HTTP エンドポイント、オプションのセキュリティメタデータを記述します。EMQX は受信時にカードを A2A レジストリに記録します。

最低限必要なフィールド：

フィールド	型	説明
name	文字列	人間が読みやすいエージェント名。
description	文字列	エージェントの概要説明。
version	文字列	バージョン文字列（例: "1.0.0"）。
url	文字列（URI）	エージェントのエンドポイントURI。任意。
skills	配列	少なくとも1つのスキルオブジェクト。各スキルは id、name、description を持つ。

最小限のエージェントカード例：

{
  "name": "IoT Operations Agent",
  "description": "工場のテレメトリを監視し、修復アクションを調整します。",
  "version": "1.2.3",
  "url": "mqtts://broker.example.com:8883",
  "skills": [
    {
      "id": "device-diagnostics",
      "name": "デバイス診断",
      "description": "テレメトリを分析し、デバイスの異常を検出します。"
    }
  ]
}

capabilities、securitySchemes、supportedInterfaces、拡張パラメータを含む完全なエージェントカードスキーマは、A2A仕様をご参照ください。

エージェントのライブネス

エージェントカードは、エージェントが切断された後もリテインドメッセージとして保持されます。EMQX は接続状態を追跡し、検出メッセージをサブスクライバーに転送する際に MQTT v5 のユーザープロパティを付与します。

ユーザープロパティ	値	意味
a2a-status	online	エージェントの MQTT 接続がアクティブ。
a2a-status	offline	エージェントが切断済み（正常切断または LWT による）。
a2a-status-source	broker	状態は EMQX によって設定された。
a2a-status-source	agent	状態はエージェント自身によって設定された。
a2a-status-source	lwt	状態は予期しない切断（ラストウィル）を反映。

エージェントは検出トピックに対して、a2a-status=offline と a2a-status-source=lwt を含むラストウィルメッセージを設定し、非正常切断時にサブスクライバーへ自動通知されるようにすべきです。

インタラクションパターン

A2A over MQTT は、エージェント間で以下のインタラクションパターンをサポートします。すべて MQTT v5 の Response Topic と Correlation Data プロパティを用いてリクエスト／リプライのルーティングを行い、リクエスター生成の Task.id でタスクの状態をライフサイクル全体にわたり追跡します。

パターン	説明
1リクエスト・1レスポンス	リクエスターがタスクリクエストをパブリッシュし、レスポンダーが指定された Response Topic に単一の返信をパブリッシュ。
ストリーミングレスポンス	レスポンダーが複数の状態および成果物更新メッセージをパブリッシュし、最終的なタスク完了状態に到達するまで続ける。
マルチターン会話	関連タスクを Task.context_id でグループ化し、中断されたタスクを再開可能にする。
共有プールディスパッチ	複数のエージェントインスタンスが MQTT の共有サブスクリプションを使い、プールトピックを共有してロードバランスされたリクエスト処理を行う。
タスクハンドオーバー	レスポンダーエージェントが進行中のタスクを別のインスタンスに a2a-responder-agent-id ユーザープロパティを使って委譲する。
OAuth 2.0 認可	リクエスト毎にベアラートークンを a2a-authorization MQTT ユーザープロパティとして渡す。
エンドツーエンドセキュリティ	オプションの ubsp-v1 セキュリティプロファイルにより、信頼できないブローカー環境でもペイロードをエンドツーエンドで暗号化可能。

各パターンの詳細な仕様は、A2A over MQTT トランスポート仕様をご覧ください。

例：工場アラート対応のワークフロー

2つのエージェントが協力して工場フロアのアラートに対応します。異常を検知して診断タスクを委譲する モニターエージェント と、それを処理して結果をストリーム配信する 修理エージェント です。

ステップ1: 両エージェントが登録。 各エージェントは自身のエージェントカードを検出トピックにリテインドメッセージとしてパブリッシュします。EMQX はカードを記録し、両エージェントをオンライン状態としてマークします。

ステップ2: モニターエージェントが修理エージェントを検出。 モニターエージェントは $a2a/v1/discovery/com.example/factory-a/+ をサブスクライブし、修理エージェントのリテインドカードを即座に受信、機器故障の診断が可能であることを確認します。

ステップ3: モニターエージェントがタスクリクエストを送信。 モーター line-7 から異常振動の読み取りが届きます。モニターエージェントは修理エージェントのリクエストトピックに、ユニークな Task.id と MQTT の Response Topic プロパティを設定してリクエストをパブリッシュします。

ステップ4: 修理エージェントが状態更新をストリーム配信。 修理エージェントは返信トピックに進捗更新をパブリッシュし、最終的に completed 状態（ベアリング摩耗検出、検査予定）を送信します。各更新は元の Correlation Data をエコーし、モニターエージェントがリクエストに紐付けて処理可能にします。