feat(streaming): normalize errors raised while reading streams

README.md

@@ -657,6 +657,14 @@ response), a subclass of `openai.APIStatusError` is raised, containing `status_c
 
 All errors inherit from `openai.APIError`.
 
+When consuming a `Stream` or `AsyncStream`, read timeouts raise `APITimeoutError`
+and other HTTPX request failures raise `APIConnectionError`. Catch these SDK
+exceptions instead of raw HTTPX exceptions; the original exception is available
+as `__cause__`. Stream consumption is not automatically retried, because replaying
+a request could duplicate output already delivered to your application.
+The Assistants event-handler helpers and raw `with_streaming_response` iterators
+retain their existing exception behavior.
+
 ```python
 import openai
 from openai import OpenAI

src/openai/_httpx2.py

@@ -17,6 +17,7 @@ class _LegacyHttpxModule(Protocol):
     Timeout: type[httpx2.Timeout]
     Limits: type[httpx2.Limits]
     TimeoutException: type[httpx2.TimeoutException]
+    RequestError: type[httpx2.RequestError]
     HTTPStatusError: type[httpx2.HTTPStatusError]
     StreamConsumed: type[httpx2.StreamConsumed]
     RequestNotRead: type[httpx2.RequestNotRead]
@@ -99,6 +100,11 @@ def timeout_exceptions() -> tuple[type[httpx2.TimeoutException], ...]:
     return (httpx2.TimeoutException,) if module is None else (httpx2.TimeoutException, module.TimeoutException)
 
 
+def request_exceptions() -> tuple[type[httpx2.RequestError], ...]:
+    module = _loaded_legacy_httpx()
+    return (httpx2.RequestError,) if module is None else (httpx2.RequestError, module.RequestError)
+
+
 def status_exceptions() -> tuple[type[httpx2.HTTPStatusError], ...]:
     module = _loaded_legacy_httpx()
     return (httpx2.HTTPStatusError,) if module is None else (httpx2.HTTPStatusError, module.HTTPStatusError)

src/openai/_streaming.py

@@ -11,7 +11,8 @@
 import httpx2
 
 from ._utils import is_mapping, extract_type_var_from_base
-from ._exceptions import APIError
+from ._httpx2 import request_exceptions, timeout_exceptions
+from ._exceptions import APIError, APITimeoutError, APIConnectionError
 
 if TYPE_CHECKING:
     from ._client import OpenAI, AsyncOpenAI
@@ -51,7 +52,12 @@ def __iter__(self) -> Iterator[_T]:
             yield item
 
     def _iter_events(self) -> Iterator[ServerSentEvent]:
-        yield from self._decoder.iter_bytes(self.response.iter_bytes())
+        try:
+            yield from self._decoder.iter_bytes(self.response.iter_bytes())
+        except timeout_exceptions() as err:
+            raise APITimeoutError(request=self.response.request) from err
+        except request_exceptions() as err:
+            raise APIConnectionError(request=self.response.request) from err
 
     def __stream__(self) -> Iterator[_T]:
         cast_to = cast(Any, self._cast_to)
@@ -160,8 +166,13 @@ async def __aiter__(self) -> AsyncIterator[_T]:
             yield item
 
     async def _iter_events(self) -> AsyncIterator[ServerSentEvent]:
-        async for sse in self._decoder.aiter_bytes(self.response.aiter_bytes()):
-            yield sse
+        try:
+            async for sse in self._decoder.aiter_bytes(self.response.aiter_bytes()):
+                yield sse
+        except timeout_exceptions() as err:
+            raise APITimeoutError(request=self.response.request) from err
+        except request_exceptions() as err:
+            raise APIConnectionError(request=self.response.request) from err
 
     async def __stream__(self) -> AsyncIterator[_T]:
         cast_to = cast(Any, self._cast_to)

src/openai/lib/streaming/_assistants.py

@@ -8,10 +8,11 @@
 from ._deltas import accumulate_delta as accumulate_delta
 from ..._utils import consume_sync_iterator, consume_async_iterator
 from ..._compat import model_dump
-from ..._httpx2 import timeout_exceptions
+from ..._httpx2 import request_exceptions, timeout_exceptions
 from ..._models import construct_type
 from ..._streaming import Stream, AsyncStream
 from ...types.beta import AssistantStreamEvent
+from ..._exceptions import APIConnectionError
 from ...types.beta.threads import (
     Run,
     Text,
@@ -29,6 +30,26 @@ def _timeout_exceptions() -> tuple[type[Exception], ...]:
     return (*timeout_exceptions(), asyncio.TimeoutError)
 
 
+def _iter_events(stream: Stream[AssistantStreamEvent]) -> Iterator[AssistantStreamEvent]:
+    # Preserve legacy transport exceptions without unwrapping errors from user callbacks.
+    try:
+        yield from stream
+    except APIConnectionError as exc:
+        if isinstance(exc.__cause__, request_exceptions()):
+            raise exc.__cause__ from None
+        raise
+
+
+async def _aiter_events(stream: AsyncStream[AssistantStreamEvent]) -> AsyncIterator[AssistantStreamEvent]:
+    try:
+        async for event in stream:
+            yield event
+    except APIConnectionError as exc:
+        if isinstance(exc.__cause__, request_exceptions()):
+            raise exc.__cause__ from None
+        raise
+
+
 class AssistantEventHandler:
     text_deltas: Iterable[str]
     """Iterator over just the text deltas in the stream.
@@ -407,7 +428,7 @@ def __stream__(self) -> Iterator[AssistantStreamEvent]:
             raise RuntimeError("Stream has not been started yet")
 
         try:
-            for event in stream:
+            for event in _iter_events(stream):
                 self._emit_sse_event(event)
 
                 yield event
@@ -839,7 +860,7 @@ async def __stream__(self) -> AsyncIterator[AssistantStreamEvent]:
             raise RuntimeError("Stream has not been started yet")
 
         try:
-            async for event in stream:
+            async for event in _aiter_events(stream):
                 await self._emit_sse_event(event)
 
                 yield event

tests/test_httpx2.py

@@ -21,6 +21,7 @@
 from openai._response import StreamAlreadyConsumed
 from openai.providers import bedrock
 from openai._constants import DEFAULT_TIMEOUT
+from openai.types.beta import AssistantStreamEvent
 
 
 def model_list(request: httpx2.Request) -> httpx2.Response:
@@ -577,7 +578,8 @@ async def async_failing_stream_handler(request: httpx2.Request) -> httpx2.Respon
 
 
 @pytest.mark.filterwarnings("ignore:The Assistants API is deprecated in favor of the Responses API:DeprecationWarning")
-async def test_assistant_stream_timeout_callbacks_preserve_httpx2_family() -> None:
+@pytest.mark.parametrize("error_type", [httpx2.ReadTimeout, httpx2.RemoteProtocolError, httpx2.DecodingError])
+async def test_assistant_stream_error_callbacks_preserve_httpx2_family(error_type: type[httpx2.RequestError]) -> None:
     class SyncHandler(openai.AssistantEventHandler):
         def __init__(self) -> None:
             super().__init__()
@@ -610,13 +612,13 @@ class FailingSyncStream(httpx2.SyncByteStream):
         @override
         def __iter__(self):
             yield b"partial"
-            raise httpx2.ReadTimeout("assistant stream timeout")
+            raise error_type("assistant stream failure")
 
     class FailingAsyncStream(httpx2.AsyncByteStream):
         @override
         async def __aiter__(self):
             yield b"partial"
-            raise httpx2.ReadTimeout("assistant stream timeout")
+            raise error_type("assistant stream failure")
 
     def sync_response(request: httpx2.Request) -> httpx2.Response:
         return httpx2.Response(
@@ -638,11 +640,11 @@ async def async_response(request: httpx2.Request) -> httpx2.Response:
         with sync_client.beta.threads.runs.stream(  # pyright: ignore[reportDeprecated]
             assistant_id="asst_test", thread_id="thread_test", event_handler=sync_handler
         ) as stream:
-            with pytest.raises(httpx2.ReadTimeout, match="assistant stream timeout"):
+            with pytest.raises(error_type, match="assistant stream failure"):
                 stream.until_done()
 
-    assert sync_handler.timed_out
-    assert isinstance(sync_handler.exception, httpx2.ReadTimeout)
+    assert sync_handler.timed_out == issubclass(error_type, httpx2.TimeoutException)
+    assert isinstance(sync_handler.exception, error_type)
 
     async_handler = AsyncHandler()
     async with AsyncOpenAI(
@@ -654,11 +656,132 @@ async def async_response(request: httpx2.Request) -> httpx2.Response:
         async with async_client.beta.threads.runs.stream(  # pyright: ignore[reportDeprecated]
             assistant_id="asst_test", thread_id="thread_test", event_handler=async_handler
         ) as async_stream:
-            with pytest.raises(httpx2.ReadTimeout, match="assistant stream timeout"):
+            with pytest.raises(error_type, match="assistant stream failure"):
                 await async_stream.until_done()
 
-    assert async_handler.timed_out
-    assert isinstance(async_handler.exception, httpx2.ReadTimeout)
+    assert async_handler.timed_out == issubclass(error_type, httpx2.TimeoutException)
+    assert isinstance(async_handler.exception, error_type)
+
+
+@pytest.mark.filterwarnings("ignore:The Assistants API is deprecated in favor of the Responses API:DeprecationWarning")
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+@pytest.mark.parametrize("timeout", [True, False], ids=["timeout", "connection"])
+async def test_assistant_stream_hook_api_errors_are_not_unwrapped(sync: bool, timeout: bool) -> None:
+    hook_request = httpx2.Request("GET", "https://example.test/hook")
+
+    if timeout:
+        cause: httpx2.RequestError = httpx2.ReadTimeout("hook timeout")
+        hook_error: APIConnectionError = APITimeoutError(hook_request)
+    else:
+        cause = httpx2.RemoteProtocolError("hook connection error")
+        hook_error = APIConnectionError(request=hook_request)
+
+    hook_error.__cause__ = cause
+
+    class SyncHandler(openai.AssistantEventHandler):
+        def __init__(self) -> None:
+            super().__init__()
+            self.timed_out = False
+            self.exception: Exception | None = None
+
+        @override
+        def on_event(self, event: AssistantStreamEvent) -> None:
+            raise hook_error
+
+        @override
+        def on_timeout(self) -> None:
+            self.timed_out = True
+
+        @override
+        def on_exception(self, exception: Exception) -> None:
+            self.exception = exception
+
+    class AsyncHandler(openai.AsyncAssistantEventHandler):
+        def __init__(self) -> None:
+            super().__init__()
+            self.timed_out = False
+            self.exception: Exception | None = None
+
+        @override
+        async def on_event(self, event: AssistantStreamEvent) -> None:
+            raise hook_error
+
+        @override
+        async def on_timeout(self) -> None:
+            self.timed_out = True
+
+        @override
+        async def on_exception(self, exception: Exception) -> None:
+            self.exception = exception
+
+    content = b'event: thread.created\ndata: {"id":"thread_test","created_at":0,"object":"thread"}\n\n'
+
+    def sync_response(request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(
+            200,
+            headers={"content-type": "text/event-stream"},
+            content=content,
+            request=request,
+        )
+
+    async def async_response(request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(
+            200,
+            headers={"content-type": "text/event-stream"},
+            content=content,
+            request=request,
+        )
+
+    if sync:
+        handler = SyncHandler()
+
+        with OpenAI(
+            api_key="test",
+            base_url="https://example.test/v1",
+            http_client=openai.DefaultHttpx2Client(
+                transport=httpx2.MockTransport(sync_response),
+                trust_env=False,
+            ),
+            max_retries=0,
+        ) as client:
+            with client.beta.threads.runs.stream(  # pyright: ignore[reportDeprecated]
+                assistant_id="asst_test",
+                thread_id="thread_test",
+                event_handler=handler,
+            ) as stream:
+                with pytest.raises(APIConnectionError) as exc_info:
+                    stream.until_done()
+
+        assert exc_info.value is hook_error
+        assert handler.exception is hook_error
+        assert not handler.timed_out
+        assert hook_error.__cause__ is cause
+
+    else:
+        handler = AsyncHandler()
+
+        async with AsyncOpenAI(
+            api_key="test",
+            base_url="https://example.test/v1",
+            http_client=openai.DefaultAsyncHttpx2Client(
+                transport=httpx2.MockTransport(async_response),
+                trust_env=False,
+            ),
+            max_retries=0,
+        ) as client:
+            async with client.beta.threads.runs.stream(  # pyright: ignore[reportDeprecated]
+                assistant_id="asst_test",
+                thread_id="thread_test",
+                event_handler=handler,
+            ) as stream:
+                with pytest.raises(APIConnectionError) as exc_info:
+                    await stream.until_done()
+
+        assert exc_info.value is hook_error
+        assert handler.exception is hook_error
+        assert not handler.timed_out
+        assert hook_error.__cause__ is cause
 
 
 async def test_sigv4_provider_preserves_httpx2_family_and_rejects_one_shot_bodies() -> None: