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",
@@ -94,7 +93,7 @@ filterwarnings = [
   # ResourceWarnings from asyncio event loop/socket cleanup during garbage collection
   # These appear intermittently between tests due to pytest-asyncio and sse-starlette timing
   "ignore:unclosed event loop:ResourceWarning",
-  "ignore:unclosed transport:ResourceWarning", 
+  "ignore:unclosed transport:ResourceWarning",
   "ignore:unclosed <socket.socket:ResourceWarning",
 ]
 

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,29 +128,29 @@ 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:
@@ -146,9 +180,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_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):

tests/client/transports/test_rest_client.py

@@ -29,7 +29,6 @@
 from a2a.utils.errors import A2A_ERROR_MAPPING
 from google.protobuf import json_format
 from google.protobuf.timestamp_pb2 import Timestamp
-from httpx_sse import EventSource, ServerSentEvent
 
 
 @pytest.fixture
@@ -52,8 +51,8 @@ def mock_agent_card() -> MagicMock:
 
 
 async def async_iterable_from_list(
-    items: list[ServerSentEvent],
-) -> AsyncGenerator[ServerSentEvent, None]:
+    items: list[str],
+) -> AsyncGenerator[str, None]:
     """Helper to create an async iterable from a list."""
     for item in items:
         yield item
@@ -84,18 +83,13 @@ async def test_send_message_streaming_timeout(
         params = SendMessageRequest(
             message=new_text_message(text='Hello stream')
         )
-        mock_event_source = AsyncMock(spec=EventSource)
-        mock_event_source.response = MagicMock(spec=httpx.Response)
-        mock_event_source.response.headers = {
-            'content-type': 'text/event-stream'
-        }
-        mock_event_source.response.raise_for_status.return_value = None
-        mock_event_source.aiter_sse.side_effect = httpx.TimeoutException(
+        mock_response = AsyncMock(spec=httpx.Response)
+        mock_response.headers = {'content-type': 'text/event-stream'}
+        mock_response.raise_for_status.return_value = None
+        mock_response.aiter_lines.side_effect = httpx.TimeoutException(
             'Read timed out'
         )
-        mock_aconnect_sse.return_value.__aenter__.return_value = (
-            mock_event_source
-        )
+        mock_aconnect_sse.return_value.__aenter__.return_value = mock_response
 
         with pytest.raises(A2AClientError) as exc_info:
             _ = [
@@ -290,15 +284,11 @@ async def test_send_message_streaming_with_new_extensions(
             message=new_text_message(text='Hello stream')
         )
 
-        mock_event_source = AsyncMock(spec=EventSource)
-        mock_event_source.response = MagicMock(spec=httpx.Response)
-        mock_event_source.response.headers = {
-            'content-type': 'text/event-stream'
-        }
-        mock_event_source.aiter_sse.return_value = async_iterable_from_list([])
-        mock_aconnect_sse.return_value.__aenter__.return_value = (
-            mock_event_source
-        )
+        mock_response = AsyncMock(spec=httpx.Response)
+        mock_response.headers = {'content-type': 'text/event-stream'}
+        mock_response.raise_for_status.return_value = None
+        mock_response.aiter_lines.return_value = async_iterable_from_list([])
+        mock_aconnect_sse.return_value.__aenter__.return_value = mock_response
 
         context = ClientCallContext(
             service_parameters={
@@ -339,24 +329,16 @@ async def test_send_message_streaming_server_error_propagates(
             message=new_text_message(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,
         )
-
-        async def empty_aiter():
-            if False:
-                yield
-
-        mock_event_source.response = mock_response
-        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):
@@ -706,21 +688,11 @@ async def test_rest_streaming_methods_prepend_tenant(  # noqa: PLR0913
         method = getattr(client, method_name)
 
         # 2. Setup mocks
-        mock_event_source = AsyncMock(spec=EventSource)
-        mock_event_source.response = MagicMock(spec=httpx.Response)
-        mock_event_source.response.headers = {
-            'content-type': 'text/event-stream'
-        }
-        mock_event_source.response.raise_for_status.return_value = None
-
-        async def empty_aiter():
-            if False:
-                yield
-
-        mock_event_source.aiter_sse.return_value = empty_aiter()
-        mock_aconnect_sse.return_value.__aenter__.return_value = (
-            mock_event_source
-        )
+        mock_response = AsyncMock(spec=httpx.Response)
+        mock_response.headers = {'content-type': 'text/event-stream'}
+        mock_response.raise_for_status.return_value = None
+        mock_response.aiter_lines.return_value = async_iterable_from_list([])
+        mock_aconnect_sse.return_value.__aenter__.return_value = mock_response
 
         # 3. Call the method
         async for _ in method(request=request_obj):