diff --git a/pyproject.toml b/pyproject.toml index c1fbb9129..066983a36 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/a2a/client/transports/http_helpers.py b/src/a2a/client/transports/http_helpers.py index 0a73ed83c..76c441425 100644 --- a/src/a2a/client/transports/http_helpers.py +++ b/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: diff --git a/tests/client/transports/test_http_helpers.py b/tests/client/transports/test_http_helpers.py new file mode 100644 index 000000000..8ef5d2d89 --- /dev/null +++ b/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'] diff --git a/tests/client/transports/test_jsonrpc_client.py b/tests/client/transports/test_jsonrpc_client.py index 85c89b62e..7e6680e5f 100644 --- a/tests/client/transports/test_jsonrpc_client.py +++ b/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): diff --git a/tests/client/transports/test_rest_client.py b/tests/client/transports/test_rest_client.py index f7281c06a..43a651552 100644 --- a/tests/client/transports/test_rest_client.py +++ b/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): diff --git a/uv.lock b/uv.lock index d09af8654..accc48f81 100644 --- a/uv.lock +++ b/uv.lock @@ -16,7 +16,6 @@ dependencies = [ { name = "google-api-core" }, { name = "googleapis-common-protos" }, { name = "httpx" }, - { name = "httpx-sse" }, { name = "json-rpc" }, { name = "packaging" }, { name = "protobuf" }, @@ -118,7 +117,6 @@ requires-dist = [ { name = "grpcio-tools", marker = "extra == 'all'", specifier = ">=1.60" }, { name = "grpcio-tools", marker = "extra == 'grpc'", specifier = ">=1.60" }, { name = "httpx", specifier = ">=0.28.1" }, - { name = "httpx-sse", specifier = ">=0.4.0" }, { name = "json-rpc", specifier = ">=1.15.0" }, { name = "opentelemetry-api", marker = "extra == 'all'", specifier = ">=1.33.0" }, { name = "opentelemetry-api", marker = "extra == 'telemetry'", specifier = ">=1.33.0" }, @@ -1112,15 +1110,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" }, ] -[[package]] -name = "httpx-sse" -version = "0.4.3" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/0f/4c/751061ffa58615a32c31b2d82e8482be8dd4a89154f003147acee90f2be9/httpx_sse-0.4.3.tar.gz", hash = "sha256:9b1ed0127459a66014aec3c56bebd93da3c1bc8bb6618c8082039a44889a755d", size = 15943, upload-time = "2025-10-10T21:48:22.271Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/d2/fd/6668e5aec43ab844de6fc74927e155a3b37bf40d7c3790e49fc0406b6578/httpx_sse-0.4.3-py3-none-any.whl", hash = "sha256:0ac1c9fe3c0afad2e0ebb25a934a59f4c7823b60792691f779fad2c5568830fc", size = 8960, upload-time = "2025-10-10T21:48:21.158Z" }, -] - [[package]] name = "identify" version = "2.6.18"