refactor: use ClientCallContext for HTTP arguments in stream requests, convert rpc_request.data to dict, and adjust client blocking configuration logic.

Author: guglielmo-san

src/a2a/client/base_client.py

@@ -94,9 +94,9 @@ async def send_message(
         async for client_event in self._process_stream(stream):
             yield client_event
 
-    def _apply_client_config(self, request: SendMessageRequest):
+    def _apply_client_config(self, request: SendMessageRequest) -> None:
         if not request.configuration.blocking and self._config.polling:
-            request.configuration.blocking = self._config.polling
+            request.configuration.blocking = not self._config.polling
         if (
             not request.configuration.HasField('push_notification_config')
             and self._config.push_notification_configs

src/a2a/client/transports/jsonrpc.py

@@ -77,7 +77,9 @@ async def send_message(
             params=json_format.MessageToDict(request),
             _id=str(uuid4()),
         )
-        response_data = await self._send_request(rpc_request.data, context)
+        response_data = await self._send_request(
+            dict(rpc_request.data), context
+        )
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise self._create_jsonrpc_error(json_rpc_response.error)
@@ -99,7 +101,7 @@ async def send_message_streaming(
             _id=str(uuid4()),
         )
         async for event in self._send_stream_request(
-            rpc_request.data,
+            dict(rpc_request.data),
             context,
         ):
             yield event
@@ -116,7 +118,9 @@ async def get_task(
             params=json_format.MessageToDict(request),
             _id=str(uuid4()),
         )
-        response_data = await self._send_request(rpc_request.data, context)
+        response_data = await self._send_request(
+            dict(rpc_request.data), context
+        )
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise self._create_jsonrpc_error(json_rpc_response.error)
@@ -135,7 +139,9 @@ async def list_tasks(
             params=json_format.MessageToDict(request),
             _id=str(uuid4()),
         )
-        response_data = await self._send_request(rpc_request.data, context)
+        response_data = await self._send_request(
+            dict(rpc_request.data), context
+        )
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise self._create_jsonrpc_error(json_rpc_response.error)
@@ -156,7 +162,9 @@ async def cancel_task(
             params=json_format.MessageToDict(request),
             _id=str(uuid4()),
         )
-        response_data = await self._send_request(rpc_request.data, context)
+        response_data = await self._send_request(
+            dict(rpc_request.data), context
+        )
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise self._create_jsonrpc_error(json_rpc_response.error)
@@ -175,7 +183,9 @@ async def create_task_push_notification_config(
             params=json_format.MessageToDict(request),
             _id=str(uuid4()),
         )
-        response_data = await self._send_request(rpc_request.data, context)
+        response_data = await self._send_request(
+            dict(rpc_request.data), context
+        )
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise self._create_jsonrpc_error(json_rpc_response.error)
@@ -196,7 +206,9 @@ async def get_task_push_notification_config(
             params=json_format.MessageToDict(request),
             _id=str(uuid4()),
         )
-        response_data = await self._send_request(rpc_request.data, context)
+        response_data = await self._send_request(
+            dict(rpc_request.data), context
+        )
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise self._create_jsonrpc_error(json_rpc_response.error)
@@ -217,7 +229,9 @@ async def list_task_push_notification_configs(
             params=json_format.MessageToDict(request),
             _id=str(uuid4()),
         )
-        response_data = await self._send_request(rpc_request.data, context)
+        response_data = await self._send_request(
+            dict(rpc_request.data), context
+        )
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise self._create_jsonrpc_error(json_rpc_response.error)
@@ -241,7 +255,9 @@ async def delete_task_push_notification_config(
             params=json_format.MessageToDict(request),
             _id=str(uuid4()),
         )
-        response_data = await self._send_request(rpc_request.data, context)
+        response_data = await self._send_request(
+            dict(rpc_request.data), context
+        )
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise self._create_jsonrpc_error(json_rpc_response.error)
@@ -259,7 +275,7 @@ async def subscribe(
             _id=str(uuid4()),
         )
         async for event in self._send_stream_request(
-            rpc_request.data,
+            dict(rpc_request.data),
             context,
         ):
             yield event
@@ -283,7 +299,7 @@ async def get_extended_agent_card(
             _id=str(uuid4()),
         )
         response_data = await self._send_request(
-            rpc_request.data,
+            dict(rpc_request.data),
             context,
         )
         json_rpc_response = JSONRPC20Response(**response_data)

src/a2a/client/transports/rest.py

@@ -337,14 +337,12 @@ async def _send_stream_request(
         method: str,
         target: str,
         tenant: str,
-        http_kwargs: dict[str, Any] | None = None,
+        context: ClientCallContext | None = None,
         **kwargs: Any,
     ) -> AsyncGenerator[StreamResponse]:
-        final_kwargs = dict(http_kwargs or {})
-        final_kwargs.update(kwargs)
-        headers = dict(self.httpx_client.headers.items())
-        headers.update(final_kwargs.get('headers', {}))
-        final_kwargs['headers'] = headers
+        http_kwargs = self._get_http_args(context)
+        headers = http_kwargs.get('headers')
+        timeout = http_kwargs.get('timeout', httpx.USE_CLIENT_DEFAULT)
 
         path = self._get_path(target, tenant)
 
@@ -353,7 +351,9 @@ async def _send_stream_request(
             method,
             f'{self.url}{path}',
             self._handle_http_error,
-            **final_kwargs,
+            headers=headers,
+            timeout=timeout,
+            **kwargs,
         ):
             event: StreamResponse = Parse(sse_data, StreamResponse())
             yield event