fix: Improve streaming errors handling (#576)

# Description

Refine error management for the streaming operation. Previously, errors
were converted into stream parts, resulting in the loss of status info.
The updated logic now first verifies if the request was successful; if
it failed, a client error is returned, preserving the relevant status
information.

- [x] Follow the [`CONTRIBUTING`
Guide](https://github.com/a2aproject/a2a-python/blob/main/CONTRIBUTING.md).
- [x] Make your Pull Request title in the
<https://www.conventionalcommits.org/> specification.
- Important Prefixes for
[release-please](https://github.com/googleapis/release-please):
- `fix:` which represents bug fixes, and correlates to a
[SemVer](https://semver.org/) patch.
- `feat:` represents a new feature, and correlates to a SemVer minor.
- `feat!:`, or `fix!:`, `refactor!:`, etc., which represent a breaking
change (indicated by the `!`) and will result in a SemVer major.
- [x] Ensure the tests and linter pass (Run `bash scripts/format.sh`
from the repository root to format)
- [x] Appropriate docs were updated (if necessary)

Fixes #502  🦕

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