Python A2A: Expose supported_protocol_bindings as configurable parameter (#6098)

* Expose supported_protocol_bindings as configurable parameter on A2AAgent

Add supported_protocol_bindings parameter to A2AAgent.__init__() allowing
users to configure which A2A protocol bindings (JSONRPC, GRPC, HTTP+JSON)
the client prefers when connecting to remote agents.

- Defaults to ["JSONRPC"] matching current behavior
- Passes through to ClientConfig for transport negotiation
- Replaces 4 hardcoded references with the configurable value

Closes #6057

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Fix empty list falsy trap and add fallback path test coverage

- Use 'is not None' check instead of 'or' to preserve explicit empty list
- Add test verifying empty list is not silently replaced with defaults
- Add test verifying fallback path uses custom bindings

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Document known protocol binding values in docstring

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Use Literal union for protocol binding type hint

Provides IDE autocomplete for known values while keeping the type
open for custom bindings (Literal is str at runtime).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Giles Odigwe
2026-05-28 12:05:13 -07:00
committed by GitHub
Unverified
parent d2f79930d5
commit e9a606344a
2 changed files with 97 additions and 5 deletions
@@ -176,6 +176,7 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
http_client: httpx.AsyncClient | None = None,
auth_interceptor: AuthInterceptor | None = None,
timeout: float | httpx.Timeout | None = None,
supported_protocol_bindings: list[Literal["JSONRPC", "GRPC", "HTTP+JSON"] | str] | None = None,
**kwargs: Any,
) -> None:
"""Initialize the A2AAgent.
@@ -193,6 +194,9 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
timeout: Request timeout configuration. Can be a float (applied to all timeout components),
httpx.Timeout object (for full control), or None (uses 10.0s connect, 60.0s read,
10.0s write, 5.0s pool - optimized for A2A operations).
supported_protocol_bindings: List of protocol bindings to use for transport negotiation.
Known values: "JSONRPC", "GRPC", "HTTP+JSON". Defaults to ["JSONRPC"].
The A2A spec treats this as an open-form string, so custom bindings are also accepted.
kwargs: any additional properties, passed to BaseAgent.
"""
# Default name/description from agent_card when not explicitly provided
@@ -205,6 +209,7 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
super().__init__(id=id, name=name, description=description, **kwargs)
self._http_client: httpx.AsyncClient | None = http_client
self._timeout_config = self._create_timeout_config(timeout)
bindings = supported_protocol_bindings if supported_protocol_bindings is not None else ["JSONRPC"]
if client is not None:
self.client = client
self._non_streaming_client: Client | None = None
@@ -214,7 +219,7 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
if url is None:
raise ValueError("Either agent_card or url must be provided")
# Create minimal agent card from URL
agent_card = minimal_agent_card(url, ["JSONRPC"])
agent_card = minimal_agent_card(url, bindings)
# Create or use provided httpx client
if http_client is None:
@@ -229,13 +234,13 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
streaming_config = ClientConfig(
httpx_client=http_client,
streaming=True,
supported_protocol_bindings=["JSONRPC"],
supported_protocol_bindings=bindings,
)
# Create non-streaming client (single request/response for stream=False)
non_streaming_config = ClientConfig(
httpx_client=http_client,
streaming=False,
supported_protocol_bindings=["JSONRPC"],
supported_protocol_bindings=bindings,
)
streaming_factory = ClientFactory(streaming_config)
non_streaming_factory = ClientFactory(non_streaming_config)
@@ -256,7 +261,7 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
"Provide a 'url' argument or ensure 'agent_card.supported_interfaces' "
"contains at least one interface with a URL."
) from transport_error
fallback_card = minimal_agent_card(fallback_url, ["JSONRPC"])
fallback_card = minimal_agent_card(fallback_url, bindings)
try:
self.client = streaming_factory.create(fallback_card, interceptors=interceptors) # type: ignore
self._non_streaming_client = non_streaming_factory.create(