fix(transport): activate required Agent Card extensions on all requests#197
fix(transport): activate required Agent Card extensions on all requests#197kuangmi-bit wants to merge 1 commit into
Conversation
When a SUT declares required extensions in its Agent Card (capabilities.extensions[] with required: true), the TCK transport clients now automatically include the A2A-Extensions header (HTTP) or a2a-extensions metadata (gRPC) on every request. Previously, transport clients only sent A2A-Version. This caused SUTs that enforce required extensions to return ExtensionSupportRequiredError on every positive test — breaking CORE-SEND-*, CORE-STREAM-*, GRPC-ERR-*, and related tests. Changes: - _helpers.py: add get_required_extensions() and A2A_EXTENSIONS_HEADER - base.py: accept required_extensions keyword in __init__ - http_json_client.py / jsonrpc_client.py: include A2A-Extensions in default httpx.Client headers when extensions are declared - grpc_client.py: replace static _METADATA with dynamic _metadata property that appends a2a-extensions when required - conftest.py: extract required extensions from agent_card and pass them to all transport client constructors Negative tests (test_error_handling.py) use raw _jsonrpc_call / _rest_call helpers that bypass the transport clients entirely, so they continue to omit extensions by design. This fixes a2aproject#193
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces support for required extensions in the A2A protocol by extracting them from the agent card and sending them via the A2A-Extensions header or metadata across all transport clients (gRPC, HTTP+JSON, and JSON-RPC). The review feedback suggests making the extraction helper more defensive against malformed input and refactoring the transport clients to store the required_extensions attribute in the base class to eliminate duplicate initialization logic.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| def get_required_extensions(agent_card: dict) -> list[str]: | ||
| """Extract required extension URIs from an Agent Card. | ||
|
|
||
| Per A2A spec §8.3.3, ``capabilities.extensions[]`` entries with | ||
| ``required: true`` must be activated on every request via the | ||
| ``A2A-Extensions`` header (HTTP) or ``a2a-extensions`` metadata (gRPC). | ||
|
|
||
| Returns a list of extension URIs (may be empty). | ||
| """ | ||
| caps = agent_card.get("capabilities", {}) | ||
| extensions = caps.get("extensions", []) | ||
| return [ | ||
| ext["uri"] | ||
| for ext in extensions | ||
| if isinstance(ext, dict) and ext.get("required") and ext.get("uri") | ||
| ] |
There was a problem hiding this comment.
Since this is a Technology Compatibility Kit (TCK), it must be highly resilient to non-compliant or malformed responses from the System Under Test (SUT). If the SUT returns an invalid agent card (e.g., a JSON list instead of an object, or if capabilities or extensions are None or of unexpected types), this function will raise an unhandled AttributeError or TypeError and crash the test suite.
Additionally, we should deduplicate the extracted extension URIs to avoid sending duplicate values in the headers/metadata, and ensure that required is strictly a boolean True and uri is a string.
Let's refactor this function to be fully defensive and robust.
def get_required_extensions(agent_card: dict) -> list[str]:
"""Extract required extension URIs from an Agent Card.
Per A2A spec §8.3.3, ``capabilities.extensions[]`` entries with
``required: true`` must be activated on every request via the
``A2A-Extensions`` header (HTTP) or ``a2a-extensions`` metadata (gRPC).
Returns a list of extension URIs (may be empty).
"""
if not isinstance(agent_card, dict):
return []
caps = agent_card.get("capabilities")
if not isinstance(caps, dict):
return []
extensions = caps.get("extensions")
if not isinstance(extensions, list):
return []
seen = set()
required = []
for ext in extensions:
if isinstance(ext, dict) and ext.get("required") is True:
uri = ext.get("uri")
if isinstance(uri, str) and uri not in seen:
seen.add(uri)
required.append(uri)
return required| def __init__(self, base_url: str, transport: str, *, required_extensions: list[str] | None = None) -> None: | ||
| self.base_url = base_url | ||
| self.transport = transport |
There was a problem hiding this comment.
To avoid duplicating the initialization of required_extensions across all three transport client subclasses (GrpcClient, HttpJsonClient, and JsonRpcClient), we can store it directly in the base class BaseTransportClient as self.required_extensions.
| def __init__(self, base_url: str, transport: str, *, required_extensions: list[str] | None = None) -> None: | |
| self.base_url = base_url | |
| self.transport = transport | |
| def __init__(self, base_url: str, transport: str, *, required_extensions: list[str] | None = None) -> None: | |
| self.base_url = base_url | |
| self.transport = transport | |
| self.required_extensions: list[str] = required_extensions or [] |
| def __init__(self, base_url: str, *, required_extensions: list[str] | None = None) -> None: | ||
| super().__init__(base_url, TRANSPORT) | ||
| self._required_extensions: list[str] = required_extensions or [] | ||
| default_headers: dict[str, str] = {A2A_VERSION_HEADER: A2A_VERSION} | ||
| if self._required_extensions: | ||
| default_headers[A2A_EXTENSIONS_HEADER] = ",".join(self._required_extensions) | ||
| self._client = httpx.Client( | ||
| base_url=base_url, | ||
| timeout=httpx.Timeout(5.0, read=30.0), | ||
| headers={A2A_VERSION_HEADER: A2A_VERSION}, | ||
| headers=default_headers, | ||
| ) |
There was a problem hiding this comment.
We can pass required_extensions to super().__init__() and use the base class attribute self.required_extensions to avoid duplicate initialization.
def __init__(self, base_url: str, *, required_extensions: list[str] | None = None) -> None:
super().__init__(base_url, TRANSPORT, required_extensions=required_extensions)
default_headers: dict[str, str] = {A2A_VERSION_HEADER: A2A_VERSION}
if self.required_extensions:
default_headers[A2A_EXTENSIONS_HEADER] = ",".join(self.required_extensions)
self._client = httpx.Client(
base_url=base_url,
timeout=httpx.Timeout(5.0, read=30.0),
headers=default_headers,
)| def __init__(self, base_url: str, *, required_extensions: list[str] | None = None) -> None: | ||
| super().__init__(base_url, TRANSPORT) | ||
| self._required_extensions: list[str] = required_extensions or [] | ||
| self._id_counter = itertools.count(1) | ||
| default_headers: dict[str, str] = {A2A_VERSION_HEADER: A2A_VERSION} | ||
| if self._required_extensions: | ||
| default_headers[A2A_EXTENSIONS_HEADER] = ",".join(self._required_extensions) | ||
| self._client = httpx.Client( | ||
| base_url=base_url, | ||
| timeout=httpx.Timeout(5.0, read=30.0), | ||
| headers={A2A_VERSION_HEADER: A2A_VERSION}, | ||
| headers=default_headers, | ||
| ) |
There was a problem hiding this comment.
We can pass required_extensions to super().__init__() and use the base class attribute self.required_extensions to avoid duplicate initialization.
def __init__(self, base_url: str, *, required_extensions: list[str] | None = None) -> None:
super().__init__(base_url, TRANSPORT, required_extensions=required_extensions)
self._id_counter = itertools.count(1)
default_headers: dict[str, str] = {A2A_VERSION_HEADER: A2A_VERSION}
if self.required_extensions:
default_headers[A2A_EXTENSIONS_HEADER] = ",".join(self.required_extensions)
self._client = httpx.Client(
base_url=base_url,
timeout=httpx.Timeout(5.0, read=30.0),
headers=default_headers,
)1 participant
What
When a SUT declares required extensions in its Agent Card (
capabilities.extensions[]withrequired: true), the TCK transport clients now automatically include theA2A-Extensionsheader (HTTP) ora2a-extensionsmetadata (gRPC) on every request.Before: transport clients only sent
A2A-Version. SUTs enforcing required extensions returnedExtensionSupportRequiredErroron every positive test — breaking CORE-SEND-, CORE-STREAM-, GRPC-ERR-*, and related tests.After: required extensions are activated automatically. SUTs that declare and enforce extensions can pass the full conformance suite.
Changes
tck/transport/_helpers.py: addget_required_extensions()andA2A_EXTENSIONS_HEADERconstanttck/transport/base.py: acceptrequired_extensionskeyword inBaseTransportClient.__init__()tck/transport/http_json_client.py: includeA2A-Extensionsin defaulthttpx.Clientheaderstck/transport/jsonrpc_client.py: same treatmenttck/transport/grpc_client.py: replace static_METADATAwith dynamic_metadataproperty that appendsa2a-extensionstupletests/compatibility/conftest.py: extract extensions fromagent_cardfixture and pass to all transport client constructorsDesign decisions
required_extensions=None, preserving backward compatibility for direct instantiation without extensions.test_error_handling.pyuses raw_jsonrpc_call/_rest_callhelpers that bypass transport clients entirely.A2A-Extensions: uri1,uri2).Test results
This fixes #193