Python A2A: Expose supported_protocol_bindings as configurable parameter

python/packages/a2a/agent_framework_a2a/_agent.py

@@ -98,6 +98,7 @@ def __init__(
         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.
@@ -115,6 +116,9 @@ def __init__(
             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
@@ -127,6 +131,7 @@ def __init__(
         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
@@ -136,7 +141,7 @@ def __init__(
             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:
@@ -151,13 +156,13 @@ def __init__(
         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)
@@ -178,7 +183,7 @@ def __init__(
                     "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(

python/packages/a2a/tests/test_a2a_agent.py

@@ -701,7 +701,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: