refactor(integrations): migrate A2A SDK to 1.1.2

[AlexanderZ-Band]

Summary

• Migrate Band A2A integrations and six examples to a2a-sdk>=1.1.2,<2 using official route factories, helpers, and task lifecycle APIs.
• Expose per-peer JSON-RPC/REST routes with 1.0 and v0.3 discovery compatibility.
• Preserve Band-specific streaming reduction, room/context correlation, timeout handling, and UUID peer aliases.
• Use the Fern client retry policy for gateway peer discovery: this reduces retry headroom from the removed bespoke loop, but adds Retry-After-aware backoff.

Validation

• uv run pytest tests/ --ignore=tests/integration/ --ignore=tests/e2e/ -q --no-cov
• uv run ruff check .
• uv run ruff format --check .
• uv run pyrefly check
• Isolated wheel installs and imports for a2a_gateway and a2a_gateway_demo extras.

[linear-code]

INT-970

INT-1149

[amit-gazal-thenvoi]

Blocking: per-alias Mount("/{tenant}") shadows the REST binding for every peer after the first

Nice migration overall — leaning on the upstream route factories and deleting the hand-rolled JSON-RPC/SSE layer is clearly the right call. One issue in GatewayServer._build_app blocks it, though.

The issue

create_rest_routes(..., enable_v0_3_compat=True, path_prefix=...) returns 23 routes, and the last one is a catch-all Mount(path="/{tenant}") for multi-tenant hosts. server.py:119-125 calls it once per alias and flattens everything into a single list, so the first alias's tenant mount ends up ahead of every later alias's flat routes.

Starlette dispatches on the first Match.FULL, and a Mount matches on prefix alone — so /agents/<anything>/... is captured by alias #1's mount and 404s inside its inner router, which only knows alias #1's paths.

Agent-card and JSON-RPC routes survive purely because protocol_routes is concatenated before rest_routes (server.py:127). Everything under the REST binding does not.

Measured against the real GatewayServer._build_app() with two peers:

alias	/.well-known/agent-card.json	POST /message:stream
weather-agent (first registered)	200	200
uuid-weather	200	404
billing-agent	200	404
uuid-billing	200	404

Impact: the REST streaming endpoint works for exactly one peer, and never for a UUID alias. A single-peer gateway still loses its UUID alias. This is the endpoint advertised in README.md:611, examples/a2a_gateway/README.md, examples/a2a_gateway/01_basic_gateway.py:90,99, and examples/run_agent.py:983.

Why CI is green

Every test in test_server.py uses one peer (build_server(), line 42), and the only multi-alias test — test_uuid_peer_alias_serves_the_same_agent_card (line 106) — hits the card route, which is in the group that isn't shadowed. Nothing exercises a second peer's REST route.

Test to reproduce

Add to tests/integrations/a2a/gateway/test_server.py (reuses the existing FakeExecutor and make_peer):

def build_multi_peer_server() -> GatewayServer:
    weather = make_peer("uuid-weather", "Weather Agent", "Gets weather info")
    billing = make_peer("uuid-billing", "Billing Agent", "Answers billing questions")
    return GatewayServer(
        peers={"weather-agent": weather, "billing-agent": billing},
        peers_by_uuid={weather.id: weather, billing.id: billing},
        gateway_url="http://localhost:10000",
        port=10000,
        executor_factory=lambda _slug: FakeExecutor(),
    )


@pytest_asyncio.fixture
async def multi_peer_client() -> AsyncIterator[httpx.AsyncClient]:
    transport = ASGITransport(app=build_multi_peer_server()._build_app())
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        yield client


async def test_rest_binding_is_reachable_for_every_peer_and_alias(
    multi_peer_client: httpx.AsyncClient,
) -> None:
    """Every peer must serve the REST route the gateway docs advertise.

    A gateway hosting more than one peer is the normal case, and each peer is
    addressable by slug and by UUID, so all four routes have to answer.
    """
    aliases = ("weather-agent", "uuid-weather", "billing-agent", "uuid-billing")

    reached_handler = {}
    for alias in aliases:
        response = await multi_peer_client.post(
            f"/agents/{alias}/message:stream",
            headers={"A2A-Version": "1.0"},
            json={
                "message": {
                    "messageId": "message-1",
                    "role": "ROLE_USER",
                    "parts": [{"text": "Hello"}],
                }
            },
        )
        reached_handler[alias] = response.status_code

    assert reached_handler == dict.fromkeys(aliases, 200)

On this branch it fails with:

Differing items:
{'uuid-weather':  404} != {'uuid-weather':  200}
{'billing-agent': 404} != {'billing-agent': 200}
{'uuid-billing':  404} != {'uuid-billing':  200}

(weather-agent: 200 is the one matching item.)

Suggested fix

This gateway namespaces peers by path and isn't multi-tenant, so the tenant mounts can be dropped — in server.py:119:

rest_routes.extend(
    route
    for route in create_rest_routes(
        handler,
        enable_v0_3_compat=True,
        path_prefix=f"/agents/{alias}",
    )
    # Each call also returns a catch-all "/{tenant}" mount for multi-tenant
    # hosts. Peers here are namespaced by path, and the first mount would
    # shadow every later peer's REST routes.
    if not isinstance(route, Mount)
)

plus from starlette.routing import BaseRoute, Mount, Route.

Verified: with that change the new test passes and the rest of the suite is unaffected (36 passed).

---

Related, separate from the above: /v1/ is the legacy REST path

Worth fixing while in here. Under a2a-sdk 1.1.2 the two REST paths map to different protocol versions:

Path	Protocol
/agents/<peer>/message:stream	1.0
/agents/<peer>/v1/message:stream	v0.3 compat

Posting with A2A-Version: 1.0 to the /v1/ path returns:

400 — A2A version '1.0' is not supported by this handler. Expected version '0.3'.

So the /v1 prefix reads like a newer route but is the legacy binding. examples/a2a_gateway/README.md currently tells 1.0 clients to "use the versioned route" /v1/message:stream with a SendMessageRequest wrapper — a 1.0 client needs /agents/<peer>/message:stream instead. Same path appears in README.md:611, examples/a2a_gateway/01_basic_gateway.py:90,99, and examples/run_agent.py:983.

Also note test_rest_stream_runs_through_upstream_handler covers /message:stream, so the /v1 path the docs advertise has no test.

🤖 Review generated with Claude Code

[amit-gazal-thenvoi]

Follow-up: secondary findings

Separate from the routing bug above. Each of these was reproduced by running it against a2a-sdk==1.1.2, not just read off the diff. Roughly in severity order.

---

1. GET /agents/<peer>/tasks returns conversation content, unauthenticated

Adopting create_rest_routes brings in task endpoints the gateway never had before. The task listing is reachable with no credentials and inlines the agent's reply text:

GET /agents/weather-agent/tasks   ->  200

{"tasks":[{"id":"3c7e…","contextId":"4242…","status":{"state":"TASK_STATE_COMPLETED",
 "message":{"role":"ROLE_AGENT","parts":[{"text":"The salary review is scheduled Friday"}]…

The pre-PR server exposed only the card, message:stream, and JSON-RPC message/send|stream — no way to enumerate or read past tasks. Combined with the host="0.0.0.0" bind (server.py:166, pre-existing) and no auth layer, this is a content-disclosure change rather than just a larger surface. tasks/{id} and tasks/{id}:cancel are live too; extendedAgentCard and the push-notification config routes currently fail closed with 400.

If these aren't wanted, they can be filtered out in the same pass as the tenant mounts.

---

2. _handle_event lost its try/finally, stranding a dead task_id on the room

integrations/a2a/adapter.py:185-190. On main, _emit_task_event + cleanup ran in a finally, so a delivery failure still finalized the task. Now they're sequential, so if _deliver_task_update raises the room keeps pointing at a completed task.

Reproduced with tools.send_message raising on a completed task carrying artifact text:

send_message awaited : 1
exception propagated : RuntimeError('platform down')
_tasks               : {'room-1': 'task-1'}   <- stale
task event emitted   : 0                      <- rehydration record lost
outbound task_id     : 'task-1'               <- next turn continues a completed task

on_message catches the exception and posts an error event, so nothing crashes — the room just silently keeps sending task_id of a finished task on every subsequent turn. Restoring the finally fixes it.

---

3. Each alias gets its own task store, so tasks are invisible across aliases

server.py:92-96 constructs a fresh DefaultRequestHandler + InMemoryTaskStore per alias — 2 handlers and 2 distinct stores for a single peer. A task started on the slug can't be fetched or cancelled on the UUID:

task created via POST /agents/weather-agent
  GET /agents/weather-agent/tasks/<id>  -> 200
  GET /agents/uuid-w/tasks/<id>         -> 404

Note this is currently masked by the tenant-mount bug (the UUID alias 404s for unrelated reasons) — it surfaces the moment that's fixed, so worth handling in the same change. One handler per peer, shared across its aliases, solves this and item 5 together.

---

4. The new httpx.AsyncClient is never closed — and stop() is not the hook

integrations/a2a/adapter.py:106 creates a client the adapter owns. main didn't have this: it used ClientFactory.connect(...), which managed its own transport. So the PR introduces an ownership obligation it doesn't discharge.

The fix isn't a stop() method — Agent.stop() only ever calls cleanup_all():

# agent.py:304
cleanup_all = getattr(self._adapter, "cleanup_all", None)

Neither A2A adapter overrides it, so async def cleanup_all(self) on A2AAdapter is the right place to aclose().

Related, pre-existing (not this PR): A2AGatewayAdapter.stop() (gateway/adapter.py:264) is dead code for the same reason — nothing in the SDK calls it, so the uvicorn server leaks on agent shutdown. It exists on main too, but SimpleAdapter.cleanup_all's own docstring names "a self-hosted server" as exactly what belongs in that hook, and this PR is already in the file.

---

5. _task_cache entries created by resubscribe are never reclaimed

on_cleanup (integrations/a2a/adapter.py:300) derives its keys from _task_senders, but _try_resubscribe → _reduce_task_event populates _task_cache without ever touching _task_senders:

after resubscribe-style reduce : _task_cache=[('room-2','task-9')]  _task_senders=[]
after on_cleanup('room-2')     : _task_cache=[('room-2','task-9')]   <- still there

Small, but it's an unbounded dict on a long-lived adapter. Iterating _task_cache's own keys fixes it.

---

6. Route count grows ~52 per peer

Measured on the real _build_app():

peers	top-level routes
1	53
2	105
5	261
50	2601

Two aliases × (1 card + 1 legacy card + 1 JSON-RPC + 23 REST). Starlette matches linearly, so every request on a 50-peer gateway walks a 2601-entry list, and each peer also carries 2 handlers + 2 task stores. A single parameterized /agents/{peer_id} route set with peer resolution inside the executor would collapse this and fix item 3 at the same time.

---

7. Timeout path logs success

gateway/adapter.py:326-331. _await_response swallows TimeoutError, so the else: branch runs anyway:

WARNING A2A response timed out: room=room-1 task=b16ba967… timeout=0.01s
INFO    A2A request completed: room=room-1 task=b16ba967…

---

Smaller items

• protocol.py:94 — is_terminal_state() is dead code; no caller in src, tests, or examples.
• server.py:42 — self.peers_by_uuid is now write-only; aliasing is derived from peer.id in _build_app instead.
• integrations/a2a/adapter.py:174 — getattr(event, "HasField", lambda _name: False)("message"). StreamResponse always has HasField, and all six _handle_event tests pass real protobuf events, so nothing motivates the guard. It's also a trap: MagicMock().HasField("message") is truthy, so a future mock-based test would silently take the message branch. Plain event.HasField("message") reads as what's meant.
• gateway/types.py:86 — def _message(self, content: str): is missing its return type.
• server.py:147 — _handle_list_peers re-imports JSONResponse although it's already imported at line 20; the signature is also (_request: Any) -> Any where (request: Request) -> JSONResponse is available.
• gateway/adapter.py:298 — _establish_request runs before the try at line 306, so peer-not-found and room-creation failures skip the structured logger.exception the rest of the method provides.
• CLAUDE.md:209-211 — the A2A key-files table still lists only types.py; the new protocol.py and config.py aren't there, and response_timeout_s isn't documented outside the README snippet.
• test_adapter.py:80 — test_timeout_is_adapter_configuration mostly asserts the constructor stored its argument. test_timeout_returns_terminal_failure is the one carrying the weight.

---

Verified as fine

Worth recording so they don't get re-litigated:

• Dropping _list_peers_page_with_retry is safe. The Fern client's _should_retry already covers {429, 408, 409} and 5xx with jittered backoff honoring Retry-After / RateLimit-Reset, and DEFAULT_REQUEST_OPTIONS = {"max_retries": 3} is threaded through. Fewer attempts than the old five, but it respects server-supplied backoff, which the hand-rolled loop did not.
• Gateway cancellation is correct. ActiveTask.cancel calls self._producer_task.cancel() before agent_executor.cancel, so _execute_a2a's CancelledError handler runs and pending_task's finally releases the room slot.
• raise InternalError() / raise UnsupportedOperationError() in the examples is valid — in 1.1.2 these derive from A2AError(Exception), so dropping the ServerError wrapper is right.
• PendingA2ATask's lock + done guard correctly make fail/complete/cancel idempotent and mutually exclusive.

🤖 Review generated with Claude Code

[amit-gazal-thenvoi]

Request changes

The A2A 1.1.2 migration is the right direction — leaning on the upstream route factories and deleting ~3200 lines of hand-rolled JSON-RPC/SSE is a clear win, protocol.py's reducer is well factored, and the aliasing test in test_protocol.py is the kind that catches real regressions. One blocker and a few correctness issues before it lands.

Everything below was reproduced by running it against a2a-sdk==1.1.2, not read off the diff.

Blocking

• Per-alias Mount("/{tenant}") shadows the REST binding for every peer after the first, so /agents/<peer>/message:stream works for exactly one peer and never for a UUID alias. Repro test + one-line fix inline on server.py.

Should fix before merge

• GET /agents/<peer>/tasks returns conversation content unauthenticated — new exposure vs. the pre-PR server.
• _handle_event lost its try/finally, stranding a completed task_id on the room when delivery fails.
• Per-alias task stores make a task invisible across a peer's own aliases (surfaces once the mount bug is fixed).
• The new httpx.AsyncClient is never closed; cleanup_all() is the hook, not stop().

Smaller — _task_cache leak, contradictory timeout log, ~52 routes per peer, dead is_terminal_state() / peers_by_uuid, the getattr(HasField) guard, missing return type, redundant import.

Not inline-able

• CLAUDE.md:209-211 — the A2A key-files table still lists only types.py; the new protocol.py and config.py aren't there, and response_timeout_s isn't documented outside the README snippet.
• Docs point 1.0 clients at the wrong path: under 1.1.2, /agents/<peer>/v1/message:stream is the v0.3 binding (A2A-Version: 1.0 there returns 400 - A2A version '1.0' is not supported by this handler. Expected version '0.3'). 1.0 clients need /agents/<peer>/message:stream. Affects README.md:611, examples/a2a_gateway/README.md, examples/a2a_gateway/01_basic_gateway.py:90,99, examples/run_agent.py:983. test_rest_stream_runs_through_upstream_handler covers /message:stream, so the documented /v1 path has no test.
• test_adapter.py — test_timeout_is_adapter_configuration mostly asserts the constructor stored its argument; test_timeout_returns_terminal_failure is the one carrying the weight.

Verified as fine, so they don't get re-litigated: dropping _list_peers_page_with_retry is safe (Fern's _should_retry already covers {429, 408, 409} + 5xx with Retry-After-aware backoff, max_retries=3 threaded through); gateway cancellation is correct (ActiveTask.cancel cancels the producer before calling agent_executor.cancel, so pending_task's finally releases the slot); raise InternalError() in the examples is valid since 1.1.2 errors derive from A2AError(Exception); and PendingA2ATask's lock + done guard correctly make the terminal transitions idempotent.

Detail on each of these is in the two comments above.

🤖 Review generated with Claude Code

[AlexanderZ-Band]

Self-review round (25db472)

Ran a deep verification pass over the full diff before requesting re-review. Everything below was reproduced empirically before fixing:

Security — JSON-RPC task exposure (completes the earlier REST fix): the REST filtering in f4985f3 closed only half the surface. The JSON-RPC dispatcher still served ListTasks (returned every caller's tasks with full message history — all unauthenticated callers share one task-store identity), plus push-config and extended-card methods. Now closed via ALLOWED_JSONRPC_METHODS (messaging + task-UUID-gated operations, 1.0 and v0.3 spellings), guarded by test_non_messaging_jsonrpc_methods_are_not_exposed. Follow-up for the broader bind/auth question: INT-1149.

Socket leak on stop: cancelling serve() skipped uvicorn's shutdown; a stop/start cycle failed with EADDRINUSE. Now sets should_exit and awaits. Verified rebind works.

Docs that didn't work: all three newly-authored gateway curl commands failed against the shipped server (v0.3 payload shapes fed to the 1.0 parser → -32602; /v1/message:stream mislabeled as the 1.0 REST binding → silent 200 + empty stream; demo docstring missing the A2A-Version header). All rewritten to shapes verified by execution.

Hardening: pre-migration (0.x) terminal state names recognized on rehydration; cleanup_all fails in-flight gateway requests; concurrent-request rejection no longer leaks the internal room id; resubscribe stream closed deterministically; proto presence checks fixed; a2a-sdk capped <2; demo discover_peer no longer closes the shared connection pool.

Coverage restored (dropped by the test rewrite, behaviors still live): auth-plumbing identity into ClientConfig, history-converter round trip with the new state vocabulary, remote-error and failed-task paths, context continuity across turns, bootstrap gating, PendingA2ATask terminal idempotency. Net +17 tests; full suite 4044 passed, ruff/pyrefly clean.

🤖 Generated with Claude Code

https://claude.ai/code/session_011Y2bdwA8CxTkD9SXxfU8et

[amit-gazal-thenvoi]

Migration itself checks out — I verified every changed call against a real a2a-sdk==1.1.2 install (ClientFactory.create_from_url, send_message(SendMessageRequest), subscribe(SubscribeToTaskRequest(id=...)), async Client.close(), new_text_message kwargs, TaskUpdater.start_work/complete/failed/cancel, RequestContext id generation, and InternalError/UnsupportedOperationError now being real exceptions). Locally: 4279 passed / 82 skipped, ruff clean, pyrefly 0 errors, README markdown-docs green.

One blocking item below (the a2a_gateway extra can't import, one line to fix), one question, one bit of praise.

[amit-gazal-thenvoi]

Approved. Just one blocking issue.