fix(client): add extra_params option to RestTransport for provider-specific query params (Fixes #1089)

src/a2a/client/client.py

@@ -74,6 +74,15 @@ class ClientConfig:
     push_notification_config: TaskPushNotificationConfig | None = None
     """Push notification configuration to use for every request."""
 
+    extra_params: dict[str, str] | None = None
+    """Optional query parameters to append to every request URL.
+
+    Useful for provider-specific requirements, such as
+    ``{'alt': 'sse'}`` for Google A2A API streaming endpoints where
+    the transcoding layer wraps SSE streams inside JSON arrays by
+    default without this parameter.
+    """
+
 
 class ClientCallContext(BaseModel):
     """A context passed with each client call, allowing for call-specific.

src/a2a/client/client_factory.py

@@ -38,13 +38,13 @@
 try:
     from a2a.client.transports.grpc import GrpcTransport
 except ImportError:
     GrpcTransport = None  # type: ignore # pyright: ignore
 
 
 try:
     from a2a.compat.v0_3.grpc_transport import CompatGrpcTransport
 except ImportError:
     CompatGrpcTransport = None  # type: ignore # pyright: ignore
 
 logger = logging.getLogger(__name__)
 
@@ -157,12 +157,14 @@
                         self._httpx_client,
                         card,
                         url,
+                        extra_params=config.extra_params,
                     )
 
                 return RestTransport(
                     self._httpx_client,
                     card,
                     url,
+                    extra_params=config.extra_params,
                 )
 
             self.register(

src/a2a/client/transports/rest.py

@@ -84,18 +84,29 @@ def _parse_rest_error(
 
 @trace_class(kind=SpanKind.CLIENT)
 class RestTransport(ClientTransport):
-    """A REST transport for the A2A client."""
+    """A REST transport for the A2A client.
+
+    Args:
+        httpx_client: The async HTTP client to use for requests.
+        agent_card: The AgentCard describing the remote agent.
+        url: The base URL of the agent.
+        extra_params: Optional query parameters to append to every
+            request URL. Useful for provider-specific requirements, such
+            as ``{'alt': 'sse'}`` for Google A2A API streaming endpoints.
+    """
 
     def __init__(
         self,
         httpx_client: httpx.AsyncClient,
         agent_card: AgentCard,
         url: str,
+        extra_params: dict[str, str] | None = None,
     ):
         """Initializes the RestTransport."""
         self.url = url.removesuffix('/')
         self.httpx_client = httpx_client
         self.agent_card = agent_card
+        self._extra_params = extra_params or {}
 
     async def send_message(
         self,
@@ -376,6 +387,7 @@ async def _send_stream_request(
             self._handle_http_error,
             self._handle_sse_error,
             json=json,
+            params=self._extra_params or None,
             **http_kwargs,
         ):
             event: StreamResponse = Parse(sse_data, StreamResponse())

src/a2a/compat/v0_3/rest_transport.py

@@ -58,21 +58,33 @@
 
 @trace_class(kind=SpanKind.CLIENT)
 class CompatRestTransport(ClientTransport):
-    """A backward compatible REST transport for A2A v0.3."""
+    """A backward compatible REST transport for A2A v0.3.
+
+    Args:
+        httpx_client: The async HTTP client to use for requests.
+        agent_card: The AgentCard describing the remote agent.
+        url: The base URL of the agent.
+        subscribe_method_override: Override for the subscribe method.
+        extra_params: Optional query parameters to append to every
+            request URL. Useful for provider-specific requirements, such
+            as ``{'alt': 'sse'}`` for Google A2A API streaming endpoints.
+    """
 
     def __init__(
         self,
         httpx_client: httpx.AsyncClient,
         agent_card: AgentCard | None,
         url: str,
         subscribe_method_override: str | None = None,
+        extra_params: dict[str, str] | None = None,
     ):
         """Initializes the CompatRestTransport."""
         self.url = url.removesuffix('/')
         self.httpx_client = httpx_client
         self.agent_card = agent_card
         self._subscribe_method_override = subscribe_method_override
         self._subscribe_auto_method_override = subscribe_method_override is None
+        self._extra_params = extra_params or {}
 
     async def send_message(
         self,
@@ -389,6 +401,7 @@ async def _send_stream_request(
             f'{self.url}{path}',
             self._handle_http_error,
             json=json,
+            params=self._extra_params or None,
             **http_kwargs,
         ):
             event_proto = a2a_v0_3_pb2.StreamResponse()

tests/client/transports/test_rest_client.py

@@ -738,3 +738,123 @@ async def empty_aiter():
 
         # url is 3rd positional argument in aconnect_sse(client, method, url, ...)
         assert args[2] == f'http://agent.example.com/api{expected_path}'
+
+    @pytest.mark.asyncio
+    @patch('a2a.client.transports.http_helpers._SSEEventSource')
+    async def test_extra_params_passed_to_stream_request(
+        self,
+        mock_aconnect_sse: AsyncMock,
+        mock_httpx_client: AsyncMock,
+        mock_agent_card: MagicMock,
+    ) -> None:
+        """Verify extra_params are passed as params kwarg in stream requests."""
+        extra_params = {'alt': 'sse'}
+        client = RestTransport(
+            httpx_client=mock_httpx_client,
+            agent_card=mock_agent_card,
+            url='http://agent.example.com/api',
+            extra_params=extra_params,
+        )
+        request = SendMessageRequest(
+            message=new_text_message(text='Hello stream'),
+        )
+        mock_event_source = AsyncMock(spec=EventSource)
+        mock_event_source.response = MagicMock(spec=httpx.Response)
+        mock_event_source.response.headers = {
+            'content-type': 'text/event-stream'
+        }
+        mock_event_source.response.raise_for_status.return_value = None
+
+        async def empty_aiter() -> AsyncGenerator:
+            if False:
+                yield
+
+        mock_event_source.aiter_sse.return_value = empty_aiter()
+        mock_aconnect_sse.return_value.__aenter__.return_value = (
+            mock_event_source
+        )
+
+        async for _ in client.send_message_streaming(request=request):
+            pass
+
+        _, kwargs = mock_aconnect_sse.call_args
+        assert kwargs.get('params') == extra_params
+
+    @pytest.mark.asyncio
+    @patch('a2a.client.transports.http_helpers._SSEEventSource')
+    async def test_extra_params_none_does_not_break(
+        self,
+        mock_aconnect_sse: AsyncMock,
+        mock_httpx_client: AsyncMock,
+        mock_agent_card: MagicMock,
+    ) -> None:
+        """Verify that extra_params=None (default) does not add params."""
+        client = RestTransport(
+            httpx_client=mock_httpx_client,
+            agent_card=mock_agent_card,
+            url='http://agent.example.com/api',
+        )
+        request = SendMessageRequest(
+            message=new_text_message(text='Hello stream'),
+        )
+        mock_event_source = AsyncMock(spec=EventSource)
+        mock_event_source.response = MagicMock(spec=httpx.Response)
+        mock_event_source.response.headers = {
+            'content-type': 'text/event-stream'
+        }
+        mock_event_source.response.raise_for_status.return_value = None
+
+        async def empty_aiter() -> AsyncGenerator:
+            if False:
+                yield
+
+        mock_event_source.aiter_sse.return_value = empty_aiter()
+        mock_aconnect_sse.return_value.__aenter__.return_value = (
+            mock_event_source
+        )
+
+        async for _ in client.send_message_streaming(request=request):
+            pass
+
+        _, kwargs = mock_aconnect_sse.call_args
+        assert kwargs.get('params') is None
+
+    @pytest.mark.asyncio
+    @patch('a2a.client.transports.http_helpers._SSEEventSource')
+    async def test_extra_params_empty_does_not_add_params(
+        self,
+        mock_aconnect_sse: AsyncMock,
+        mock_httpx_client: AsyncMock,
+        mock_agent_card: MagicMock,
+    ) -> None:
+        """Verify that empty extra_params={} does not add params."""
+        client = RestTransport(
+            httpx_client=mock_httpx_client,
+            agent_card=mock_agent_card,
+            url='http://agent.example.com/api',
+            extra_params={},
+        )
+        request = SendMessageRequest(
+            message=new_text_message(text='Hello stream'),
+        )
+        mock_event_source = AsyncMock(spec=EventSource)
+        mock_event_source.response = MagicMock(spec=httpx.Response)
+        mock_event_source.response.headers = {
+            'content-type': 'text/event-stream'
+        }
+        mock_event_source.response.raise_for_status.return_value = None
+
+        async def empty_aiter() -> AsyncGenerator:
+            if False:
+                yield
+
+        mock_event_source.aiter_sse.return_value = empty_aiter()
+        mock_aconnect_sse.return_value.__aenter__.return_value = (
+            mock_event_source
+        )
+
+        async for _ in client.send_message_streaming(request=request):
+            pass
+
+        _, kwargs = mock_aconnect_sse.call_args
+        assert kwargs.get('params') is None

tests/compat/v0_3/test_rest_transport.py

@@ -635,6 +635,7 @@ async def mock_generator(*args, **kwargs):
         'http://example.com/test',
         transport._handle_http_error,
         json=None,
+        params=None,
         headers={'a2a-version': '0.3'},
     )
 
@@ -664,3 +665,37 @@ async def test_compat_rest_transport_execute_request(
     mock_send_http_request.assert_called_once_with(
         transport.httpx_client, mock_request, transport._handle_http_error
     )
+
+
+@pytest.mark.asyncio
+@patch('a2a.compat.v0_3.rest_transport.send_http_stream_request')
+async def test_compat_rest_transport_extra_params(
+    mock_send_http_stream_request: AsyncMock,
+    mock_httpx_client: AsyncMock,
+    agent_card: AgentCard,
+) -> None:
+    """Verify extra_params are passed to stream requests in CompatRestTransport."""
+    transport = CompatRestTransport(
+        httpx_client=mock_httpx_client,
+        agent_card=agent_card,
+        url='http://example.com',
+        extra_params={'alt': 'sse'},
+    )
+    request = SendMessageRequest(
+        message=Message(message_id='msg-1', role=Role.ROLE_USER),
+    )
+
+    async def mock_generator():
+        yield b'{"task": {"id": "task-123"}}'
+
+    mock_send_http_stream_request.return_value = mock_generator()
+
+    events = [
+        event
+        async for event in transport.send_message_streaming(request=request)
+    ]
+    assert len(events) == 1
+
+    mock_send_http_stream_request.assert_called_once()
+    _, kwargs = mock_send_http_stream_request.call_args_list[0]
+    assert kwargs.get('params') == {'alt': 'sse'}