refactor(http): eliminate third-party httpx-sse dependency

pyproject.toml

@@ -9,7 +9,6 @@ requires-python = ">=3.10"
 keywords = ["A2A", "A2A SDK", "A2A Protocol", "Agent2Agent", "Agent 2 Agent"]
 dependencies = [
   "httpx>=0.28.1",
-  "httpx-sse>=0.4.0",
   "pydantic>=2.11.3",
   "protobuf>=5.29.5,<7",
   "google-api-core>=1.26.0",

src/a2a/client/transports/http_helpers.py

@@ -6,8 +6,6 @@
 
 import httpx
 
-from httpx_sse import EventSource, SSEError
-
 from a2a.client.client import ClientCallContext
 from a2a.client.errors import A2AClientError, A2AClientTimeoutError
 
@@ -36,10 +34,6 @@ def handle_http_exceptions(
         if status_error_handler:
             status_error_handler(e)
         raise A2AClientError(f'HTTP Error {e.response.status_code}: {e}') from e
-    except SSEError as e:
-        raise A2AClientError(
-            f'Invalid SSE response or protocol error: {e}'
-        ) from e
     except httpx.RequestError as e:
         raise A2AClientError(f'Network communication error: {e}') from e
     except json.JSONDecodeError as e:
@@ -69,6 +63,46 @@ async def send_http_request(
         return response.json()
 
 
+async def parse_sse_stream(
+    response: httpx.Response,
+) -> AsyncGenerator[tuple[str, str], None]:
+    """Yields (event_name, data) from a streaming httpx.Response.
+
+    Conforms to the W3C Server-Sent Events specification.
+    """
+    event_name = 'message'
+    payload_chunks: list[str] = []
+
+    async for line in response.aiter_lines():
+        raw_line = line.rstrip('\r\n')
+
+        # Empty line denotes the completion of the current event block
+        if not raw_line:
+            if payload_chunks:
+                yield event_name, '\n'.join(payload_chunks)
+            event_name = 'message'
+            payload_chunks = []
+            continue
+
+        # Ignore comment lines
+        if raw_line.startswith(':'):
+            continue
+
+        # Split key and value by first colon
+        parts = raw_line.split(':', 1)
+        key = parts[0]
+        val = parts[1] if len(parts) > 1 else ''
+
+        # Strip a single optional leading space
+        if val.startswith(' '):
+            val = val[1:]
+
+        if key == 'event':
+            event_name = val
+        elif key == 'data':
+            payload_chunks.append(val)
+
+
 async def send_http_stream_request(
     httpx_client: httpx.AsyncClient,
     method: str,
@@ -94,40 +128,44 @@ async def send_http_stream_request(
     with handle_http_exceptions(status_error_handler):
         async with _SSEEventSource(
             httpx_client, method, url, **kwargs
-        ) as event_source:
+        ) as response:
             try:
-                event_source.response.raise_for_status()
+                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()
+                await 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(
+            if 'text/event-stream' not in response.headers.get(
                 'content-type', ''
             ):
-                content = await event_source.response.aread()
+                content = await response.aread()
                 yield content.decode('utf-8')
                 return
 
-            async for sse in event_source.aiter_sse():
-                if not sse.data:
+            async for event_name, data in parse_sse_stream(response):
+                if not data:
                     continue
-                if sse.event == 'error':
-                    sse_error_handler(sse.data)
-                yield sse.data
+                if event_name == 'error':
+                    sse_error_handler(data)
+                yield data
 
 
 class _SSEEventSource:
-    """Class-based replacement for ``httpx_sse.aconnect_sse``.
-
-    ``aconnect_sse`` is an ``@asynccontextmanager`` whose internal async
-    generator gets tracked by the event loop. When the enclosing async
-    generator is abandoned, the event loop's generator cleanup collides
-    with the cascading cleanup — see https://bugs.python.org/issue38559.
-
-    Plain ``__aenter__``/``__aexit__`` coroutines avoid this entirely.
+    """Class-based context manager for managing streaming HTTP connections.
+
+    Using a class with `__aenter__` and `__aexit__` instead of an
+    `@asynccontextmanager` decorated function prevents event loop finalization
+    crashes (specifically `RuntimeError: GeneratorExit thrown into an active
+    generator`).
+
+    This error occurs when an outer async generator (e.g., streaming reader)
+    is abandoned early, causing the Python event loop to concurrently throw
+    GeneratorExit into the nested context manager's suspended generator.
+    Coroutine-based context managers are not async generators and avoid this
+    collision (see https://bugs.python.org/issue38559).
     """
 
     def __init__(
@@ -146,9 +184,9 @@ def __init__(
         self._client = client
         self._response: httpx.Response | None = None
 
-    async def __aenter__(self) -> EventSource:
+    async def __aenter__(self) -> httpx.Response:
         self._response = await self._client.send(self._request, stream=True)
-        return EventSource(self._response)
+        return self._response
 
     async def __aexit__(self, *args: object) -> None:
         if self._response is not None:

tests/client/transports/test_http_helpers.py

@@ -0,0 +1,64 @@
+import httpx
+import pytest
+
+from a2a.client.errors import A2AClientError
+from a2a.client.transports.http_helpers import (
+    _default_sse_error_handler,
+    parse_sse_stream,
+    send_http_stream_request,
+)
+
+
+def test_default_sse_error_handler():
+    with pytest.raises(
+        A2AClientError, match='SSE stream error event received: error_msg'
+    ):
+        _default_sse_error_handler('error_msg')
+
+
+@pytest.mark.asyncio
+async def test_parse_sse_stream_edge_cases():
+    async def mock_aiter_lines():
+        yield ': comment line (should be ignored)\n'
+        yield 'event: custom_event\n'
+        yield 'data:  hello\n'
+        yield 'data: world\n'
+        yield '\n'
+        yield '\n'
+        yield 'data: \n'
+        yield '\n'
+
+    response = httpx.Response(200)
+    response.aiter_lines = mock_aiter_lines  # type: ignore
+
+    events = [e async for e in parse_sse_stream(response)]
+    assert events == [
+        ('custom_event', ' hello\nworld'),
+        ('message', ''),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_send_http_stream_request_non_sse(mocker):
+    client = httpx.AsyncClient()
+    request = httpx.Request('GET', 'http://test')
+    response = httpx.Response(
+        200,
+        headers={'Content-Type': 'application/json'},
+        content=b'plain error response',
+        request=request,
+    )
+
+    mocker.patch(
+        'a2a.client.transports.http_helpers._SSEEventSource.__aenter__',
+        return_value=response,
+    )
+    mocker.patch(
+        'a2a.client.transports.http_helpers._SSEEventSource.__aexit__',
+        return_value=None,
+    )
+
+    chunks = [
+        c async for c in send_http_stream_request(client, 'GET', 'http://test')
+    ]
+    assert chunks == ['plain error response']

tests/client/transports/test_jsonrpc_client.py

@@ -2,6 +2,7 @@
 
 import json
 
+from collections.abc import AsyncGenerator
 from unittest.mock import AsyncMock, MagicMock, patch
 from uuid import uuid4
 
@@ -31,7 +32,14 @@
 )
 from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP
 from google.protobuf import json_format
-from httpx_sse import EventSource, SSEError
+
+
+async def async_iterable_from_list(
+    items: list[str],
+) -> AsyncGenerator[str, None]:
+    """Helper to create an async iterable from a list."""
+    for item in items:
+        yield item
 
 
 @pytest.fixture
@@ -439,17 +447,17 @@ async def test_send_message_streaming_sse_error(
         transport: JsonRpcTransport,
     ):
         request = create_send_message_request()
-        mock_event_source = AsyncMock()
-        mock_event_source.response.raise_for_status = MagicMock()
-        mock_event_source.response.headers = {
-            'content-type': 'text/event-stream'
-        }
-        mock_event_source.aiter_sse = MagicMock(
-            side_effect=SSEError('Simulated SSE error')
-        )
-        mock_aconnect_sse.return_value.__aenter__.return_value = (
-            mock_event_source
+        mock_response = AsyncMock(spec=httpx.Response)
+        mock_response.raise_for_status = MagicMock()
+        mock_response.headers = {'content-type': 'text/event-stream'}
+        mock_response.aiter_lines.return_value = async_iterable_from_list(
+            [
+                'event: error',
+                'data: Simulated SSE error',
+                '',
+            ]
         )
+        mock_aconnect_sse.return_value.__aenter__.return_value = mock_response
 
         with pytest.raises(A2AClientError):
             async for _ in transport.send_message_streaming(request):
@@ -463,19 +471,13 @@ async def test_send_message_streaming_request_error(
         transport: JsonRpcTransport,
     ):
         request = create_send_message_request()
-        mock_event_source = AsyncMock()
-        mock_event_source.response.raise_for_status = MagicMock()
-        mock_event_source.response.headers = {
-            'content-type': 'text/event-stream'
-        }
-        mock_event_source.aiter_sse = MagicMock(
-            side_effect=httpx.RequestError(
-                'Simulated request error', request=MagicMock()
-            )
-        )
-        mock_aconnect_sse.return_value.__aenter__.return_value = (
-            mock_event_source
+        mock_response = AsyncMock(spec=httpx.Response)
+        mock_response.raise_for_status = MagicMock()
+        mock_response.headers = {'content-type': 'text/event-stream'}
+        mock_response.aiter_lines.side_effect = httpx.RequestError(
+            'Simulated request error', request=MagicMock()
         )
+        mock_aconnect_sse.return_value.__aenter__.return_value = mock_response
 
         with pytest.raises(A2AClientError):
             async for _ in transport.send_message_streaming(request):
@@ -489,17 +491,13 @@ async def test_send_message_streaming_timeout(
         transport: JsonRpcTransport,
     ):
         request = create_send_message_request()
-        mock_event_source = AsyncMock()
-        mock_event_source.response.raise_for_status = MagicMock()
-        mock_event_source.response.headers = {
-            'content-type': 'text/event-stream'
-        }
-        mock_event_source.aiter_sse = MagicMock(
-            side_effect=httpx.TimeoutException('Timeout')
-        )
-        mock_aconnect_sse.return_value.__aenter__.return_value = (
-            mock_event_source
+        mock_response = AsyncMock(spec=httpx.Response)
+        mock_response.raise_for_status = MagicMock()
+        mock_response.headers = {'content-type': 'text/event-stream'}
+        mock_response.aiter_lines.side_effect = httpx.TimeoutException(
+            'Timeout'
         )
+        mock_aconnect_sse.return_value.__aenter__.return_value = mock_response
 
         with pytest.raises(A2AClientError, match='timed out'):
             async for _ in transport.send_message_streaming(request):
@@ -574,24 +572,16 @@ async def test_send_message_streaming_server_error_propagates(
         )
         request = create_send_message_request(text='Error stream')
 
-        mock_event_source = AsyncMock(spec=EventSource)
-        mock_response = MagicMock(spec=httpx.Response)
+        mock_response = AsyncMock(spec=httpx.Response)
         mock_response.status_code = 403
+        mock_response.headers = {'content-type': 'text/event-stream'}
         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
-
-        async def empty_aiter():
-            if False:
-                yield
-
-        mock_event_source.aiter_sse = MagicMock(return_value=empty_aiter())
-        mock_aconnect_sse.return_value.__aenter__.return_value = (
-            mock_event_source
-        )
+        mock_response.aiter_lines.return_value = async_iterable_from_list([])
+        mock_aconnect_sse.return_value.__aenter__.return_value = mock_response
 
         with pytest.raises(A2AClientError) as exc_info:
             async for _ in client.send_message_streaming(request=request):