-
Notifications
You must be signed in to change notification settings - Fork 269
Description
Is your feature request related to a problem? Please describe.
Right now if I want to support an A2A extension in the Client I have to create a custom interceptor. Here's what I've resolved to doing:
from typing import Any
from a2a.client import ClientCallContext, ClientCallInterceptor
from a2a.extensions.common import HTTP_EXTENSION_HEADER
from a2a.types import AgentCard
class ExtensionsInterceptor(ClientCallInterceptor):
def __init__(self, extensions: list[str]):
"""
Args:
extensions: List of extension URIs to signal support for.
"""
self.extensions = extensions
async def intercept(
self,
method_name: str,
request_payload: dict[str, Any],
http_kwargs: dict[str, Any],
agent_card: AgentCard | None,
context: ClientCallContext | None,
) -> tuple[dict[str, Any], dict[str, Any]]:
"""Add extension header to HTTP request."""
if self.extensions:
headers = http_kwargs.get("headers", {})
existing_extensions = headers.get(HTTP_EXTENSION_HEADER, "")
split = existing_extensions.split(", ") if existing_extensions else []
headers[HTTP_EXTENSION_HEADER] = ", ".join(self.extensions + split)
http_kwargs["headers"] = headers
return request_payload, http_kwargsand then set that interceptor when creating a Client
Describe the solution you'd like
Ideally extension support would be first class in the Client. Maybe add it to ClientConfig like:
class ClientConfig:
...
extensions: list[str]Describe alternatives you've considered
Using the interceptor described above.
Additional context
This is required for the A2A <> x402 integration.
Code of Conduct
- I agree to follow this project's Code of Conduct