From e9a606344adbe5e41d9376b1f8508da593ea6c3b Mon Sep 17 00:00:00 2001 From: Giles Odigwe <79032838+giles17@users.noreply.github.com> Date: Thu, 28 May 2026 12:05:13 -0700 Subject: [PATCH] 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> --- .../a2a/agent_framework_a2a/_agent.py | 13 ++- python/packages/a2a/tests/test_a2a_agent.py | 89 ++++++++++++++++++- 2 files changed, 97 insertions(+), 5 deletions(-) diff --git a/python/packages/a2a/agent_framework_a2a/_agent.py b/python/packages/a2a/agent_framework_a2a/_agent.py index 8d6a0496c4..c620176779 100644 --- a/python/packages/a2a/agent_framework_a2a/_agent.py +++ b/python/packages/a2a/agent_framework_a2a/_agent.py @@ -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( diff --git a/python/packages/a2a/tests/test_a2a_agent.py b/python/packages/a2a/tests/test_a2a_agent.py index 37c1efedee..0ff5978c87 100644 --- a/python/packages/a2a/tests/test_a2a_agent.py +++ b/python/packages/a2a/tests/test_a2a_agent.py @@ -703,7 +703,94 @@ def test_a2a_agent_initialization_with_timeout_parameter() -> None: assert isinstance(timeout_arg, httpx.Timeout) -# region Continuation Token Tests +def test_a2a_agent_initialization_with_supported_protocol_bindings() -> None: + """Test A2AAgent initialization with custom supported_protocol_bindings.""" + with ( + patch("agent_framework_a2a._agent.httpx.AsyncClient") as mock_async_client, + patch("agent_framework_a2a._agent.ClientConfig") as mock_config, + patch("agent_framework_a2a._agent.ClientFactory") as mock_factory, + ): + mock_async_client.return_value = MagicMock() + mock_client_instance = MagicMock() + mock_factory.return_value.create.return_value = mock_client_instance + + A2AAgent( + name="Test Agent", + url="https://test-agent.example.com", + supported_protocol_bindings=["GRPC", "JSONRPC"], + ) + + # Verify ClientConfig was called with our custom bindings for both streaming and non-streaming + assert mock_config.call_count == 2 + for call in mock_config.call_args_list: + assert call.kwargs["supported_protocol_bindings"] == ["GRPC", "JSONRPC"] + + +def test_a2a_agent_initialization_defaults_to_jsonrpc() -> None: + """Test A2AAgent defaults to JSONRPC when supported_protocol_bindings is not provided.""" + with ( + patch("agent_framework_a2a._agent.httpx.AsyncClient") as mock_async_client, + patch("agent_framework_a2a._agent.ClientConfig") as mock_config, + patch("agent_framework_a2a._agent.ClientFactory") as mock_factory, + ): + mock_async_client.return_value = MagicMock() + mock_client_instance = MagicMock() + mock_factory.return_value.create.return_value = mock_client_instance + + A2AAgent(name="Test Agent", url="https://test-agent.example.com") + + # Verify ClientConfig was called with default JSONRPC bindings + assert mock_config.call_count == 2 + for call in mock_config.call_args_list: + assert call.kwargs["supported_protocol_bindings"] == ["JSONRPC"] + + +def test_a2a_agent_initialization_empty_list_preserved() -> None: + """Test that an explicit empty list is preserved and not replaced with defaults.""" + with ( + patch("agent_framework_a2a._agent.httpx.AsyncClient") as mock_async_client, + patch("agent_framework_a2a._agent.ClientConfig") as mock_config, + patch("agent_framework_a2a._agent.ClientFactory") as mock_factory, + ): + mock_async_client.return_value = MagicMock() + mock_client_instance = MagicMock() + mock_factory.return_value.create.return_value = mock_client_instance + + A2AAgent( + name="Test Agent", + url="https://test-agent.example.com", + supported_protocol_bindings=[], + ) + + # Verify ClientConfig was called with the explicit empty list, not the default + assert mock_config.call_count == 2 + for call in mock_config.call_args_list: + assert call.kwargs["supported_protocol_bindings"] == [] + + +def test_a2a_agent_fallback_uses_custom_bindings() -> None: + """Test that transport fallback path uses custom bindings.""" + mock_agent_card = MagicMock() + mock_agent_card.supported_interfaces = [MagicMock(url="https://fallback.example.com")] + + mock_factory = MagicMock() + # First create() call fails (primary streaming), then fallback calls succeed + primary_error = Exception("no compatible transports found") + mock_factory.create.side_effect = [primary_error, MagicMock(), MagicMock()] + + with ( + patch("agent_framework_a2a._agent.ClientFactory", return_value=mock_factory), + patch("agent_framework_a2a._agent.minimal_agent_card") as mock_minimal_card, + patch("agent_framework_a2a._agent.httpx.AsyncClient"), + ): + A2AAgent( + name="test-agent", + agent_card=mock_agent_card, + supported_protocol_bindings=["GRPC", "HTTP+JSON"], + ) + + # Verify minimal_agent_card was called with the custom bindings + mock_minimal_card.assert_called_once_with("https://fallback.example.com", ["GRPC", "HTTP+JSON"]) async def test_working_task_emits_continuation_token(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None: