fix: fix error handling for gRPC and SSE streaming (#879)

Reproduced in `test_client_server_integration.py`.

### gRPC

`validate_async_generator` decorator was applied on top of the method
above A2A error handling. Compat handler was already refactored in a way
which made it possible to apply it on a nested function. It was done
there and v1 handler was refactored in the same way.

### SSE streaming

Iterator wrapped into `validate_async_generator` is assigned to
`EventSourceResponse` and is returned from the method, so when it throws
`rest_stream_error_handler` has no effect on it.


https://github.com/a2aproject/a2a-python/blob/4630efd0ca4bf6934a7d9215ef2a2986b6e6e73a/src/a2a/server/apps/rest/rest_adapter.py#L155-L163

Instead of throwing on the first iteration, throw on the method
invocation itself to avoid more sophisticated error handling (i.e.
reading one item to trigger error) by removing separate handling for
async generator.

Client-level handling is also updated to properly handle non-200 status
code for streaming and non-streaming response in case of JSON-RPC error.

Author: ishymko

src/a2a/client/transports/http_helpers.py

@@ -78,7 +78,22 @@ async def send_http_stream_request(
         async with aconnect_sse(
             httpx_client, method, url, **kwargs
         ) as event_source:
-            event_source.response.raise_for_status()
+            try:
+                event_source.response.raise_for_status()
+            except httpx.HTTPStatusError as e:
+                # Read upfront streaming error content immediately, otherwise lower-level handlers
+                # (e.g. response.json()) crash with 'ResponseNotRead' Access errors.
+                await event_source.response.aread()
+                raise e
+
+            # If the response is not a stream, read it standardly (e.g., upfront JSON-RPC error payload)
+            if 'text/event-stream' not in event_source.response.headers.get(
+                'content-type', ''
+            ):
+                content = await event_source.response.aread()
+                yield content.decode('utf-8')
+                return
+
             async for sse in event_source.aiter_sse():
                 if not sse.data:
                     continue

src/a2a/compat/v0_3/grpc_handler.py

@@ -29,7 +29,7 @@
 from a2a.server.request_handlers.request_handler import RequestHandler
 from a2a.types.a2a_pb2 import AgentCard
 from a2a.utils.errors import A2AError, InvalidParamsError
-from a2a.utils.helpers import maybe_await, validate, validate_async_generator
+from a2a.utils.helpers import maybe_await, validate
 
 
 logger = logging.getLogger(__name__)
@@ -170,17 +170,17 @@ async def _handler(
             context, _handler, a2a_v0_3_pb2.SendMessageResponse()
         )
 
-    @validate_async_generator(
-        lambda self: self.agent_card.capabilities.streaming,
-        'Streaming is not supported by the agent',
-    )
     async def SendStreamingMessage(
         self,
         request: a2a_v0_3_pb2.SendMessageRequest,
         context: grpc.aio.ServicerContext,
     ) -> AsyncIterable[a2a_v0_3_pb2.StreamResponse]:
         """Handles the 'SendStreamingMessage' gRPC method (v0.3)."""
 
+        @validate(
+            lambda _: self.agent_card.capabilities.streaming,
+            'Streaming is not supported by the agent',
+        )
         async def _handler(
             server_context: ServerCallContext,
         ) -> AsyncIterable[a2a_v0_3_pb2.StreamResponse]:
@@ -233,17 +233,17 @@ async def _handler(
 
         return await self._handle_unary(context, _handler, a2a_v0_3_pb2.Task())
 
-    @validate_async_generator(
-        lambda self: self.agent_card.capabilities.streaming,
-        'Streaming is not supported by the agent',
-    )
     async def TaskSubscription(
         self,
         request: a2a_v0_3_pb2.TaskSubscriptionRequest,
         context: grpc.aio.ServicerContext,
     ) -> AsyncIterable[a2a_v0_3_pb2.StreamResponse]:
         """Handles the 'TaskSubscription' gRPC method (v0.3)."""
 
+        @validate(
+            lambda _: self.agent_card.capabilities.streaming,
+            'Streaming is not supported by the agent',
+        )
         async def _handler(
             server_context: ServerCallContext,
         ) -> AsyncIterable[a2a_v0_3_pb2.StreamResponse]:
@@ -260,17 +260,17 @@ async def _handler(
         async for item in self._handle_stream(context, _handler):
             yield item
 
-    @validate(
-        lambda self: self.agent_card.capabilities.push_notifications,
-        'Push notifications are not supported by the agent',
-    )
     async def CreateTaskPushNotificationConfig(
         self,
         request: a2a_v0_3_pb2.CreateTaskPushNotificationConfigRequest,
         context: grpc.aio.ServicerContext,
     ) -> a2a_v0_3_pb2.TaskPushNotificationConfig:
         """Handles the 'CreateTaskPushNotificationConfig' gRPC method (v0.3)."""
 
+        @validate(
+            lambda _: self.agent_card.capabilities.push_notifications,
+            'Push notifications are not supported by the agent',
+        )
         async def _handler(
             server_context: ServerCallContext,
         ) -> a2a_v0_3_pb2.TaskPushNotificationConfig:

src/a2a/compat/v0_3/rest_handler.py

@@ -31,7 +31,6 @@
 from a2a.utils import constants
 from a2a.utils.helpers import (
     validate,
-    validate_async_generator,
     validate_version,
 )
 from a2a.utils.telemetry import SpanKind, trace_class
@@ -85,7 +84,7 @@ async def on_message_send(
         return MessageToDict(pb2_v03_resp)
 
     @validate_version(constants.PROTOCOL_VERSION_0_3)
-    @validate_async_generator(
+    @validate(
         lambda self: self.agent_card.capabilities.streaming,
         'Streaming is not supported by the agent',
     )
@@ -143,7 +142,7 @@ async def on_cancel_task(
         return MessageToDict(pb2_v03_task)
 
     @validate_version(constants.PROTOCOL_VERSION_0_3)
-    @validate_async_generator(
+    @validate(
         lambda self: self.agent_card.capabilities.streaming,
         'Streaming is not supported by the agent',
     )