はじめに
AIエージェント同士をつなぐオープン標準 A2A（Agent2Agent）プロトコル を、SDKを使わない生実装と実測ログで調べていくシリーズの第4回です。
- 第1回: A2AでAIエージェント同士はどう会話するのか? — Agent Card と最小エージェントを依存ゼロで作り、curlで話しかける
- 第2回: A2Aエージェントは時間のかかる依頼をどう処理するのか? — Taskライフサイクル・中断と再開・SSE・成果物（Artifact）
- 第3回: A2Aエージェント同士をつなぐと何が起きるのか? — 2体のエージェントをつなぎ、role・ID・中断の伝搬を実測する
- 第4回（本記事）: A2Aのプッシュ通知はどう動くのか? — webhookで結果を受け取る
- 第5回: A2Aの実装同士は相互運用できるのか? — 公式SDKとの挙動差分と0.3系/1.0系の非互換
第3回では、受付エージェントがワーカーへ依頼を取り次ぐ構成を動かしました。受付はワーカーの応答が返るまでHTTP接続を保持して待つ（ブロッキング）方式でした。今回は結果の受け取り方を変えます。受付はワーカーに受領書だけもらって接続を切り、結果はワーカーからの逆向きHTTP POSTで受け取る。ここで言う受領書とは、returnImmediately: true を付けたSendMessageに対してワーカーが即座に返すレスポンスのことで、中身は状態が TASK_STATE_SUBMITTED のTask（taskIdだけが決まっていて、結果はまだ無い）です。これがA2Aのプッシュ通知です。
本記事のゴールは、次の3点を実感を持って言えるようになることです。
- プッシュ通知は、SSE（Server-Sent Events）と同じ形式のイベント（StreamResponse）を、逆向きのHTTP POST（webhook）で届ける仕組みである
- 通知を受け取るクライアント側が自分のHTTPサーバを持つことで成立し、その局面ではクライアントとサーバの役割が反転する
- webhookで届いた通知は「Taskの状態を見に行くきっかけ」として使い、正式な状態はGetTaskで取り直すと、順序・重複・欠落に強くなる
検証環境: macOS / Node.js v25.6.1 / A2Aプロトコル v1.0（仕様。本文とコード中の§番号はこの仕様の節番号です）
前提: 第2回の task-server.mjs と、第3回の reception-server.mjs を同じディレクトリに用意してください。本記事では両方に数十行ずつ追記します。
第3回の構成が抱える課題
第3回で作った構成を改めて示します。利用者（curl）が受付エージェントに依頼し、受付がワーカーへA2Aで取り次ぐ2段構成です。
第3回の受付は、[受付 → ワーカー] のSendMessageを送ったあと、ワーカーの応答（終端または中断）が返るまでHTTP接続を保持して待ちました。その間、[利用者 → 受付] の接続も開いたままです。このブロッキング取り次ぎには、構造的な課題があります。利用者の待ち時間3.9秒の内訳は、ほぼ丸ごと「受付がワーカーの応答を待って接続を保持している時間」です。ホップが増えるほど、待ち時間とHTTP接続の保持と、各ホップのタイムアウト設計が積み上がっていきます。
そこで登場するのが、本シリーズで最後に扱う主要機能、プッシュ通知です。発想を逆にします。受付はワーカーに「受領書だけちょうだい、終わったらこのURLに知らせて」と頼んで接続を切り、結果はワーカーからの逆向きHTTP POSTで受け取るのです。この通知の届け先URLを一般に webhook と呼びます。
プッシュ通知の仕組み
仕組みは3つの要素でできています（A2A仕様の§4.3「Push Notification Objects」にデータ構造が、§4.3.3「Push Notification Payload」に配信の作法が定義されています）。
-
登録: クライアントがSendMessageの
configuration.taskPushNotificationConfigに、届け先URLと認証情報を載せる -
配信: エージェントはタスクに動きがあるたび、そのURLへHTTP POSTを送る。ボディは第2回のSSEで見たものと同じStreamResponse（statusUpdate / artifactUpdate）で、
Content-Type: application/a2a+json。エージェントは、登録時に受け取った認証情報をAuthorizationヘッダに載せて名乗る - 受信側（通知を受け取るクライアント）の義務: 受け取ったら2xxを即返す（義務）。同じ通知が2回以上届いても壊れないように処理する（推奨）。後者が必要なのは、送信側に求められる配信保証が「配信を最低1回は試みる」であって「ちょうど1回届ける」ではないためです。たとえば受信側が返した2xxが途中で失われると、送信側は届いていないと判断して同じ通知を再送します。受信側は「同じ通知がまた来た」ときに成果物の二重登録などが起きない作りにしておく必要があります。仕様はさらに、通知に含まれるtaskIdが自分の待っているタスクのものかを照合することも受信側の義務にしています（本記事の受付は、依頼ごとに発行したトークンで通知と自タスクを対応付けるため、taskIdの照合は省いています。実装都合の簡略化です）
ここで気づくはずです。受け取る側にHTTPサーバが必要なのだと。だからプッシュ通知は「エージェント2体」と相性が良いのです。受付はもともとHTTPサーバなので、webhookの受け口を1つ足すだけで済みます。
ワーカーへの追加実装
第2回の task-server.mjs に、通知を送る側（ワーカー）の実装を足します。断片中のコメントは本記事向けに「追加:」「変更:」の注記を付け直しています。動作に関わるコードは第2回の付録に対する差分そのものです。まずAgent Cardの宣言を2箇所変えます。capabilities.pushNotifications を false から true に（streaming と extendedAgentCard は第2回のまま）、そして能力が変わるので version を 0.1.0 から 0.2.0 に上げます（第1回で説明したとおり、versionが変わるとETagが変わり、クライアントのCardキャッシュが自然に無効化されます）。
version: '0.2.0', // 変更: 0.1.0 → 0.2.0（プッシュ通知対応。ETagも連動して変わる）
// ...
capabilities: { streaming: true, pushNotifications: true, extendedAgentCard: false }, // 変更: pushNotifications のみ false → true
新規タスクのオブジェクトにフィールドを2つ足します。
streams: [],
waiters: [],
pushConfigs: [], // 追加: 登録されたwebhook設定
pushChain: Promise.resolve(), // 追加: 配信を直列化するためのPromise
pushChain について補足します。SSEは1本の接続に順番に書き出す仕組みなので、送った順に届くことが保証されます。一方webhookは通知1件ごとに別々のHTTP POSTなので、送信側が複数の通知を同時に投げると、ネットワークの都合で「WORKINGよりCOMPLETEDが先に届く」といった入れ替わりが起こりえます。つまり仕組みとしての順序保証が無く、順序を守りたければ誰かが自分で守る必要があります。本実装のワーカーは送信側で守ることにし、「前の配信が完了してから次を送る」ように配信を数珠つなぎにしています。その数珠つなぎに使っているのが pushChain で、Promise（「非同期処理の完了」を表すJavaScriptの標準オブジェクト）を次々に連結していく形です。
登録の受け付けと、配信の関数を追加します。
// SendMessageのconfigurationからwebhook登録を受け付ける（A2A仕様 §3.2.2）
function registerPush(task, params) {
const pnc = params?.configuration?.taskPushNotificationConfig;
if (pnc?.url) {
task.pushConfigs.push({ id: crypto.randomUUID(), url: pnc.url, authentication: pnc.authentication });
}
}
// プッシュ通知（A2A仕様 §4.3.3: SSEと同じStreamResponseを、逆向きのHTTP POSTで届ける）
// 配信はタスクごとに直列化する（webhookには本来SSEのような順序保証がないため）
function notifyPush(task, streamResponse) {
for (const cfg of task.pushConfigs ?? []) {
task.pushChain = task.pushChain.then(async () => {
const body = JSON.stringify(streamResponse);
const headers = { 'Content-Type': 'application/a2a+json' };
if (cfg.authentication) headers.Authorization = `${cfg.authentication.scheme} ${cfg.authentication.credentials}`;
wire('サーバ → webhook', `POST ${cfg.url}\n${Object.entries(headers).map(([k, v]) => `${k}: ${v}`).join('\n')}\n\n${body}`);
try {
const res = await fetch(cfg.url, { method: 'POST', headers, body, signal: AbortSignal.timeout(10_000) });
wire('webhook → サーバ', `HTTP ${res.status}`);
} catch (e) {
wire('webhook配信失敗', String(e.message ?? e)); // 学習用実装なので再送はしない（仕様は最低1回の配信試行を要求）
}
});
}
}
あとは配信のきっかけです。broadcast()（SSE配信）の末尾に1行足すだけです。SSEに1件流れるのと同じきっかけ・同じ回数で、webhookにも配信されます。
function broadcast(task, streamResponse) {
for (const s of task.streams) {
// （既存のSSE配信処理）
}
notifyPush(task, streamResponse); // 追加: 同じStreamResponseをwebhookにも配る
}
最後に handleSendMessage の2箇所で登録を受け付けます。新規タスクでは最初の遷移（SUBMITTED）より前に登録します（そうしないと最初の通知が届きません）。
新規タスクの分岐（else 側。tasks.set の直後）:
tasks.set(task.id, task);
registerPush(task, params); // 追加: SUBMITTEDの通知から届くよう、最初の遷移より先に登録
setState(task, 'TASK_STATE_SUBMITTED');
再開の分岐（msg.taskId がある側。履歴に追記した直後）:
task.history.push({ ...msg, contextId: task.contextId });
registerPush(task, params); // 追加: 再開時の追加登録も受け付ける
受付エージェント側の実装
第3回の reception-server.mjs に、webhookの受け口と、環境変数 DELEGATE_MODE=push で切り替わる取り次ぎモードを追加します（全文は巻末の付録に掲載しています。ここでは追加・変更した箇所を順に示します。ログファイル名を環境変数 WIRE_LOG で変えられるようにした程度の細かな差は付録で確認してください）。
まず、起動時の環境変数でモードを切り替えられるようにします（既定は第3回と同じblocking）。
// 取り次ぎ方式（第4回で追加）:
// blocking = ワーカーの応答（終端または中断）までHTTP接続を保持して待つ（第3回の方式）
// push = returnImmediatelyで受領書だけもらって接続を閉じ、結果はワーカーからのwebhookで受け取る
const DELEGATE_MODE = process.env.DELEGATE_MODE ?? 'blocking';
pushモードでは、受付が依頼ごとにトークンを発行してワーカーに渡し、届いたwebhookの差出人確認と「どの受付タスクの通知か」の対応付けに使います。その台帳です。
// pushモード: 受付が発行したトークン → 対応する受付タスク（webhookの差出人確認と対応付けに使う）
const pendingByToken = new Map();
settleFromWorker() の冒頭には、同じ通知が2回以上届いても壊れないようにするガードを足します（受付側のタスクがもう終端なら何もしない。中断中に同じ中断通知がまた来ても何もしない）。
// webhookは重複しうるので冪等に処理する（A2A仕様 §4.3.3 SHOULD）
if (TERMINAL.has(task.status.state)) return;
if (task.status.state === 'TASK_STATE_INPUT_REQUIRED' && workerTask.status.state === 'TASK_STATE_INPUT_REQUIRED') return;
delegate()（新規依頼の取り次ぎ）には、pushモードの分岐を追加します。SendMessageの configuration にwebhookの登録（届け先URLと、受付が発行したトークン）と returnImmediately: true を載せてワーカーに送り、受領書が返ったら関数を抜けます。以降の状態遷移はwebhookが運んできます。
if (DELEGATE_MODE === 'push') {
// pushモード: webhookを登録し、受領書だけもらって接続を閉じる。結果はwebhookが運んでくる
const token = crypto.randomUUID();
pendingByToken.set(token, task);
const out = await callWorker('SendMessage', {
message: { messageId: crypto.randomUUID(), role: 'ROLE_USER', parts: [{ text }] },
configuration: {
returnImmediately: true,
taskPushNotificationConfig: {
url: `${BASE}/webhook/worker`,
authentication: { scheme: 'Bearer', credentials: token }, // 受付が発行。webhookの差出人確認に使う
},
},
});
if (out.error) {
pendingByToken.delete(token);
return setState(task, 'TASK_STATE_FAILED', `計算担当がエラーを返しました: ${out.error.code} ${out.error.message}`);
}
task.workerTaskId = out.result.task.id;
return; // 以降の状態遷移は webhookハンドラ → settleFromWorker が引き継ぐ
}
forwardAnswer()（中断からの再開）にも同様の分岐を追加します。webhook登録は初回依頼のものがワーカー側タスクに残っているので、再登録は不要です。
if (DELEGATE_MODE === 'push') {
// webhook登録は初回依頼時のものがワーカー側タスクに生きているので、再登録は不要
const out = await callWorker('SendMessage', {
message: { messageId: crypto.randomUUID(), taskId: task.workerTaskId, role: 'ROLE_USER', parts: [{ text }] },
configuration: { returnImmediately: true },
});
if (out.error) return setState(task, 'TASK_STATE_FAILED', `計算担当がエラーを返しました: ${out.error.code} ${out.error.message}`);
return;
}
handleSendMessage() の末尾、利用者への応答を返す前の待ち方も変わります。blockingモードでは取り次ぎ（work）が終わった時点で自タスクは決まっていますが、pushモードでは work は受領書で戻ってくるので、webhookで自タスクが終端か中断に達するまで別途待ちます。
// A2A仕様 §3.2.2: 既定はブロッキング。
// blockingモードでは work の完了時点でタスクは終端または中断に達している。pushモードでは work は受領書で戻るため、終端または中断の通知（webhook）まで別途待つ
if (params?.configuration?.returnImmediately) {
void work; // 受領書を先に返し、取り次ぎは裏で続ける
} else {
await work;
await waitForSettled(task);
}
return { result: { task: taskView(task) } };
最後に、webhookの受け口です。リクエスト処理の先頭（Agent CardやJSON-RPCの分岐より前）に追加します。トークンを照合し、2xxを先に返してから、終端か中断の通知ならGetTaskで正式な状態を取り直して自タスクに反映します。
// ---- webhook受け口（pushモード）: ワーカーからのStreamResponseを受け取る ----
if (req.method === 'POST' && req.url === '/webhook/worker') {
const auth = req.headers['authorization'] ?? '';
const token = auth.startsWith('Bearer ') ? auth.slice('Bearer '.length) : '';
const task = pendingByToken.get(token);
if (!task) {
// 登録した覚えのないトークン = 差出人を確認できない通知は受け取らない
return reply(401, { 'Content-Type': 'application/json' }, JSON.stringify({ error: 'unknown or missing token' }));
}
let ev;
try {
ev = JSON.parse(body);
} catch {
return reply(400, { 'Content-Type': 'application/json' }, JSON.stringify({ error: 'invalid JSON' }));
}
reply(200, { 'Content-Type': 'application/json' }, '{}'); // 受領の2xxは先に返す義務（A2A仕様 §4.3.3）
// 通知は呼び鈴として使い、正式な状態はGetTaskで取り直す（webhookの順序・重複・欠落に依存しない）
const state = ev.statusUpdate?.status?.state;
if (state && (TERMINAL.has(state) || state === 'TASK_STATE_INPUT_REQUIRED')) {
const out = await callWorker('GetTask', { id: ev.statusUpdate.taskId });
if (!out.error) settleFromWorker(task, out.result); // GetTaskのresultはTaskそのもの
if (TERMINAL.has(task.status.state)) pendingByToken.delete(token); // INPUT_REQUIREDなら再開に備えて残す
}
return;
}
最後に通信ログ（wire log）の記録です。webhook経路のリクエストと応答には別の方向ラベルを付け、Authorization ヘッダも記録対象に加えます。ステップ2で見るログの見え方に関わる変更です。
const isWebhook = req.url?.startsWith('/webhook/');
const headerDump = ['host', 'content-type', 'a2a-version', 'if-none-match', 'authorization']
.filter((h) => req.headers[h] !== undefined)
.map((h) => `${h}: ${req.headers[h]}`)
.join('\n');
wire(isWebhook ? 'ワーカー → 受付 (webhook)' : '利用者 → 受付', `${req.method} ${req.url}\n${headerDump}${body ? '\n\n' + body : ''}`);
const reply = (status, headers, payload) => {
wire(isWebhook ? '受付 → ワーカー (webhook応答)' : '受付 → 利用者', `HTTP ${status}\n${Object.entries(headers).map(([k, v]) => `${k}: ${v}`).join('\n')}${payload ? '\n\n' + payload : ''}`);
以上をまとめると、pushモードの受付は次のように動きます。
- 依頼ごとにトークンを発行し、SendMessageの
configuration.taskPushNotificationConfigに受付のwebhook URL（/webhook/worker）とそのトークンを載せて（これが「仕組み」で述べた登録です）、returnImmediately: trueでワーカーに送る - ワーカーから受領書（SUBMITTED状態のTaskを含むSendMessageのレスポンス）が返ってきたら、その時点で [受付 → ワーカー] のHTTP接続を閉じる。ワーカーの作業完了は待たない
- 届いた通知のトークンを照合する（発行した覚えのないものは401）
- 終端（
TASK_STATE_COMPLETED/TASK_STATE_FAILED/TASK_STATE_CANCELED/TASK_STATE_REJECTED）か中断（TASK_STATE_INPUT_REQUIRED）の通知が来たら、GetTaskで正式な状態を取り直してから自タスクに反映する
4の考え方を、本記事では「呼び鈴パターン」と呼ぶことにします（一般用語ではなく本シリーズでの呼び名です）。webhookには順序保証がなく、重複も欠落もありえます。届いた通知の中身を直接信じて状態を組み立てるのではなく、webhookで届いた通知は「Taskの状態を見に行くきっかけ」としてだけ使い、正式な状態は必ずGetTaskで取り直す。こうすると順序・重複・欠落のどれが起きても、ワーカー上の本物のTaskに収束します。第2回の「状態の正はTaskにある」が、ここでも効いています。
通信の中身を確認する
curlで通信の中身を確認していきます。手順は次の5ステップです。
- pushモードで起動して、同じ依頼を送る
- 通信ログで登録とwebhookを見る
- 通信の流れ全体を時系列で見る
- 偽のwebhookを投げ込んでみる
- 自分がwebhookの受け口になってみる（おすすめの実験）
後続のcurlコマンドで繰り返し使う値（受付とワーカーのURL、共通のHTTPヘッダ）を、あらかじめシェル変数に入れておきます。
R=http://localhost:4103
W=http://localhost:4102
H1='Content-Type: application/json'
H2='A2A-Version: 1.0'
ステップ1: pushモードで起動して、同じ依頼を送る
ワーカーを再起動し、受付を DELEGATE_MODE=push で起動し直します（第3回と同じく、それぞれ別のターミナルで）。
node task-server.mjs # ワーカー（プッシュ通知対応版）
DELEGATE_MODE=push node reception-server.mjs # 受付（pushモード）
まず、ワーカーのCardが変わったことを確認します。
curl -s $W/.well-known/agent-card.json | grep -E '"version"|"pushNotifications"'
"version": "0.2.0" と "pushNotifications": true の2行が出るはずです。確認できたら、第3回のステップ2と全く同じ依頼を送ります。
time curl -s -X POST $R/a2a/v1 -H "$H1" -H "$H2" \
-d '{"jsonrpc":"2.0","id":1,"method":"SendMessage","params":{"message":{"messageId":"u-1","role":"ROLE_USER","parts":[{"text":"{\"a\": 19, \"b\": 23}"}]}}}'
応答の要点はこうです（時間は time で計測）。
状態: TASK_STATE_COMPLETED / 成果物: 19 + 23 = 42 （実測 3.994秒）
利用者から見える応答も、待ち時間（第3回のブロッキング取り次ぎ3.862秒 / 今回3.994秒）もほぼ同じです。受付の内部の待ち方が変わったことは、利用者には見えません。エージェントの不透明さ（opaque）は、こういうところにも及びます。
なお、受付のAgent Cardは pushNotifications: false のままです。この宣言は「自分のクライアントに向けてプッシュ通知を送れるか」を表すもので、受付が（ワーカーのクライアントとして）webhookを受け取れるかどうかとは別の話です。サーバ役とクライアント役のうち、Cardに現れるのはサーバとしての顔だけです。
ステップ2: 通信ログで登録とwebhookを見る
reception-wire.log を見ると、まず [受付 → ワーカー] のSendMessageリクエストの中に、プッシュ通知の登録が含まれています。登録のための専用リクエストを別に送るのではなく、依頼と同じ1つのリクエストの configuration.taskPushNotificationConfig に、届け先URLと認証情報を書いて送っているのが分かります。
"configuration": {
"returnImmediately": true,
"taskPushNotificationConfig": {
"url": "http://localhost:4103/webhook/worker",
"authentication": {
"scheme": "Bearer",
"credentials": "e3925beb-ce25-4406-a8f4-251e8284741a"
}
}
}
credentials の値は受付がこの依頼のために発行したトークンです。そして少し後に、[ワーカー → 受付 (webhook)] の逆向きPOSTが届いています。
POST /webhook/worker
host: localhost:4103
content-type: application/a2a+json
authorization: Bearer e3925beb-ce25-4406-a8f4-251e8284741a
{"statusUpdate":{"taskId":"ab6c498e-...","contextId":"3006a70b-...","status":{"state":"TASK_STATE_SUBMITTED","timestamp":"2026-08-27T08:33:02.643Z"}}}
登録時に渡したトークンが、Authorization ヘッダでそのまま返ってきています。ワーカーはトークンの意味を知らず、「通知のときはこれを名乗れ」と言われたものを載せているだけです。普段のAPI認証と資格情報（クレデンシャル）の向きが逆で、これから通知を受け取るクライアント（受付）が発行した資格情報を、これから通知を送るサーバ（ワーカー）が提示する形になっています。webhookの局面ではクライアントとサーバの役割が反転するので、認証の向きも一緒に反転するのです。
届いたwebhookを時系列に並べると、第2回のSSEと同じイベント列です。
statusUpdate : TASK_STATE_SUBMITTED
statusUpdate : TASK_STATE_WORKING
artifactUpdate: 19 + 23 = 42
statusUpdate : TASK_STATE_COMPLETED
配信手段がSSE（順方向のストリーム）からwebhook（逆向きのPOST）に変わっても、イベントの形式は同じ。これが本記事のいちばん大事な観察です。
ステップ3: 通信の流れ全体を時系列で見る
reception-wire.log のエントリの方向ラベルだけを時系列に並べます。
08:33:02.592 [利用者 → 受付] 依頼
08:33:02.599 [受付 → ワーカー] Agent Card取得（初回のみ）
08:33:02.634 [受付 → ワーカー] SendMessage（returnImmediately + webhook登録）
08:33:02.653 [ワーカー → 受付] 受領書（ここで接続は閉じる。往復19ミリ秒）
08:33:02.664 [ワーカー → 受付 (webhook)] SUBMITTED
08:33:03.450 [ワーカー → 受付 (webhook)] WORKING
08:33:06.458 [ワーカー → 受付 (webhook)] artifactUpdate
08:33:06.486 [ワーカー → 受付 (webhook)] COMPLETED
08:33:06.503 [受付 → ワーカー] GetTask（呼び鈴を聞いて正式な状態を取りに行く）
08:33:06.534 [受付 → 利用者] COMPLETEDのTask
（Agent Card取得への応答・webhookへの2xx応答・GetTaskへの応答は省略しています）
第3回のブロッキング取り次ぎでは受付⇔ワーカーの接続が3.8秒間張りっぱなしでしたが、pushモードでは受領書の往復19ミリ秒で閉じ、結果は約3.9秒かけて逆向きに届いています。利用者の合計待ち時間は変わりません。変わったのは受付⇔ワーカー間の接続保持時間で、両者の間のタイムアウト設計から作業時間そのものが消えました。中断（INPUT_REQUIRED）の伝搬も同じ経路で動きます（利用者からの見え方は第3回と同一です）。
ステップ4: 偽のwebhookを投げ込んでみる
webhookの受け口は、ネットワークに開いたただのHTTPエンドポイントです。ポート4103に届く者なら誰でもPOSTできます。そこで、ワーカーになりすまして偽の完了通知を投げ込んでみます。
curl -s -i -X POST $R/webhook/worker -H 'Content-Type: application/a2a+json' \
-H 'Authorization: Bearer fake-token' \
-d '{"statusUpdate":{"taskId":"x","status":{"state":"TASK_STATE_COMPLETED"}}}'
HTTP/1.1 401 Unauthorized
受付は「自分が発行した覚えのないトークン」なので門前払いします。これが1段目の防御です。
仮にトークンが漏れてこの照合を突破されたとしても、受付は通知に書かれた state を読まず、通知の taskId をもとにワーカーへGetTaskして本物の状態を確認します。偽のtaskIdなら、ワーカーが「そのTaskは無い」（-32001 TaskNotFoundError）を返すので受付は何もしません。本物のtaskIdを書かれても、ワーカーが返すのは実際の状態（この時点ならWORKING）なので、偽の完了は成立しません。これが2段目、呼び鈴パターンによる防御です。トークン照合と呼び鈴パターンの二段防御です。
ただし、この受付は「通知のtaskIdが自分の待っているワーカー側タスクと一致するか」を照合していません（先に述べた、仕様のMUSTを省いた箇所です）。正しいトークンを持つ相手が、同じワーカー上の別の完了済みタスクのtaskIdを書いて送ると、受付はそのタスクの結果を自タスクに反映してしまいます。トークンが漏れない前提で成り立っている簡略化であり、仕様が「Clients MUST validate the task ID matches an expected task」（§4.3.3）と義務にしている照合なので、実運用の実装では省かずに加えるべき箇所です。
ステップ5（おすすめの実験）: 自分がwebhookの受け口になってみる
ここまでwebhookは受付が受けていましたが、最後にあなた自身が受けてみましょう。ターミナルAで待ち受けます（nc はポートで待ち受けて届いたバイト列をそのまま表示するコマンドです）。
nc -l 5555
ターミナルBから、ワーカーに直接、あなたのnc宛のwebhook登録付きで依頼を送ります。
リクエストのJSONは長いので、ヒアドキュメントで整形したまま -d @-（標準入力から読む指定）で渡します。
curl -s -X POST $W/a2a/v1 -H "$H1" -H "$H2" -d @- <<'EOF'
{
"jsonrpc": "2.0", "id": 1, "method": "SendMessage",
"params": {
"message": {
"messageId": "w-1", "role": "ROLE_USER",
"parts": [{ "text": "{\"a\": 19, \"b\": 23}" }]
},
"configuration": {
"returnImmediately": true,
"taskPushNotificationConfig": {
"url": "http://localhost:5555/hook",
"authentication": { "scheme": "Bearer", "credentials": "my-secret" }
}
}
}
}
EOF
直後、ターミナルAに生のHTTP POSTがそのまま現れます。
POST /hook HTTP/1.1
host: localhost:5555
Content-Type: application/a2a+json
Authorization: Bearer my-secret
content-length: 198
{"statusUpdate":{"taskId":"d53a17d7-...","contextId":"73ff3c67-...","status":{"state":"TASK_STATE_SUBMITTED","timestamp":"2026-08-27T07:43:10.839Z"}}}
（一部のヘッダは省略しています）
ずっとcurlで「送る側」だった通信を、初めて「受ける側」から見た瞬間です。なお nc は応答を返さないので、ワーカー側は約10秒（AbortSignal.timeout）待ってから配信失敗と判断し、接続を切ります。その時点で nc は終了し、後続のイベントはワーカー側の wire.log に「webhook配信失敗」として記録されます（配信は直列化されているため、記録が出るまで10秒以上かかります）。これはこれで、「webhookを受け取る側が2xxを返さないと配信は失敗扱いになる」という仕様の動きの観察になっています。
実測して初めて分かったこと
1. プッシュ通知は「逆向きリクエスト」ではなく「役割の反転」
A2AにはMCPのelicitationのような「サーバがクライアントを呼び返す」仕組みがありません。プッシュ通知はその代替に見えますが、実体は「通知を受け取るクライアント（依頼を出した側）が自分のHTTPサーバを持ち、その局面だけクライアントとサーバの役割が入れ替わる」ことです。認証の向き（クライアントが発行したトークンをサーバが名乗る）も、この反転の帰結です。
2. 配信手段は変わってもイベントの形式は変わらない
webhookで届いたのは、SSEと同じStreamResponse（今回は statusUpdate と artifactUpdate の2種が計4件）でした。これで更新の受け取り方が3方式（ポーリング / SSE / webhook）出揃いましたが、どれもTaskを正とし、同じ形式のイベント（StreamResponse）を運ぶだけです。ネットワーク要件だけが違います。
| 方式 | 更新を受け取るクライアント側に必要なもの |
|---|---|
| ポーリング（GetTask） | なし（送信だけで済む） |
| SSE | なし（ただし接続の保持が必要） |
| プッシュ通知（webhook） | 相手から届くHTTPを受けられる公開エンドポイント |
組織間・国際間の接続では、外からのHTTPを受ける口（インバウンド）をファイアウォールで開けられないことが多く、その場合は「長時間タスクでもSSE+切断時の再購読」が現実解になります。プロトコルが3方式を用意しているのは、この使い分けの余地を残すためです。
3. 通知の中身は信じず、状態はGetTaskで取り直す
webhookには順序保証がなく、重複も欠落も起こりえます。受付はwebhookで届いた通知を「Taskの状態を見に行くきっかけ」としてだけ使い、正式な状態はGetTaskで取り直す作りにしました（本記事で言う呼び鈴パターン）。こうすると順序・重複・欠落・なりすましのどれが起きても、ワーカー上の本物のTaskに収束します。第2回で確認した「状態の正はTaskにある」が、通知の設計でも効いています。
MCPとの対比
比較の前提として、本シリーズが対象にしてきたMCPは2025-06-18版で、HTTPで使う場合のトランスポートはStreamable HTTPです。
| 観点 | A2A（v1.0） | MCP（2025-06-18版） |
|---|---|---|
| サーバからクライアントへのメッセージ送信 | できる。Taskの状態更新や成果物をSSEまたはwebhookで送る | できる。SSEストリームに通知やリクエストを流す |
| 追加入力（聞き返し）の伝え方 | Taskの状態を INPUT_REQUIRED にし、その状態更新をSSE / webhook / GetTaskで知らせる。続きはクライアントが新しいSendMessage（taskId付き）として送る。通知に応答は要求されず、入力の形式を指定する仕組みも無い |
elicitation（利用者への入力要求）を、id付きのJSON-RPCリクエストとして送る。求める入力の形式をJSON Schemaで指定し、クライアントはそのidへの応答（accept / decline / cancel と入力内容）を返す |
| サーバからの送信が届く条件 | SSEはPOST応答が開いている間。webhookはクライアントがHTTPサーバを持っていれば、接続が切れた後でも届く | セッションのSSE接続が開いている間 |
まず押さえたいのは、サーバからクライアントへメッセージを送ること自体は、どちらのプロトコルでもできるという点です。A2Aは本記事のwebhookや第2回のSSEで状態更新を送りますし、MCPもSSEストリームにサーバ発のメッセージを流せます。違いは2つあります。
1つ目は、聞き返しの形です。A2Aでも、Taskが INPUT_REQUIRED になったことをwebhookやSSEでクライアントに知らせられるので、「サーバが入力を求める」こと自体は機能として可能です。違いはその表し方にあります。MCPのelicitationは、id付きのリクエストとそのidへのレスポンスという1対の通信で、元の tools/call はその応答を待って止まります。求める入力の形式もJSON Schemaで指定します。A2Aでは INPUT_REQUIRED は状態であり、それを知らせる通知に応答は要求されません。続きはクライアントが別の新しい依頼（taskId付きSendMessage）として送り、両者はtaskIdで結び付きます。求める入力の形式を指定する仕組みも無く、状態に添えたMessageの文章で伝えるだけです。第3回で見た中断の伝搬が「状態を見て自分の状態を決める」だけで成立したのは、聞き返しがリクエストではなく状態だからです。
2つ目は、届く条件です。MCPのサーバ発メッセージはセッションの接続の上を流れるので、接続が切れると届きません。A2Aのwebhookは、クライアントが自分のHTTPサーバを持つことで、元の接続が閉じた後でもサーバから届く経路を作ります。「接続の中で逆向きに送る」か「接続の外に逆向きの経路を持つ」かの違いです。
まとめ
- プッシュ通知は、SSEと同じ形式のイベント（statusUpdate / artifactUpdate）を逆向きのHTTP POSTで届ける仕組み。登録は専用のリクエストではなく、依頼のSendMessageの
configuration.taskPushNotificationConfigに書いて一緒に送る - 通知を受け取る側（クライアント）も自分のHTTPサーバを持つ必要がある。webhookの局面ではクライアントとサーバの役割が反転し、認証の向き（クライアントが発行した資格情報をサーバが名乗る）も反転する
- webhookを受け取る側（本記事では受付）の作法は、2xxの即答（義務）・重複しても壊れない処理（推奨）・差出人確認。webhookで届いた通知はTaskの状態を見に行くきっかけとして使い、正式な状態はGetTaskで取り直す（呼び鈴パターン）と、順序・重複・欠落・なりすましに強くなる
- 利用者から見える応答も待ち時間も変わらない。変わるのはホップ間の接続保持時間で、タイムアウト設計から作業時間が消える
- 更新の受け取りは3方式（ポーリング / SSE / webhook）が出揃った。違いはイベントの形式ではなくネットワーク要件（webhookだけはクライアント側にインバウンドの到達性が要る）
次回（最終回）: ここまで自作実装だけで進めてきましたが、「他人の実装」と本当に会話できるのでしょうか。公式SDK（@a2a-js/sdk）でサーバとクライアントを立てて自作実装と突き合わせ、挙動の差分を実測します。仕様のエラーコード表に従わない公式実装、0.3系と1.0系がメソッド名の時点で通じ合えない現実など、相互運用の生々しい景色をお見せします。
付録: reception-server.mjs の全文（プッシュ通知対応版）
本文の体験で使用した受付エージェントの全文です。第3回の版に、webhookの受け口と DELEGATE_MODE=push の取り次ぎモードを追加したものです。
// A2A v1.0 エージェント間連携の学習用「受付エージェント」（依存ゼロNode, JSON-RPCバインディング）
//
// 役割: 計算依頼の受付窓口。自分では計算せず、Slow Calc Agent（第2回の task-server.mjs。同じディレクトリに置く）に
// A2Aで依頼を転送する。つまり「A2Aサーバでありながら、別のエージェントに対してはA2Aクライアント」。
//
// 第3回の観察ポイント:
// - エージェントが送る側に回っても、Messageの role は ROLE_USER（Roleは会話の立場であって人間/AIの区別ではない）
// - taskId / contextId はホップごとに独立（利用者⇔受付のTaskと、受付⇔ワーカーのTaskは別物）
// - INPUT_REQUIRED は伝搬する: ワーカーの質問を受付が自タスクの中断として利用者へ中継し、
// 利用者の回答を受付がワーカーのタスク再開として転送する
// - 受付が送るMessageのmessageIdは受付が発番する（利用者のmessageIdを使い回さない。発番者=作成者）
//
// 対応メソッド: SendMessage / GetTask（ストリーミング非対応 = Cardの capabilities.streaming: false）
//
// 取り次ぎ方式は DELEGATE_MODE 環境変数で切り替え（第4回で追加）:
// blocking（既定） = ワーカーの応答（終端または中断）まで接続を保持（第3回の方式）
// push = webhookを登録して接続を閉じ、結果はワーカーからの逆向きHTTP POSTで受ける
// どちらのモードでも利用者から見える応答は同じ（内部の待ち方は利用者に対して不透明）
import http from 'node:http';
import crypto from 'node:crypto';
import fs from 'node:fs';
import { fileURLToPath } from 'node:url';
const PORT = 4103;
const BASE = `http://localhost:${PORT}`;
const WORKER_BASE = process.env.WORKER_BASE ?? 'http://localhost:4102';
const WIRE_LOG = process.env.WIRE_LOG ?? fileURLToPath(new URL('./reception-wire.log', import.meta.url));
// 取り次ぎ方式（第4回で追加）:
// blocking = ワーカーの応答（終端または中断）までHTTP接続を保持して待つ（第3回の方式）
// push = returnImmediatelyで受領書だけもらって接続を閉じ、結果はワーカーからのwebhookで受け取る
const DELEGATE_MODE = process.env.DELEGATE_MODE ?? 'blocking';
const AGENT_CARD = {
name: 'Calc Reception Agent',
description: '計算依頼の受付窓口。依頼内容を確認し、計算はSlow Calc AgentにA2Aで取り次ぐ（エージェント間連携の学習用）',
version: '0.1.0',
supportedInterfaces: [
{ url: `${BASE}/a2a/v1`, protocolBinding: 'JSONRPC', protocolVersion: '1.0' },
],
capabilities: { streaming: false, pushNotifications: false, extendedAgentCard: false },
defaultInputModes: ['application/json', 'text/plain'],
defaultOutputModes: ['text/plain'],
skills: [
{
id: 'delegate-add',
name: '足し算の取り次ぎ',
description: '足し算の依頼を受け付け、計算担当のエージェントに取り次いで結果を返す',
tags: ['calculator', 'delegation'],
examples: ['{"a": 19, "b": 23}', '19 + 23'],
inputModes: ['application/json', 'text/plain'],
outputModes: ['text/plain'],
},
],
};
const CARD_ETAG = `"${AGENT_CARD.version}"`;
const PARSE_ERROR = -32700;
const INVALID_REQUEST = -32600;
const INVALID_PARAMS = -32602;
const METHOD_NOT_FOUND = -32601;
const TASK_NOT_FOUND = -32001; // A2A仕様 §5.4 TaskNotFoundError
const UNSUPPORTED_OPERATION = -32004; // A2A仕様 §5.4 UnsupportedOperationError
const VERSION_NOT_SUPPORTED = -32009; // A2A仕様 §5.4 VersionNotSupportedError
function wire(direction, text) {
fs.appendFileSync(WIRE_LOG, `[${new Date().toISOString()}] [${direction}]\n${text}\n\n`);
}
// ---- ワーカーの発見と呼び出し（ここが「A2Aクライアント」側の実装） ----
let workerEndpoint; // Agent Cardから得たRPCエンドポイント（発見は初回のみ、以後キャッシュ）
async function getWorkerEndpoint() {
if (workerEndpoint) return workerEndpoint;
const url = `${WORKER_BASE}/.well-known/agent-card.json`;
wire('受付 → ワーカー', `GET ${url}`);
const res = await fetch(url);
const card = await res.json();
wire('ワーカー → 受付', `HTTP ${res.status}\n\n${JSON.stringify(card, null, 2)}`);
workerEndpoint = card.supportedInterfaces[0].url; // 発見 → 接続の2段階（第1回と同じ流儀）
return workerEndpoint;
}
let outboundRpcId = 1; // 受付→ワーカーのJSON-RPC idは受付が発番する（利用者側のidとは別の名前空間）
async function callWorker(method, params) {
const endpoint = await getWorkerEndpoint();
const payload = { jsonrpc: '2.0', id: outboundRpcId++, method, params };
const body = JSON.stringify(payload, null, 2);
wire('受付 → ワーカー', `POST ${endpoint}\ncontent-type: application/json\na2a-version: 1.0\n\n${body}`);
const res = await fetch(endpoint, {
method: 'POST',
headers: { 'Content-Type': 'application/json', 'A2A-Version': '1.0' },
body,
});
const text = await res.text();
wire('ワーカー → 受付', `HTTP ${res.status}\n\n${text}`);
return JSON.parse(text);
}
// ---- 受付自身のTask管理（task-server.mjsの簡略版。ストリーミングなし） ----
// task = { id, contextId, status, artifacts, history, workerTaskId, waiters }
const tasks = new Map();
const TERMINAL = new Set(['TASK_STATE_COMPLETED', 'TASK_STATE_FAILED', 'TASK_STATE_CANCELED', 'TASK_STATE_REJECTED']);
const INTERRUPTED = new Set(['TASK_STATE_INPUT_REQUIRED', 'TASK_STATE_AUTH_REQUIRED']);
function agentMessage(text, taskId, contextId) {
return { messageId: crypto.randomUUID(), taskId, contextId, role: 'ROLE_AGENT', parts: [{ text }] };
}
function setState(task, state, noteText) {
const message = noteText ? agentMessage(noteText, task.id, task.contextId) : undefined;
task.status = { state, timestamp: new Date().toISOString(), ...(message ? { message } : {}) };
if (message) task.history.push(message);
if (TERMINAL.has(state) || INTERRUPTED.has(state)) {
task.waiters.splice(0).forEach((resolve) => resolve());
}
}
function taskView(task, historyLength) {
const view = {
id: task.id,
contextId: task.contextId,
status: task.status,
artifacts: task.artifacts,
// 取り次ぎ先の対応付けは（プロトコル規定外の情報なので）metadataで公開する
...(task.workerTaskId ? { metadata: { workerTaskId: task.workerTaskId } } : {}),
};
if (historyLength === 0) return view;
const h = historyLength > 0 ? task.history.slice(-historyLength) : task.history;
return { ...view, history: h };
}
function waitForSettled(task) {
if (TERMINAL.has(task.status.state) || INTERRUPTED.has(task.status.state)) return Promise.resolve();
return new Promise((resolve) => task.waiters.push(resolve));
}
// ---- 取り次ぎ本体 ----
// pushモード: 受付が発行したトークン → 対応する受付タスク（webhookの差出人確認と対応付けに使う）
const pendingByToken = new Map();
// ワーカーから返ってきたタスクの状態を、受付自身のタスクに反映する
function settleFromWorker(task, workerTask) {
// webhookは重複しうるので冪等に処理する（A2A仕様 §4.3.3 SHOULD）
if (TERMINAL.has(task.status.state)) return;
if (task.status.state === 'TASK_STATE_INPUT_REQUIRED' && workerTask.status.state === 'TASK_STATE_INPUT_REQUIRED') return;
task.workerTaskId = workerTask.id;
const state = workerTask.status.state;
if (state === 'TASK_STATE_COMPLETED') {
for (const a of workerTask.artifacts ?? []) {
task.artifacts.push({
artifactId: crypto.randomUUID(), // 受付の成果物として発番し直す（出所はmetadataに残す）
name: a.name,
parts: a.parts,
metadata: { source: 'Slow Calc Agent', workerTaskId: workerTask.id, workerArtifactId: a.artifactId },
});
}
setState(task, 'TASK_STATE_COMPLETED', '計算担当から結果を受け取りました');
} else if (state === 'TASK_STATE_INPUT_REQUIRED') {
const question = workerTask.status.message?.parts?.map((p) => p.text ?? '').join('') ?? '（質問文なし）';
setState(task, 'TASK_STATE_INPUT_REQUIRED', `計算担当からの質問です: ${question}`);
} else if (state === 'TASK_STATE_REJECTED') {
const reason = workerTask.status.message?.parts?.map((p) => p.text ?? '').join('') ?? '';
setState(task, 'TASK_STATE_REJECTED', `計算担当に断られました: ${reason}`);
} else {
setState(task, 'TASK_STATE_FAILED', `計算担当のタスクが ${state} になりました`);
}
}
// 新規依頼をワーカーへ転送する。Messageは受付が新規に作る（messageIdも受付が発番）
async function delegate(task, text) {
try {
setState(task, 'TASK_STATE_WORKING', '依頼を計算担当（Slow Calc Agent）に取り次ぎます');
if (DELEGATE_MODE === 'push') {
// pushモード: webhookを登録し、受領書だけもらって接続を閉じる。結果はwebhookが運んでくる
const token = crypto.randomUUID();
pendingByToken.set(token, task);
const out = await callWorker('SendMessage', {
message: { messageId: crypto.randomUUID(), role: 'ROLE_USER', parts: [{ text }] },
configuration: {
returnImmediately: true,
taskPushNotificationConfig: {
url: `${BASE}/webhook/worker`,
authentication: { scheme: 'Bearer', credentials: token }, // 受付が発行。webhookの差出人確認に使う
},
},
});
if (out.error) {
pendingByToken.delete(token);
return setState(task, 'TASK_STATE_FAILED', `計算担当がエラーを返しました: ${out.error.code} ${out.error.message}`);
}
task.workerTaskId = out.result.task.id;
return; // 以降の状態遷移は webhookハンドラ → settleFromWorker が引き継ぐ
}
// blockingモード（第3回と同じ）: ワーカーの応答（終端または中断）までHTTP接続を保持して待つ
const out = await callWorker('SendMessage', {
message: { messageId: crypto.randomUUID(), role: 'ROLE_USER', parts: [{ text }] },
});
if (out.error) return setState(task, 'TASK_STATE_FAILED', `計算担当がエラーを返しました: ${out.error.code} ${out.error.message}`);
settleFromWorker(task, out.result.task);
} catch (e) {
setState(task, 'TASK_STATE_FAILED', `計算担当に接続できません: ${e.message}`);
}
}
// 利用者の回答を、ワーカー側タスクの再開（taskId付きSendMessage）として転送する
async function forwardAnswer(task, text) {
try {
if (DELEGATE_MODE === 'push') {
// webhook登録は初回依頼時のものがワーカー側タスクに生きているので、再登録は不要
const out = await callWorker('SendMessage', {
message: { messageId: crypto.randomUUID(), taskId: task.workerTaskId, role: 'ROLE_USER', parts: [{ text }] },
configuration: { returnImmediately: true },
});
if (out.error) return setState(task, 'TASK_STATE_FAILED', `計算担当がエラーを返しました: ${out.error.code} ${out.error.message}`);
return;
}
const out = await callWorker('SendMessage', {
message: { messageId: crypto.randomUUID(), taskId: task.workerTaskId, role: 'ROLE_USER', parts: [{ text }] },
});
if (out.error) return setState(task, 'TASK_STATE_FAILED', `計算担当がエラーを返しました: ${out.error.code} ${out.error.message}`);
settleFromWorker(task, out.result.task);
} catch (e) {
setState(task, 'TASK_STATE_FAILED', `計算担当に接続できません: ${e.message}`);
}
}
// ---- JSON-RPCメソッド ----
async function handleSendMessage(params) {
const msg = params?.message;
if (!Array.isArray(msg?.parts) || msg.parts.length === 0) {
return { error: { code: INVALID_PARAMS, message: 'Invalid parameters' } };
}
const text = msg.parts.map((p) => p.text ?? '').join('');
let task;
let work;
if (msg.taskId) {
task = tasks.get(msg.taskId);
if (!task) return { error: { code: TASK_NOT_FOUND, message: 'Task not found' } };
if (msg.contextId && msg.contextId !== task.contextId) {
return { error: { code: INVALID_PARAMS, message: 'contextId does not match the referenced task' } };
}
if (task.status.state !== 'TASK_STATE_INPUT_REQUIRED') {
return { error: { code: UNSUPPORTED_OPERATION, message: 'Task is not waiting for input' } };
}
task.history.push({ ...msg, contextId: task.contextId });
// 再開は同期的にWORKINGへ（INPUT_REQUIREDのままだとブロッキング待機が即返ってしまう）
setState(task, 'TASK_STATE_WORKING', '回答を計算担当へ転送します');
work = forwardAnswer(task, text);
} else {
task = {
id: crypto.randomUUID(),
contextId: msg.contextId ?? crypto.randomUUID(),
status: {},
artifacts: [],
history: [],
waiters: [],
};
// 履歴のMessageは単体で読めるように所属を補完してから保存する
task.history.push({ ...msg, taskId: task.id, contextId: task.contextId });
tasks.set(task.id, task);
setState(task, 'TASK_STATE_SUBMITTED');
work = delegate(task, text);
}
// A2A仕様 §3.2.2: 既定はブロッキング。
// blockingモードでは work の完了時点でタスクは終端または中断に達している。pushモードでは work は受領書で戻るため、終端または中断の通知（webhook）まで別途待つ
if (params?.configuration?.returnImmediately) {
void work; // 受領書を先に返し、取り次ぎは裏で続ける
} else {
await work;
await waitForSettled(task);
}
return { result: { task: taskView(task) } };
}
const server = http.createServer((req, res) => {
let body = '';
req.on('data', (c) => (body += c));
req.on('end', async () => {
const isWebhook = req.url?.startsWith('/webhook/');
const headerDump = ['host', 'content-type', 'a2a-version', 'if-none-match', 'authorization']
.filter((h) => req.headers[h] !== undefined)
.map((h) => `${h}: ${req.headers[h]}`)
.join('\n');
wire(isWebhook ? 'ワーカー → 受付 (webhook)' : '利用者 → 受付', `${req.method} ${req.url}\n${headerDump}${body ? '\n\n' + body : ''}`);
const reply = (status, headers, payload) => {
wire(isWebhook ? '受付 → ワーカー (webhook応答)' : '受付 → 利用者', `HTTP ${status}\n${Object.entries(headers).map(([k, v]) => `${k}: ${v}`).join('\n')}${payload ? '\n\n' + payload : ''}`);
res.writeHead(status, headers);
res.end(payload);
};
// ---- webhook受け口（pushモード）: ワーカーからのStreamResponseを受け取る ----
if (req.method === 'POST' && req.url === '/webhook/worker') {
const auth = req.headers['authorization'] ?? '';
const token = auth.startsWith('Bearer ') ? auth.slice('Bearer '.length) : '';
const task = pendingByToken.get(token);
if (!task) {
// 登録した覚えのないトークン = 差出人を確認できない通知は受け取らない
return reply(401, { 'Content-Type': 'application/json' }, JSON.stringify({ error: 'unknown or missing token' }));
}
let ev;
try {
ev = JSON.parse(body);
} catch {
return reply(400, { 'Content-Type': 'application/json' }, JSON.stringify({ error: 'invalid JSON' }));
}
reply(200, { 'Content-Type': 'application/json' }, '{}'); // 受領の2xxは先に返す義務（A2A仕様 §4.3.3）
// 通知は呼び鈴として使い、正式な状態はGetTaskで取り直す（webhookの順序・重複・欠落に依存しない）
const state = ev.statusUpdate?.status?.state;
if (state && (TERMINAL.has(state) || state === 'TASK_STATE_INPUT_REQUIRED')) {
const out = await callWorker('GetTask', { id: ev.statusUpdate.taskId });
if (!out.error) settleFromWorker(task, out.result); // GetTaskのresultはTaskそのもの
if (TERMINAL.has(task.status.state)) pendingByToken.delete(token); // INPUT_REQUIREDなら再開に備えて残す
}
return;
}
if (req.method === 'GET' && req.url === '/.well-known/agent-card.json') {
if (req.headers['if-none-match'] === CARD_ETAG) {
return reply(304, { ETag: CARD_ETAG, 'Cache-Control': 'max-age=3600' }, '');
}
return reply(200, { 'Content-Type': 'application/json', ETag: CARD_ETAG, 'Cache-Control': 'max-age=3600' }, JSON.stringify(AGENT_CARD, null, 2));
}
if (req.method === 'POST' && req.url === '/a2a/v1') {
const respond = (obj) => reply(200, { 'Content-Type': 'application/json' }, JSON.stringify(obj, null, 2));
const err = (id, code, message) => respond({ jsonrpc: '2.0', id, error: { code, message } });
let rpc;
try {
rpc = JSON.parse(body);
} catch {
return err(null, PARSE_ERROR, 'Invalid JSON payload');
}
if (rpc.jsonrpc !== '2.0' || typeof rpc.method !== 'string') {
return err(rpc.id ?? null, INVALID_REQUEST, 'Request payload validation error');
}
if ((req.headers['a2a-version'] ?? '0.3') !== '1.0') {
return err(rpc.id, VERSION_NOT_SUPPORTED, 'Version not supported');
}
switch (rpc.method) {
case 'SendMessage': {
const out = await handleSendMessage(rpc.params);
return respond({ jsonrpc: '2.0', id: rpc.id, ...(out.error ? { error: out.error } : out) });
}
case 'GetTask': {
const task = tasks.get(rpc.params?.id);
if (!task) return err(rpc.id, TASK_NOT_FOUND, 'Task not found');
// rpc GetTask returns (Task): resultはTaskそのもの（result.taskに包まない）
return respond({ jsonrpc: '2.0', id: rpc.id, result: taskView(task, rpc.params?.historyLength) });
}
default:
// ストリーミング等は非対応（Cardの capabilities.streaming: false と整合）
return err(rpc.id, METHOD_NOT_FOUND, 'Method not found');
}
}
reply(404, { 'Content-Type': 'application/json' }, JSON.stringify({ error: 'not found' }));
});
});
server.listen(PORT, () => {
console.log(`Calc Reception Agent (A2A v1.0, JSON-RPC binding) : ${BASE}`);
console.log(` Agent Card : ${BASE}/.well-known/agent-card.json`);
console.log(` JSON-RPC : POST ${BASE}/a2a/v1`);
console.log(` worker : ${WORKER_BASE}（Slow Calc Agent）`);
console.log(` 取り次ぎ : ${DELEGATE_MODE}${DELEGATE_MODE === 'push' ? `（webhook受け口: POST ${BASE}/webhook/worker）` : ''}`);
console.log(` wire log : ${WIRE_LOG}`);
});