fix: correctly handle streaming errors

Author: lkawka

src/a2a/client/transports/jsonrpc.py

@@ -174,13 +174,16 @@ async def send_message_streaming(
             **modified_kwargs,
         ) as event_source:
             try:
+                event_source.response.raise_for_status()
                 async for sse in event_source.aiter_sse():
                     response = SendStreamingMessageResponse.model_validate(
                         json.loads(sse.data)
                     )
                     if isinstance(response.root, JSONRPCErrorResponse):
                         raise A2AClientJSONRPCError(response.root)
                     yield response.root.result
+            except httpx.HTTPStatusError as e:
+                raise A2AClientHTTPError(e.response.status_code, str(e)) from e
             except SSEError as e:
                 raise A2AClientHTTPError(
                     400, f'Invalid SSE response or protocol error: {e}'

src/a2a/client/transports/rest.py

@@ -152,10 +152,13 @@ async def send_message_streaming(
             **modified_kwargs,
         ) as event_source:
             try:
+                event_source.response.raise_for_status()
                 async for sse in event_source.aiter_sse():
                     event = a2a_pb2.StreamResponse()
                     Parse(sse.data, event)
                     yield proto_utils.FromProto.stream_response(event)
+            except httpx.HTTPStatusError as e:
+                raise A2AClientHTTPError(e.response.status_code, str(e)) from e
             except SSEError as e:
                 raise A2AClientHTTPError(
                     400, f'Invalid SSE response or protocol error: {e}'

tests/client/transports/test_jsonrpc_client.py

@@ -880,6 +880,44 @@ async def test_send_message_streaming_with_new_extensions(
             },
         )
 
+    @pytest.mark.asyncio
+    @patch('a2a.client.transports.jsonrpc.aconnect_sse')
+    async def test_send_message_streaming_server_error_propagates(
+        self,
+        mock_aconnect_sse: AsyncMock,
+        mock_httpx_client: AsyncMock,
+        mock_agent_card: MagicMock,
+    ):
+        """Test that send_message_streaming propagates server errors (e.g., 403, 500) directly."""
+        client = JsonRpcTransport(
+            httpx_client=mock_httpx_client,
+            agent_card=mock_agent_card,
+        )
+        params = MessageSendParams(
+            message=create_text_message_object(content='Error stream')
+        )
+
+        mock_event_source = AsyncMock(spec=EventSource)
+        mock_response = MagicMock(spec=httpx.Response)
+        mock_response.status_code = 403
+        mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
+            'Forbidden',
+            request=httpx.Request('POST', 'http://test.url'),
+            response=mock_response,
+        )
+        mock_event_source.response = mock_response
+        mock_event_source.aiter_sse.return_value = async_iterable_from_list([])
+        mock_aconnect_sse.return_value.__aenter__.return_value = (
+            mock_event_source
+        )
+
+        with pytest.raises(A2AClientHTTPError) as exc_info:
+            async for _ in client.send_message_streaming(request=params):
+                pass
+
+        assert exc_info.value.status_code == 403
+        mock_aconnect_sse.assert_called_once()
+
     @pytest.mark.asyncio
     async def test_get_card_no_card_provided_with_extensions(
         self, mock_httpx_client: AsyncMock

tests/client/transports/test_rest_client.py

@@ -7,14 +7,13 @@
 from httpx_sse import EventSource, ServerSentEvent
 
 from a2a.client import create_text_message_object
+from a2a.client.errors import A2AClientHTTPError
 from a2a.client.transports.rest import RestTransport
 from a2a.extensions.common import HTTP_EXTENSION_HEADER
 from a2a.types import (
     AgentCapabilities,
     AgentCard,
-    AgentSkill,
     MessageSendParams,
-    Role,
 )
 
 
@@ -130,6 +129,45 @@ async def test_send_message_streaming_with_new_extensions(
             },
         )
 
+    @pytest.mark.asyncio
+    @patch('a2a.client.transports.rest.aconnect_sse')
+    async def test_send_message_streaming_server_error_propagates(
+        self,
+        mock_aconnect_sse: AsyncMock,
+        mock_httpx_client: AsyncMock,
+        mock_agent_card: MagicMock,
+    ):
+        """Test that send_message_streaming propagates server errors (e.g., 403, 500) directly."""
+        client = RestTransport(
+            httpx_client=mock_httpx_client,
+            agent_card=mock_agent_card,
+        )
+        params = MessageSendParams(
+            message=create_text_message_object(content='Error stream')
+        )
+
+        mock_event_source = AsyncMock(spec=EventSource)
+        mock_response = MagicMock(spec=httpx.Response)
+        mock_response.status_code = 403
+        mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
+            'Forbidden',
+            request=httpx.Request('POST', 'http://test.url'),
+            response=mock_response,
+        )
+        mock_event_source.response = mock_response
+        mock_event_source.aiter_sse.return_value = async_iterable_from_list([])
+        mock_aconnect_sse.return_value.__aenter__.return_value = (
+            mock_event_source
+        )
+
+        with pytest.raises(A2AClientHTTPError) as exc_info:
+            async for _ in client.send_message_streaming(request=params):
+                pass
+
+        assert exc_info.value.status_code == 403
+
+        mock_aconnect_sse.assert_called_once()
+
     @pytest.mark.asyncio
     async def test_get_card_no_card_provided_with_extensions(
         self, mock_httpx_client: AsyncMock