feat(client): add RetryTransport for automatic retry with exponential backoff

[cchinchilla-dev]

Description

• Follow the CONTRIBUTING Guide.
• Make your Pull Request title in the Conventional Commits specification.
• Ensure the tests and linter pass (bash scripts/format.sh from the repository root to format).
• Appropriate docs were updated (if necessary).

Closes #871 🦕

Summary

Adds a RetryTransport decorator wrapping any ClientTransport, following the existing TenantTransportDecorator pattern.

Public API

from a2a.client.transports.retry import RetryTransport

inner = JsonRpcTransport(httpx_client=client, agent_card=card)
transport = RetryTransport(
    base=inner,
    max_retries=3,
    base_delay=1.0,
    max_delay=30.0,
    jitter=True,
    retry_predicate=None,    # Callable[[Exception], bool] | None
    on_retry=None,           # Callable[[int, Exception, float], Awaitable[None] | None] | None
)

async with transport:
    result = await transport.send_message(params)

RetryTransport, default_retry_predicate, RetryPredicate, and OnRetryCallback are exported from a2a.client.transports.

Implementation details

Default retry classification

default_retry_predicate inspects error.__cause__ (falling back to error.__context__ when __cause__ is None) and retries on:

• A2AClientTimeoutError (always — covers HTTP timeouts and gRPC DEADLINE_EXCEEDED, which _map_grpc_error normalizes to this type).
• A2AClientError chained from httpx.RequestError.
• A2AClientError chained from httpx.HTTPStatusError(408 | 429 | 502 | 503 | 504).
• A2AClientError chained from grpc.aio.AioRpcError(UNAVAILABLE | RESOURCE_EXHAUSTED).

Domain errors (TaskNotFoundError, etc., which inherit A2AError not A2AClientError), HTTP 5xx other than 502/503/504 (replaying server bugs is not safe), and JSON / SSE decode errors are never retried. Pluggable via retry_predicate.

Streaming, cancellation, resource safety

• Pre-stream retry only. send_message_streaming and subscribe retry only if the connection fails before the first event is yielded. Once any event has been emitted, subsequent failures propagate without retry — no partial-stream replay.
• No socket leaks. Inner generators are aclose()-d on every exit path — success, retry, exception, and consumer-side break. Both _execute_streaming_with_retry and the public send_message_streaming / subscribe wrappers propagate aclose through the chain.
• Cancellation. asyncio.CancelledError is re-raised explicitly from every retry path (_execute_with_retry, _execute_streaming_with_retry, _safe_aclose, _delay_and_notify) so cancellation during backoff is never swallowed.

Backoff

Full-jitter exponential: delay = random.uniform(0, min(base_delay * 2^attempt, max_delay)). Set jitter=False for deterministic delays.

Methods that bypass retry

close() delegates directly without retry — lifecycle operation, not data exchange.

Validation

Tests. 57 unit tests in tests/client/transports/test_retry.py (predicate classification, backoff math, streaming semantics, consumer-break cleanup, cancellation, sync/async on_retry, constructor validation, exports). 3 end-to-end tests in tests/integration/test_retry_integration.py exercising ClientFactory → BaseClient → RetryTransport against Starlette servers with transient 503 middleware.

Smoke-tested end-to-end against real uvicorn and grpc.aio.server localhost servers with fault injection. HTTP and gRPC paths verified for retry, exhaustion, non-retryable propagation, streaming semantics, socket cleanup, and cancellation.

Scope

Purely additive:

• src/a2a/client/transports/retry.py
• tests/client/transports/test_retry.py
• tests/integration/test_retry_integration.py
• Public exports in src/a2a/client/transports/__init__.py

No changes to existing transports, error types, or interceptor APIs. No new runtime dependencies — the grpc import is conditional, matching the SDK's existing pattern for optional dependencies.

Deferred follow-ups

These came up during review but were intentionally kept out of scope to keep this PR focused. Happy to follow up if any of them are wanted:

• ClientFactory integration (e.g., ClientConfig.transport_decorators: list[Callable[[ClientTransport], ClientTransport]]) so users can wrap declaratively instead of composing transports manually.
• Retry-After header support for HTTP 429/503 responses.
• Global deadline budget across all retry attempts.

Happy to iterate on any of this based on review.

[github-actions]

🧪 Code Coverage (vs main)

⬇️ Download Full Report

	Base	PR	Delta
src/a2a/client/transports/retry.py (new)	—	96.43%	—
Total	93.00%	93.07%	🟢 +0.07%

Generated by coverage-comment.yml

[gemini-code-assist]

Code Review

This pull request introduces a RetryTransport decorator for the A2A client, enabling automatic retries with exponential backoff and jitter for transient failures. It includes a default retry predicate that handles specific HTTP status codes, gRPC error codes, and network timeouts. The implementation covers both standard and streaming RPC methods, ensuring that streaming calls only retry if the failure occurs before the first event is yielded. Comprehensive unit and integration tests have been added to verify the retry logic across different transport protocols. I have no feedback to provide.