fix: raise APIError for top-level Responses API error events during streaming

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(

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__()