A2A Push Notifications, in depth: configuring webhooks with TaskPushNotificationConfig, the four config operations, credentials, URL policy and receivers

What a push config is

A push notification config binds one webhook to one task. The server stores it and, whenever that task changes, sends an HTTP POST to the URL. A task may have several configs at once, for example one for the agent that delegated the work and one for a monitoring system. The specification says a config must persist until the task completes or the config is explicitly deleted, and it defines create, get, list and delete operations, but no update. Changing a URL or rotating a credential therefore means creating a new config and deleting the old one.

The object, field by field

The specification also says how the credentials are used: the agent must include them in the request headers in standard HTTP form, so a config with scheme Bearer and credentials abc produces Authorization: Bearer abc. It does not define a header for token. The Python SDK sends it as X-A2A-Notification-Token when it is set; do not rely on that header from servers built on other stacks unless their documentation says so.

Check the capability first

Push is optional. An agent advertises it with capabilities.pushNotifications: true in its Agent Card, and if the flag is false or absent every config operation must fail with PushNotificationNotSupportedError: JSON-RPC code -32003, gRPC FAILED_PRECONDITION, HTTP 400 with a google.rpc.ErrorInfo detail to tell it apart from other 400s. Clients should read the flag from the card before sending a config and fall back to streaming or polling when it is missing; A2A capabilities covers the card side.

Registering: inline or after the fact

There are two ways to attach a config. The first is inline, inside SendMessageConfiguration.taskPushNotificationConfig, with the task ID left empty because the task does not exist yet. Combined with returnImmediately: true, the call returns the new task straight away and the server sends every later update to the webhook. This is the right default for long work: there is no window between task creation and registration in which an early update could be missed.

POST /message:send HTTP/1.1
Host: research-agent.example.com
Content-Type: application/a2a+json
A2A-Version: 1.0
Authorization: Bearer <client-to-server token>

{
  "message": {
    "role": "ROLE_USER",
    "messageId": "0b6f2c1e-5d1a-4c55-9a7e-2f1d8c3e9a10",
    "parts": [{"text": "Assess supplier risk for ACME Ltd, full report."}]
  },
  "configuration": {
    "returnImmediately": true,
    "taskPushNotificationConfig": {
      "url": "https://hooks.buyer.example.com/a2a/7f3c",
      "token": "t_7f3c_9b1e",
      "authentication": {"scheme": "Bearer", "credentials": "whsec_only_for_this_config"}
    }
  }
}

The second is the Create operation on an existing task. Use it to add a second subscriber, to move notifications to a new endpoint, or after a client restart when you want to be told about a task you are already tracking. The REST binding nests configs under the task; the JSON-RPC and gRPC bindings expose the same four operations as methods. List supports pageSize and pageToken and returns nextPageToken. Delete must be idempotent, so a cleanup job can retry it safely.

# REST: register (or add a second) config for an existing task
POST /tasks/43667960-d455-4453-b0cf-1bae4955270d/pushNotificationConfigs
{"url": "https://hooks.buyer.example.com/a2a/monitoring",
 "authentication": {"scheme": "Bearer", "credentials": "whsec_monitoring_only"}}

# -> 200 with the stored config; the server assigns "id"
{"id": "cfg-2", "taskId": "43667960-...", "url": "https://hooks.buyer.example.com/a2a/monitoring",
 "authentication": {"scheme": "Bearer", "credentials": "..."}}

GET    /tasks/{taskId}/pushNotificationConfigs?pageSize=50      # List, then follow nextPageToken
GET    /tasks/{taskId}/pushNotificationConfigs/cfg-2            # Get
DELETE /tasks/{taskId}/pushNotificationConfigs/cfg-2            # Delete (idempotent)

# JSON-RPC: same operations as methods; Create's params are the config object itself
{"jsonrpc": "2.0", "id": 7, "method": "CreateTaskPushNotificationConfig",
 "params": {"taskId": "43667960-...", "url": "https://hooks.buyer.example.com/a2a/monitoring",
            "authentication": {"scheme": "Bearer", "credentials": "whsec_monitoring_only"}}}

What the webhook receives

Each notification is an HTTP POST whose body is a StreamResponse, the same wrapper used by streaming, containing exactly one of task, message, statusUpdate or artifactUpdate. The specification's example uses the application/a2a+json content type; the Python SDK posts with its HTTP client's JSON helper, which sends application/json, so a receiver should accept both. A status update carries the task ID, context ID and the new status with its state, for example TASK_STATE_INPUT_REQUIRED or TASK_STATE_COMPLETED; the meaning of each state is in A2A task lifecycle states.

The specification's guarantees are deliberately modest. Agents must attempt delivery at least once per configured webhook, may retry with exponential backoff, should time out webhook requests after about 10 to 30 seconds, and may give up after a number of consecutive failures. Clients must answer with a 2xx status, must validate that the notification is authentic, should check that the task ID is one they created, and should process notifications idempotently because duplicates may occur. Treat a notification as a prompt to fetch state with GetTask, not as the state itself.

Choosing and rotating credentials

The credentials in a config are a secret you hand to another organization's server, and that server will store them. Give each config its own single-purpose secret that authorizes exactly one thing: posting to that webhook. Never reuse the client's own API credentials, and never put a credential that works anywhere else into credentials. The specification recommends unique, single-purpose tokens per config, treating them as secrets and rotating them periodically.

Bearer with a random 256-bit secret is the simplest scheme that meets those rules. The token field adds a second, task-bound value that is cheap to check and makes a replay of one config's notification to another config's endpoint fail. Rotation follows from the missing update operation: create a config with the new secret, accept both secrets at the receiver for an overlap window, delete the old config, then stop accepting the old secret. Encode the config in the webhook path, as the receiver below does, so you can look up the expected secret before parsing the body.

The server side: store, validate, deliver

A server that accepts configs is agreeing to make outbound HTTP requests to URLs chosen by its clients, which is a classic server-side request forgery risk. The specification says agents should reject private ranges, localhost and link-local addresses and use allowlists where appropriate. Validate at Create time, and again at delivery time, because a hostname that resolved to a public address yesterday can resolve to an internal one today. Connect to the address you validated rather than resolving again.

import asyncio, ipaddress, random, socket
from urllib.parse import urlsplit

def validate_push_url(url: str, allow_hosts: set[str] | None = None) -> list[str]:
    """Run at Create time AND before each delivery (DNS can change)."""
    u = urlsplit(url)
    if u.scheme != "https":
        raise ValueError("webhook must use https")
    if allow_hosts is not None and u.hostname not in allow_hosts:
        raise ValueError("host not on allowlist")
    addrs = {ai[4][0] for ai in socket.getaddrinfo(u.hostname, u.port or 443)}
    for a in addrs:
        ip = ipaddress.ip_address(a)
        ip = getattr(ip, "ipv4_mapped", None) or ip   # ::ffff:127.0.0.1 is loopback
        if not ip.is_global:                          # private, loopback, link-local...
            raise ValueError(f"{u.hostname} resolves to non-public address {ip}")
    return sorted(addrs)    # connect to one of these, not to a fresh lookup

async def deliver(http, cfg, stream_response: dict, max_attempts=6):
    headers = {"Content-Type": "application/a2a+json"}
    if cfg.authentication and cfg.authentication.credentials:
        headers["Authorization"] = f"{cfg.authentication.scheme} {cfg.authentication.credentials}"
    if cfg.token:
        headers["X-A2A-Notification-Token"] = cfg.token   # SDK convention, not spec
    for attempt in range(max_attempts):
        try:
            validate_push_url(cfg.url)
        except ValueError:
            break                              # policy rejection: never retry
        try:
            # sketch: post() resolves again; production code must pin the validated IP
            r = await http.post(cfg.url, json=stream_response, headers=headers, timeout=15)
            if 200 <= r.status_code < 300:
                return True
            if r.status_code in (400, 401, 403, 404, 410):
                break                          # receiver rejected it; retrying will not help
        except Exception:
            pass
        await asyncio.sleep(min(300, 2 ** attempt) * random.uniform(0.5, 1.5))
    record_failure(cfg)                        # count consecutive failures; disable past N
    return False

Delivery policy is yours to set within the specification's bounds. Note that the reference Python SDK's base sender, at the time of writing, makes one POST per event per config, logs failures without retrying, and screens URLs only if you pass it a validator; production servers usually wrap or replace it with a queue and retries like the sketch above. Delete or expire configs once a task is terminal, cap the number of configs per task and per client, and record config creation and deletion as audit events. The broader threat model is in A2A security.

The client side: a receiver that verifies and acks fast

The receiver has one job during the request: decide quickly whether the notification is authentic and new, enqueue it, and return 2xx. Any slow work inside the handler risks the sender's timeout and a duplicate retry. Compare secrets in constant time, check that the task ID belongs to the config whose path was called, deduplicate, and leave the real work to a worker that calls GetTask.

import hmac, json
from aiohttp import web

async def a2a_hook(request: web.Request) -> web.Response:
    cfg = CONFIGS.get(request.match_info["hook_id"])          # one route per config
    if cfg is None:
        return web.Response(status=404)
    expected = f"Bearer {cfg.secret}"
    if not hmac.compare_digest(request.headers.get("Authorization", ""), expected):
        return web.Response(status=401)
    body = json.loads(await request.read())                   # accept a2a+json or json
    event = (body.get("statusUpdate") or body.get("artifactUpdate")
             or body.get("task") or body.get("message") or {})
    task_id = event.get("taskId") or event.get("id")
    if task_id != cfg.task_id:                                  # spec: check the task is ours
        return web.Response(status=403)
    key = (task_id, json.dumps(event, sort_keys=True))
    if await SEEN.add_if_absent(key, ttl_s=86_400):             # duplicates are allowed
        await QUEUE.put({"task_id": task_id, "kind": next(iter(body))})
    return web.Response(status=204)                             # ack fast, work later

# worker: on each queued item, call GetTask and act on the authoritative state
app = web.Application()
app.add_routes([web.post("/a2a/{hook_id}", a2a_hook)])

Worked example: a supplier-risk report

A procurement orchestrator delegates a supplier-risk assessment that usually takes 30 to 50 minutes. It checks the remote agent's card for pushNotifications, generates a secret and token for this task, stores them under a new hook ID, and sends the message with an inline config and returnImmediately. The server validates the URL, stores the config as cfg-1 and returns the task in the submitted state.

Twelve minutes in, the webhook receives a status update to TASK_STATE_INPUT_REQUIRED. The worker calls GetTask, reads the agent's question about which subsidiaries to include, and answers on the same task. At minute twenty the client's operations team adds a monitoring subscriber with Create, producing cfg-2. At minute thirty the orchestrator's ingress moves to a new hostname: it creates cfg-3 with a fresh secret, accepts both secrets for five minutes, then deletes cfg-1. The completion notification arrives at the new hook, the worker fetches the task and its artifacts, and a cleanup step deletes the remaining configs. A slow reconciliation loop that polls unfinished tasks every ten minutes is still running in case any notification is lost.

Failure modes

Trade-offs

Push removes polling traffic and long-lived connections, but it needs an endpoint the remote agent can reach, secrets on both sides and a reconciliation path for missed notifications. Streaming, covered in A2A streaming, is simpler when a user is watching and the task finishes within the life of a connection. Polling works through any network and needs no inbound access. Many clients use all three: stream while connected, register push for long tasks, and poll slowly as the safety net.