Agent2Agent 入門 (7) - ストリーミングと非同期操作
「Agent2Agent」のストリーミングと非同期操作についてまとめました。
前回
1. はじめに
「Agent2Agent」(A2A) は、すぐに完了しない可能性のあるタスクを処理するために設計されています。AI駆動型オペレーションの多くは、「長時間実行」や「複数ステップ」を伴い、増分的な結果を生成する、あるいは人間の介入を必要とする場合があります。A2Aは、こうした非同期のインタラクションを管理するための堅牢なメカニズムを提供し、クライアントが常時接続している場合も、より非接続で動作している場合も、更新を効果的に受信できるようにしています。
2. Server-Sent Events (SSE) によるストリーミング
増分的な結果を生成するタスク (長いドキュメントの生成やメディアのストリーミングなど) や、継続的なステータス更新を提供するタスクの場合、A2A は 「Server-Sent Events」(SSE) を使用したリアルタイム通信をサポートします。これは、クライアントがA2Aサーバとのアクティブな HTTP 接続を維持できる場合に最適です。
2-1. 主な特徴
・導入
クライアントは message/stream RPC メソッドを使用して、初期メッセージ (プロンプトやコマンドなど) を送信し、同時にそのタスクの更新をサブスクライブします。
・サーバ機能
A2Aサーバは、「Agent Card」で capabilities.streaming: true を設定することで、ストリーミングをサポートしていることを示す必要があります。
・サーバ応答 (接続)
サブスクリプションが成功すると、サーバは HTTP 200 OK ステータスと Content-Type: text/event-stream で応答します。このHTTP接続は、サーバがイベントをプッシュできるように開いたままになります。
・イベント構造
サーバはこのストリームを介してイベントを送信します。各イベントのdataフィールドには、JSON-RPC 2.0 レスポンスオブジェクト (SendStreamingMessageResponse) が含まれます。この JSON-RPC レスポンスのIDは、クライアントの元の message/stream リクエストの ID と一致します。
・イベントタイプ（SendStreamingMessageResponse.result 内）
・Task
A2A サーバが A2A クライアントのために処理しているステートフルな作業単位を表します。
・TaskStatusUpdateEvent
タスクのライフサイクル状態の変化 (例 : workingからinput-requiredまたはcompleted) を通知します。また、エージェントからの中間メッセージ (例 : 「現在データを分析中です…」) を提供することもできます。
・TaskArtifactUpdateEvent
タスクによって生成された新規または更新されたAtrifactを配信します。これは、大きなファイルやデータ構造をチャンク単位でストリーミングするために使用されます。このオブジェクト自体には、クライアントが完全なArtifactを再構成できるように、append や lastChunk などのフィールドが含まれています。
・ストリーム終了
サーバは、TaskStatusUpdateEvent で final: true を設定することで、特定のインタラクションサイクル (つまり、現在のmessage/streamリクエスト) の更新の終了を通知します。これは通常、タスクが終了状態 (completed、failed、canceled) または入力必須状態 (サーバがクライアントからのさらなる入力を期待する状態) に達したときに発生します。final: true イベントを送信した後、サーバは通常、その特定のリクエストに対するSSE接続を閉じます。
・再サブスクリプション
タスクがまだアクティブな状態でクライアントのSSE接続が早期に切断された場合 (かつサーバがそのフェーズでfinal: trueイベントを送信していない場合)、クライアントはtasks/resubscribe RPCメソッドを使用してストリームへの再接続を試みることができます。切断期間中に失われたイベントに関するサーバの動作 (例えば、バックフィルを行うか、新しい更新のみを送信するかなど) は実装に依存します。
2-2. ストリーミングを使用する場合
・長時間実行されるタスクのリアルタイムの進捗状況監視。
・大きな結果 (Artifact) を段階的に受信し、結果全体が利用可能になる前に処理を開始できるようにする。
・即時のフィードバックや部分的な応答が有益な、インタラクティブな会話型のやり取り。
・エージェントからの低レイテンシの更新を必要とするアプリケーション。
詳細な構造についてはプロトコル仕様を参照してください。
3. 切断されたシナリオにおけるプッシュ通知
実行時間が非常に長いタスク (数分、数時間、あるいは数日間など) や、クライアントが持続的な接続を維持できない、あるいは維持したくない場合 (モバイルクライアントやサーバレス関数など) のために、「A2A」はプッシュ通知による非同期更新をサポートします。このメカニズムにより、「A2A」サーバは、重要なタスク更新が発生した際に、クライアントが提供する Webhook に通知することができます。
3-1. 主な特徴
・サーバの機能
A2Aサーバは、「Agent Card」で capabilities.pushNotifications: true を設定することで、この機能をサポートしていることを示す必要があります。
・設定
クライアントはサーバに PushNotificationConfig を提供します。
・この設定は、以下の方法で提供できます。
・最初の message/send または message/stream リクエスト内（TaskSendParams のオプションの pushNotification パラメータ経由）。
・既存のタスクに対して、tasks/pushNotificationConfig/set RPC メソッドを使用して別途提供。
・PushNotificationConfig には以下の情報が含まれます。
・url
A2A サーバがタスク更新通知を (POST) 送信する HTTPS Webhook の絶対 URL。
・token (オプション)
クライアントが生成する不透明な文字列 (例 : シークレットまたはタスク固有の識別子)。サーバは、クライアントの Webhook レシーバーによる検証のために、このトークンを通知リクエスト (例 : X-A2A-Notification-Token などのカスタムヘッダー) に含める必要があります。
・authentication (オプション)
A2AサーバがクライアントのWebhook URLに対して認証を行う方法を指定するAuthenticationInfoオブジェクト。クライアント (Webhookの受信者) は、これらの認証要件を定義します。
・通知トリガー
A2Aサーバは、プッシュ通知を送信するタイミングを決定します。通常、これはタスクが重要な状態変化 (終了状態 (completed、failed、canceled、rejected) への遷移、入力必須または認証必須の状態への遷移など) に達したときに発生します。特に、関連するメッセージとアーティファクトが完全に生成され、安定した後に発生します。
・通知ペイロード
A2Aプロトコル自体は、サーバからクライアントのWebhookに送信されるプッシュ通知のHTTPボディペイロードを厳密に定義していません。ただし、通知には、クライアントがタスクIDを識別し、更新の概要 (例 : 新しいタスク状態) を理解するのに十分な情報が含まれている必要があります。サーバは、最小限のペイロード (タスクIDと新しい状態のみ) を送信することも、より包括的なペイロード (例 : 概要、または完全なタスクオブジェクト) を送信することもあります。
・クライアントアクション
プッシュ通知を受信すると (そしてその信頼性と関連性を正常に検証すると)、クライアントは通常、通知のタスク ID を指定した tasks/get RPC メソッドを使用して、新しい成果物や詳細なメッセージを含む、更新された完全な Task オブジェクトを取得します。
3-2. プッシュ通知サービス（クライアント側Webhookインフラストラクチャ）
・PushNotificationConfig.url で指定されたターゲットURLは、プッシュ通知サービスを指します。このサービスは、クライアント側 (またはクライアントがサブスクライブしているサービス) のコンポーネントであり、A2AサーバからのHTTP POST通知の受信を担当します。
・その役割は以下のとおりです。
・受信通知の認証 (つまり、正当なA2Aサーバからのものであることの確認)。
・通知の関連性の検証 (例 : トークンの確認)。
・通知またはそのコンテンツを適切なクライアントアプリケーションロジックまたはシステムへ中継する。
・シンプルなシナリオ (例 : ローカル開発) では、クライアントアプリケーション自体がWebhookエンドポイントを直接公開する場合があります。
・エンタープライズ環境または本番環境では、これは多くの場合、受信Webhookの処理、呼び出し元の認証、メッセージのルーティング (例 : メッセージキュー、内部API、モバイルプッシュ通知ゲートウェイ、またはその他のイベント駆動型システムへのルーティング) を行う、堅牢で安全なサービスです。
4. プッシュ通知のセキュリティに関する考慮事項
プッシュ通知は非同期かつサーバ側から送信されるため、セキュリティは極めて重要です。A2Aサーバ (通知の送信側) とクライアント側のWebhook受信側の両方に責任があります。
4-1. A2Aサーバのセキュリティ (クライアントWebhookへの通知送信時)
・Webhook URL 検証
・サーバは、PushNotificationConfig でクライアントが指定した URL を盲目的に信頼して POST リクエストを送信すべきではありません。悪意のあるクライアントは、内部サービスや無関係なサードパーティシステムを指す URL を提供することで、損害 (Server-Side Request Forgery - SSRF 攻撃) を引き起こしたり、分散型サービス拒否 (DDoS) 攻撃の増幅装置として機能したりする可能性があります。
・緩和策
・ホワイトリスト
可能であれば、Webhook URL の信頼できるドメインまたは IP アドレス範囲のホワイトリストを維持してください。
・所有権検証 / チャレンジレスポンス
実際の通知を送信する前に、サーバは検証手順を実行できます (理想的には実行すべきです)。例えば、提案された Webhook URL に対して、固有の検証トークン (クエリパラメータまたはヘッダーとして) を含む HTTP GET または OPTIONS リクエストを発行できます。Webhook サービスは、所有権と到達可能性を証明するために、適切な応答 (トークンのエコーバックや準備状況の確認など) を行う必要があります。A2A Python サンプルは、シンプルな検証トークンチェックメカニズムを示しています。
・ネットワーク制御
出力ファイアウォールまたはネットワークポリシーを使用して、A2A サーバがアウトバウンドHTTPリクエストを送信できる場所を制限します。
・クライアントの Webhook への認証
・A2A サーバは、PushNotificationConfig.authentication で指定されたスキームに従って、クライアントの Webhook URL に対して認証を行う必要があります。
・サーバ間Webhookの一般的な認証スキームには、以下が含まれます。
・ベアラートークン (OAuth 2.0)
A2A サーバは、クライアントの Webhook を表すオーディエンス/スコープのアクセストークンを取得し (例 : Webhook プロバイダーがサポートしている場合は OAuth 2.0 クライアント資格情報付与フローを使用)、通知 POST リクエストの Authorization: Bearer <token> ヘッダーに含めます。
・API キー
A2A サーバが特定の HTTP ヘッダー (例 : X-Api-Key) に含める、事前共有 API キー。
・HMAC署名
A2Aサーバは、HMACを使用して共有秘密鍵でリクエストペイロード (またはリクエストの一部) に署名し、その署名をヘッダー (例 : X-Hub-Signature) に含めます。Webhook受信者は、この署名を検証します。
・相互TLS (mTLS)
クライアントのWebhookインフラストラクチャでサポートされている場合、A2AサーバはクライアントTLS証明書を提示できます。
4-2. クライアント Webhook レシーバーのセキュリティ (A2A サーバからの通知を受信する場合)
(1) A2Aサーバの認証
・Webhookエンドポイントは、受信した通知リクエストが正規のA2Aサーバから発信されたものであり、偽物ではないことを厳密に検証する必要があります。
・署名/トークンの検証
・JWT (例 : ベアラートークン) を使用する場合は、JWTの署名をA2Aサーバの信頼できる公開鍵 (例 : A2Aサーバが提供するJWKSエンドポイントから取得したもの) と照合して検証してください。また、iss (発行者)、aud (オーディエンス - Webhookを識別する)、iat (発行時刻)、exp (有効期限) などのクレームも検証してください。
・HMAC署名を使用する場合は、共有シークレットを使用して受信したペイロードの署名を再計算し、リクエストヘッダーの署名と比較してください。
・APIキーを使用する場合は、キーが有効で既知のものであることを確認してください。
・PushNotificationConfig.tokenの検証
タスクの通知を設定する際に、PushNotificationConfig に不透明なトークンを指定した場合、Webhook は受信通知にこのトークンが含まれていることを確認する必要があります (例 : X-A2A-Notification-Token のようなカスタムヘッダー)。これにより、通知が特定のクライアントコンテキストとタスクを対象としていることが保証され、認証レイヤーが追加されます。
(2) リプレイ攻撃の防止
・タイムスタンプ
通知には理想的にはタイムスタンプ (例 : JWT の iat - issued at - クレーム、またはカスタムタイムスタンプヘッダー) を含める必要があります。Webhook は、攻撃者が古いキャプチャ済み通知をリプレイするのを防ぐため、古すぎる (例 : 数分以上経過している) 通知を拒否する必要があります。タイムスタンプは、署名付きペイロード (署名を使用している場合)の一部にすることで、整合性を確保する必要があります。
・ノンス/一意の ID
重要な通知の場合は、通知ごとに一意の使い捨て識別子 (ノンスまたはイベント ID) を使用することを検討してください。 Webhook は、通知の重複処理を防ぐため、受信した ID を (妥当な期間) 追跡する必要があります。JWT の jti (JWT ID) クレームはこの目的に使用できます。
(3) 安全な鍵管理とローテーション
・暗号鍵 (HMAC の場合は対称鍵、JWT 署名/mTLS の場合は非対称鍵ペア) を使用する場合は、定期的な鍵ローテーションを含む安全な鍵管理手法を実装してください。
・A2A サーバが署名し、クライアント Webhook が検証する非対称鍵の場合、JWKS (JSON Web Key Set) などのプロトコルを使用することで、サーバは公開鍵 (ローテーション中の新しい鍵を含む) を既知のエンドポイントに公開できます。これにより、クライアント Webhook は署名検証用の正しい公開鍵を動的に取得できるため、よりスムーズな鍵ローテーションが可能になります。
非対称鍵フローの例（JWT + JWKS）
(1) クライアントは、PushNotificationConfig に、authentication.schemes: ["Bearer"] と、必要に応じて JWT の発行者またはオーディエンスを指定します。
(2) A2A サーバは、通知送信時に以下の処理を行います。
・JWT を生成し、秘密鍵で署名します。JWT には、iss (発行者)、aud (オーディエンス - Webhook)、iat (発行時刻)、exp (有効期限)、jti (JWT ID)、taskId などのクレームが含まれます。
・JWT ヘッダー (alg と kid) は、署名アルゴリズムと鍵 ID を示します。
・A2A サーバは、JWKS エンドポイントを介して公開鍵を提供します (このエンドポイントの URL は、Webhook プロバイダーが認識しているか、検出されている可能性があります)。
(3) クライアント Webhook は、通知を受信すると以下の処理を行います。
・Authorization ヘッダーから JWT を抽出します。
・JWT ヘッダー内の kid を検査します。
・A2A サーバの JWKS エンドポイントから対応する公開鍵を取得します (鍵のキャッシュを推奨します)。
・公開鍵を使用して JWT 署名を検証します。
・クレーム（iss、aud、iat、exp、jti）を検証します。
・PushNotificationConfig.token が指定されている場合は、それをチェックします。
