fix(streaming): wrap mid-stream transport errors as APIError subclasses

src/anthropic/_streaming.py

@@ -13,6 +13,7 @@
 
 from ._utils import extract_type_var_from_base
 from ._compat import cached_property
+from ._exceptions import APITimeoutError, APIConnectionError
 
 if TYPE_CHECKING:
     from ._client import Anthropic, AsyncAnthropic
@@ -68,7 +69,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 httpx2.TimeoutException as err:
+            raise APITimeoutError(request=self.response.request) from err
+        except httpx2.TransportError as err:
+            raise APIConnectionError(request=self.response.request) from err
 
     @staticmethod
     def raw_events(response: httpx2.Response) -> Iterator[ServerSentEvent]:
@@ -196,8 +202,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 httpx2.TimeoutException as err:
+            raise APITimeoutError(request=self.response.request) from err
+        except httpx2.TransportError as err:
+            raise APIConnectionError(request=self.response.request) from err
 
     @staticmethod
     def raw_events(response: httpx2.Response) -> AsyncIterator[ServerSentEvent]:

tests/test_streaming.py

@@ -8,7 +8,7 @@
 
 from anthropic import Anthropic, AsyncAnthropic
 from anthropic._streaming import Stream, AsyncStream, ServerSentEvent
-from anthropic._exceptions import APIStatusError
+from anthropic._exceptions import APIStatusError, APITimeoutError, APIConnectionError
 
 from .utils import (
     STREAMS,
@@ -348,6 +348,64 @@ def body() -> Iterator[bytes]:
     assert "Overloaded" in str(exc_info.value)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+async def test_mid_stream_timeout_wrapped_as_api_timeout_error(
+    sync: bool, client: Anthropic, async_client: AsyncAnthropic
+) -> None:
+    def body() -> Iterator[bytes]:
+        yield b"event: completion\n"
+        yield b'data: {"foo":true}\n'
+        yield b"\n"
+        raise httpx2.ReadTimeout("timed out while reading the stream")
+
+    iterator = make_stream_iterator(content=body(), sync=sync, client=client, async_client=async_client)
+
+    # the first event still comes through fine
+    assert await iter_next(iterator) == {"foo": True}
+
+    with pytest.raises(APITimeoutError):
+        await iter_next(iterator)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+async def test_mid_stream_transport_error_wrapped_as_api_connection_error(
+    sync: bool, client: Anthropic, async_client: AsyncAnthropic
+) -> None:
+    def body() -> Iterator[bytes]:
+        yield b"event: completion\n"
+        yield b'data: {"foo":true}\n'
+        yield b"\n"
+        raise httpx2.RemoteProtocolError("connection dropped mid-stream")
+
+    iterator = make_stream_iterator(content=body(), sync=sync, client=client, async_client=async_client)
+
+    assert await iter_next(iterator) == {"foo": True}
+
+    with pytest.raises(APIConnectionError):
+        await iter_next(iterator)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+async def test_mid_stream_decoder_error_keeps_original_identity(
+    sync: bool, client: Anthropic, async_client: AsyncAnthropic
+) -> None:
+    # malformed UTF-8 in an SSE line: the default decoder raises
+    # UnicodeDecodeError, which is not a transport failure and must not
+    # be relabeled as APIConnectionError.
+    def body() -> Iterator[bytes]:
+        yield b"event: completion\n"
+        yield b"data: \xff\xfe not utf-8\n"
+        yield b"\n"
+
+    iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
+
+    with pytest.raises(UnicodeDecodeError):
+        await iter_next(iterator)
+
+
 def test_isinstance_check(client: Anthropic, async_client: AsyncAnthropic) -> None:
     async_stream = AsyncStream(cast_to=object, client=async_client, response=httpx2.Response(200, content=b"foo"))
     assert isinstance(async_stream, AsyncStream)