Updates

Author: ishymko

src/a2a/client/transports/http_helpers.py

@@ -12,6 +12,10 @@
 from a2a.client.errors import A2AClientError, A2AClientTimeoutError
 
 
+def _default_sse_error_handler(sse_data: str) -> NoReturn:
+    raise A2AClientError(f'SSE stream error event received: {sse_data}')
+
+
 @contextmanager
 def handle_http_exceptions(
     status_error_handler: Callable[[httpx.HTTPStatusError], NoReturn]
@@ -70,7 +74,7 @@ async def send_http_stream_request(
     method: str,
     url: str,
     status_error_handler: Callable[[httpx.HTTPStatusError], NoReturn],
-    sse_error_handler: Callable[[str], NoReturn],
+    sse_error_handler: Callable[[str], NoReturn] = _default_sse_error_handler,
     **kwargs: Any,
 ) -> AsyncGenerator[str]:
     """Sends a streaming HTTP request, yielding SSE data strings and handling exceptions.

src/a2a/compat/v0_3/jsonrpc_adapter.py

@@ -306,10 +306,9 @@ async def event_generator(
                     )
                 )
                 yield {
-                    'event': 'error',
                     'data': err_resp.model_dump_json(
                         by_alias=True, exclude_none=True
-                    ),
+                    )
                 }
 
         return EventSourceResponse(event_generator(stream_gen))

src/a2a/compat/v0_3/jsonrpc_transport.py

@@ -415,13 +415,6 @@ def _handle_http_error(self, e: httpx.HTTPStatusError) -> NoReturn:
         """Handles HTTP errors for standard requests."""
         raise A2AClientError(f'HTTP Error: {e.response.status_code}') from e
 
-    def _handle_sse_error(self, sse_data: str) -> NoReturn:
-        """Handles SSE error events by parsing JSON-RPC error payload and raising the appropriate domain error."""
-        data = json.loads(sse_data)
-        if 'error' in data:
-            raise self._create_jsonrpc_error(data['error'])
-        raise A2AClientError(f'SSE stream error: {sse_data}')
-
     async def _send_stream_request(
         self,
         json_data: dict[str, Any],
@@ -437,7 +430,6 @@ async def _send_stream_request(
             'POST',
             self.url,
             self._handle_http_error,
-            self._handle_sse_error,
             json=json_data,
             **http_kwargs,
         ):

src/a2a/compat/v0_3/rest_adapter.py

@@ -7,7 +7,6 @@
 
 
 if TYPE_CHECKING:
-    from sse_starlette.event import ServerSentEvent
     from sse_starlette.sse import EventSourceResponse
     from starlette.requests import Request
     from starlette.responses import JSONResponse, Response
@@ -18,7 +17,6 @@
     _package_starlette_installed = True
 else:
     try:
-        from sse_starlette.event import ServerSentEvent
         from sse_starlette.sse import EventSourceResponse
         from starlette.requests import Request
         from starlette.responses import JSONResponse, Response
@@ -29,7 +27,6 @@
         Request = Any
         JSONResponse = Any
         Response = Any
-        ServerSentEvent = Any
 
         _package_starlette_installed = False
 
@@ -40,7 +37,6 @@
 from a2a.server.context import ServerCallContext
 from a2a.server.routes import CallContextBuilder, DefaultCallContextBuilder
 from a2a.utils.error_handlers import (
-    build_rest_error_payload,
     rest_error_handler,
     rest_stream_error_handler,
 )
@@ -105,16 +101,9 @@ async def _handle_streaming_request(
 
         async def event_generator(
             stream: AsyncIterable[Any],
-        ) -> AsyncIterator[str | ServerSentEvent]:
-            try:
-                async for item in stream:
-                    yield json.dumps(item)
-            except Exception as e:
-                logger.exception('Error during v0.3 REST SSE stream')
-                yield ServerSentEvent(
-                    data=json.dumps(build_rest_error_payload(e)),
-                    event='error',
-                )
+        ) -> AsyncIterator[str]:
+            async for item in stream:
+                yield json.dumps(item)
 
         return EventSourceResponse(
             event_generator(method(request, call_context))

src/a2a/compat/v0_3/rest_transport.py

@@ -44,11 +44,7 @@
     TaskPushNotificationConfig,
 )
 from a2a.utils.constants import PROTOCOL_VERSION_0_3, VERSION_HEADER
-from a2a.utils.errors import (
-    A2A_REASON_TO_ERROR,
-    JSON_RPC_ERROR_CODE_MAP,
-    MethodNotFoundError,
-)
+from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP, MethodNotFoundError
 from a2a.utils.telemetry import SpanKind, trace_class
 
 
@@ -373,30 +369,6 @@ def _handle_http_error(self, e: httpx.HTTPStatusError) -> NoReturn:
 
         raise A2AClientError(f'HTTP Error {status_code}: {e}') from e
 
-    def _handle_sse_error(self, sse_data: str) -> NoReturn:
-        """Handles SSE error events by parsing the REST error payload and raising the appropriate A2AError."""
-        error_payload = json.loads(sse_data)
-        error_data = error_payload.get('error', {})
-
-        message = error_data.get('message', sse_data)
-        details = error_data.get('details', [])
-        if not isinstance(details, list):
-            details = []
-
-        for d in details:
-            if (
-                isinstance(d, dict)
-                and d.get('@type') == 'type.googleapis.com/google.rpc.ErrorInfo'
-            ):
-                reason = d.get('reason')
-                if isinstance(reason, str):
-                    exception_cls = A2A_REASON_TO_ERROR.get(reason)
-                    if exception_cls:
-                        raise exception_cls(message)
-                break
-
-        raise A2AClientError(message)
-
     async def _send_stream_request(
         self,
         method: str,
@@ -414,7 +386,6 @@ async def _send_stream_request(
             method,
             f'{self.url}{path}',
             self._handle_http_error,
-            self._handle_sse_error,
             json=json,
             **http_kwargs,
         ):

tests/compat/v0_3/test_jsonrpc_transport.py

@@ -480,7 +480,6 @@ async def mock_generator(*args, **kwargs):
         'POST',
         'http://example.com',
         transport._handle_http_error,
-        transport._handle_sse_error,
         json={'some': 'data'},
         headers={'a2a-version': '0.3'},
     )

tests/compat/v0_3/test_rest_transport.py

@@ -634,7 +634,6 @@ async def mock_generator(*args, **kwargs):
         'POST',
         'http://example.com/test',
         transport._handle_http_error,
-        transport._handle_sse_error,
         json=None,
         headers={'a2a-version': '0.3'},
     )