From 698bbc5c9ba609892c6540c08ff2fb917b0b7ec0 Mon Sep 17 00:00:00 2001 From: Mayuri Date: Sun, 9 Aug 2026 23:13:36 +0530 Subject: [PATCH] fix: raise APIError for top-level Responses API error events during streaming Stream.__stream__ / AsyncStream.__stream__ only raised an APIError when the SSE event body had a nested "error" key. Per the Responses API streaming spec, the "error" event carries its fields (type, code, message, param, sequence_number) directly on the event body with no nested "error" key, so these events were silently yielded as regular data instead of raising, leading to confusing downstream validation errors instead of a clear APIError with the server's message. Fixes #2487 --- src/openai/_streaming.py | 18 +++++++---- tests/test_streaming.py | 68 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 79 insertions(+), 7 deletions(-) diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py index 45c13cc11d..9ae053ed40 100644 --- a/src/openai/_streaming.py +++ b/src/openai/_streaming.py @@ -84,9 +84,12 @@ def __stream__(self) -> Iterator[_T]: yield process_data(data={"data": data, "event": sse.event}, cast_to=cast_to, response=response) else: data = sse.json() - if is_mapping(data) and data.get("error"): + if is_mapping(data) and (data.get("error") or sse.event == "error"): message = None - error = data.get("error") + # some events (e.g. the Responses API's `response.error` event) carry + # the error fields directly on the event body instead of nesting them + # under an "error" key, so fall back to the top-level body in that case. + error = data.get("error") if is_mapping(data.get("error")) else data if is_mapping(error): message = error.get("message") if not message or not isinstance(message, str): @@ -95,7 +98,7 @@ def __stream__(self) -> Iterator[_T]: raise APIError( message=message, request=self.response.request, - body=data["error"], + body=data.get("error") if data.get("error") is not None else data, ) yield process_data( @@ -194,9 +197,12 @@ async def __stream__(self) -> AsyncIterator[_T]: yield process_data(data={"data": data, "event": sse.event}, cast_to=cast_to, response=response) else: data = sse.json() - if is_mapping(data) and data.get("error"): + if is_mapping(data) and (data.get("error") or sse.event == "error"): message = None - error = data.get("error") + # some events (e.g. the Responses API's `response.error` event) carry + # the error fields directly on the event body instead of nesting them + # under an "error" key, so fall back to the top-level body in that case. + error = data.get("error") if is_mapping(data.get("error")) else data if is_mapping(error): message = error.get("message") if not message or not isinstance(message, str): @@ -205,7 +211,7 @@ async def __stream__(self) -> AsyncIterator[_T]: raise APIError( message=message, request=self.response.request, - body=data["error"], + body=data.get("error") if data.get("error") is not None else data, ) yield process_data( diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 04f8e51abd..145f1f904c 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -5,7 +5,7 @@ import httpx import pytest -from openai import OpenAI, AsyncOpenAI +from openai import OpenAI, APIError, AsyncOpenAI from openai._streaming import Stream, AsyncStream, ServerSentEvent @@ -216,6 +216,44 @@ def body() -> Iterator[bytes]: assert sse.json() == {"content": "известни"} +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +async def test_response_error_event_raises_api_error(sync: bool, client: OpenAI, async_client: AsyncOpenAI) -> None: + # Per https://platform.openai.com/docs/api-reference/responses-streaming/error, the + # Responses API's `error` event carries its fields directly on the event body, e.g. + # {"type": "error", "code": "...", "message": "...", "param": null, "sequence_number": 1} + # -- there is no nested "error" key. + def body() -> Iterator[bytes]: + yield b"event: error\n" + yield b'data: {"type":"error","code":"server_error","message":"boom","param":null,"sequence_number":1}\n' + yield b"\n" + + stream = make_stream(content=body(), sync=sync, client=client, async_client=async_client) + + with pytest.raises(APIError) as exc_info: + await stream_next(stream, sync=sync) + + assert exc_info.value.message == "boom" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +async def test_response_error_event_without_message_uses_default( + sync: bool, client: OpenAI, async_client: AsyncOpenAI +) -> None: + def body() -> Iterator[bytes]: + yield b"event: error\n" + yield b'data: {"type":"error","code":"server_error","param":null,"sequence_number":1}\n' + yield b"\n" + + stream = make_stream(content=body(), sync=sync, client=client, async_client=async_client) + + with pytest.raises(APIError) as exc_info: + await stream_next(stream, sync=sync) + + assert exc_info.value.message == "An error occurred during streaming" + + async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]: for chunk in iter: yield chunk @@ -246,3 +284,31 @@ def make_event_iterator( return AsyncStream( cast_to=object, client=async_client, response=httpx.Response(200, content=to_aiter(content)) )._iter_events() + + +def make_stream( + content: Iterator[bytes], + *, + sync: bool, + client: OpenAI, + async_client: AsyncOpenAI, +) -> Stream[object] | AsyncStream[object]: + request = httpx.Request("POST", "http://localhost") + + if sync: + return Stream(cast_to=object, client=client, response=httpx.Response(200, content=content, request=request)) + + return AsyncStream( + cast_to=object, + client=async_client, + response=httpx.Response(200, content=to_aiter(content), request=request), + ) + + +async def stream_next(stream: Stream[object] | AsyncStream[object], *, sync: bool) -> object: + if sync: + assert isinstance(stream, Stream) + return next(stream) + + assert isinstance(stream, AsyncStream) + return await stream.__anext__()