fix: surface Responses API streaming error events (top-level message)

[KaiyiQuan]

Fixes #2487

Problem

The SDK's own generated type ResponseErrorEvent (from the OpenAPI spec) carries message at the top level and has no nested error object:

class ResponseErrorEvent(BaseModel):
    code: Optional[str] = None
    message: str          # required, top level
    param: Optional[str] = None
    sequence_number: int
    type: Literal["error"]

But _streaming.py only raises an error when the payload has a nested error key and reads data["error"]["message"]. For a real Responses API error event ({"type": "error", "code": ..., "message": ..., ...}), the guard fails and the error is silently swallowed instead of being surfaced as APIError.

Fix

In _streaming.py, when the SSE event type is "error":

• read the top-level message first, falling back to the nested error.message shape used by chat-completions-style payloads
• raise APIError with the correct message instead of yielding the event as data
• default the error body to the nested error object when present

Verification

• top-level message event → raises with the correct message
• nested {"error": {"message": ...}} payload → still raises (backward compatible)
• error event without a message → raises with a fallback message
• normal (non-error) events → unaffected
• python -m py_compile passes on the changed module

[Copilot]

Copilot review overview

🟡 Changes recommended

Add regression tests for top-level, nested, and missing-message streaming error events.

Get a fresh assessment by requesting another Copilot review.

Review effort: Lite
Findings: 1

Open (1)

• Add sync and async coverage for streaming error fallbacks · New

What changed in this PR

Fixes Responses API streaming errors by surfacing top-level error messages while preserving nested error compatibility.

Changes:

• Handles top-level and nested streaming error payloads.
• Raises APIError with appropriate fallback messages.
• Applies behavior to synchronous and asynchronous streams.

File	Summary
src/​openai/​_streaming.py	Updates sync/async SSE error handling.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[KaiyiQuan]

Fixed in f50cb1c: tests/test_streaming.py adds sync+async regression cases for all three error-event shapes — top-level message (Responses-style), nested error.message fallback, and missing-message fallback (generic An error occurred during streaming), parametrized over both openai and azure clients.

[jnohclee-rgb]

AI-assisted independent offline check at a49d79bd7c2b3769e703919dc96866eeb8336c79, compared with pinned main 919b6236382f3f9f76d8453efb9f9ad03cff1126.

Eight error/data envelope controls across Stream and AsyncStream: 10/16 on pinned main, 16/16 on this head. The added controls include absent/non-string top-level messages, top-level versus nested message priority, nested-error backward compatibility, and ordinary event emission. Error bodies are compared as well as messages.

Scope: Actual stream classes and SSE decoder over in-memory HTTP responses; model-data processing is stubbed. No endpoint request, network or provider compatibility claim. Python 3.14.7, Pydantic 2.13.5 and shared local dependencies; this is a bounded regression matrix, not the full SDK suite or a historical lockfile matrix. Network was denied by the test sandbox.

Standalone reproducer (run separately with each source tree on PYTHONPATH)

import asyncio,importlib,json
from types import SimpleNamespace
source=importlib.import_module('openai._streaming');http=importlib.import_module('httpx2')
fallback='An error occurred during streaming'
nested={'message':'fictional nested'}
cases=[('named_top','error',{'type':'error','message':'fictional top'},'fictional top',None),('named_nested','error',{'error':nested},'fictional nested',nested),('named_empty','error',{},fallback,None),('named_bad_top_nested','error',{'message':7,'error':nested},'fictional nested',nested),('named_both','error',{'message':'fictional top','error':nested},'fictional top',nested),('unnamed_nested',None,{'error':nested},'fictional nested',nested),('ordinary',None,{'fictional':'data'},None,None),('named_ordinary','response.output_text.delta',{'type':'response.output_text.delta','delta':'fictional'},None,None)]
async def main():
 rows=[]
 for is_async in [False,True]:
  for name,event,data,expected,body in cases:
   payload=((f'event: {event}\n' if event else '')+'data: '+json.dumps(data)+'\n\n').encode()
   response=http.Response(200,request=http.Request('GET','https://example.invalid/fictional'),content=payload)
   client=SimpleNamespace(_make_sse_decoder=source.SSEDecoder,_process_response_data=lambda **kwargs:kwargs['data'])
   stream=(source.AsyncStream if is_async else source.Stream)(cast_to=object,response=response,client=client)
   actual=None;error_body=None;emitted=[];kind=None
   try:
    if is_async:emitted=[x async for x in stream]
    else:emitted=list(stream)
   except source.APIError as exc:kind=type(exc).__name__;actual=exc.message;error_body=exc.body
   except Exception as exc:kind=type(exc).__name__
   expected_body=body if body is not None else data
   ok=(kind=='APIError' and actual==expected and error_body==expected_body) if expected else kind is None and len(emitted)==1
   rows.append({'case':name,'async':is_async,'error_type':kind,'message':actual,'expected_message':expected,'body_matches':error_body==expected_body if expected else None,'emitted':len(emitted),'pass':ok})
 print(json.dumps({'source_module':source.__file__,'cases':rows,'passed':sum(r['pass'] for r in rows),'failed':sum(not r['pass'] for r in rows),'scope':'Actual Stream/AsyncStream and SSE decoder over an in-memory HTTP response; stubbed model-data processor, no client endpoint/network/provider proof.'}))
asyncio.run(main())