Repository: a2aproject/a2a-python

Issue:
[Bug]: A2A Client incorrectly handles HTTP error responses when using streaming mode
### What happened?

## Summary                                                                                          

The client side code incorrectly handles HTTP error responses when using streaming mode. Instead of
surfacing the actual HTTP status code and error message, the SDK attempts to parse error         
responses as Server-Sent Events (SSE) and reports an SSE protocol error.

For example, an authentication failure (401) gives this exception:

```
HTTP Error 400: Invalid SSE response or protocol error: Expected response header Content-Type to 
contain 'text/event-stream', got 'application/json'    
```

## Environment

 • Component: A2A Python SDK  (0.3.7)
 • Affected Module: a2a.client.base_client
 • Transport: JSON-RPC over HTTP with SSE streaming

## Expected Behavior                                                                                

When an A2A agent returns an HTTP error response (e.g., 500, 503, 401, 404):                     

 1 SDK detects the non-2xx HTTP status code                                                      
 2 SDK reads the error response body                                                             
 3 SDK raises A2AClientHTTPError with the actual status code and error message                   

### Example:                                                                                         

```
A2AClientHTTPError: HTTP 500 - Internal Server Error: Database connection failed                 
```


## Actual Behavior

The SDK attempts to parse all responses as SSE streams, regardless of HTTP status code:          

 1 SDK tries to parse response as text/event-stream                                              
 2 SDK encounters Content-Type: application/json (typical for error responses)                   
 3 SDK raises an SSE protocol error with status 400                                              
 4 The actual HTTP status code is lost                                                           

### Example:                                                                                         

```
HTTP Error 400: Invalid SSE response or protocol error: Expected response header Content-Type to 
contain 'text/event-stream', got 'application/json'                                              
```

## Impact                                                                                           

 • Lost Error Information: Actual HTTP status codes and error messages are completely lost       
 • Misleading Errors: All HTTP errors appear as "400 Bad Request"
 • Can't properly deal with expired tokens for oauth
 • Broken Retry Logic: Cannot distinguish between transient (503) and permanent (500) errors     
 • Difficult Debugging: Impossible to diagnose downstream agent failures                         

## Affected Scenarios                                                                               

This affects all HTTP error responses in streaming mode, including:                              

 • 500 Internal Server Error                                                                     
 • 503 Service Unavailable                                                                       
 • 401 Unauthorized                                                                              
 • 404 Not Found                                                                                 
 • Any other 4xx or 5xx status code          

### Relevant log output

```shell

```

### Code of Conduct

- [x] I agree to follow this project's Code of Conduct

Discussion:

Fix (patch):
diff --git a/src/a2a/client/transports/jsonrpc.py b/src/a2a/client/transports/jsonrpc.py
index 6cce1eff..32cf74f2 100644
--- a/src/a2a/client/transports/jsonrpc.py
+++ b/src/a2a/client/transports/jsonrpc.py
@@ -174,6 +174,7 @@ 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)
@@ -181,6 +182,8 @@ async def send_message_streaming(
                     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}'
diff --git a/src/a2a/client/transports/rest.py b/src/a2a/client/transports/rest.py
index 948f3f35..bdfcc8ba 100644
--- a/src/a2a/client/transports/rest.py
+++ b/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}'


Test changes:
diff --git a/tests/client/transports/test_jsonrpc_client.py b/tests/client/transports/test_jsonrpc_client.py
index d9dbafc8..edbcd6c7 100644
--- a/tests/client/transports/test_jsonrpc_client.py
+++ b/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
diff --git a/tests/client/transports/test_rest_client.py b/tests/client/transports/test_rest_client.py
index 49d20d9d..cd68b443 100644
--- a/tests/client/transports/test_rest_client.py
+++ b/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