fix: wrap mid-stream transport errors

src/anthropic/_streaming.py

@@ -10,6 +10,7 @@
 import httpx2
 
 from ._utils import is_dict, extract_type_var_from_base
+from ._exceptions import APITimeoutError, APIConnectionError
 
 if TYPE_CHECKING:
     from ._client import Anthropic, AsyncAnthropic
@@ -49,7 +50,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

@@ -7,7 +7,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
 
 _T = TypeVar("_T")
 
@@ -238,6 +238,37 @@ def body() -> Iterator[bytes]:
     assert "Overloaded" in str(exc_info.value)
 
 
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+@pytest.mark.parametrize(
+    ("transport_error", "api_error"),
+    [
+        (httpx2.ReadTimeout, APITimeoutError),
+        (httpx2.RemoteProtocolError, APIConnectionError),
+    ],
+    ids=["timeout", "connection"],
+)
+async def test_transport_error_type(
+    sync: bool,
+    transport_error: type[Exception],
+    api_error: type[APIConnectionError],
+    client: Anthropic,
+    async_client: AsyncAnthropic,
+) -> None:
+    def body() -> Iterator[bytes]:
+        yield b"event: completion\n"
+        yield b'data: {"foo":true}\n\n'
+        raise transport_error("stream interrupted")
+
+    iterator = make_stream_iterator(content=body(), sync=sync, client=client, async_client=async_client)
+
+    assert await iter_next(iterator) == {"foo": True}
+
+    with pytest.raises(api_error) as exc_info:
+        await iter_next(iterator)
+
+    assert isinstance(exc_info.value.__cause__, transport_error)
+
+
 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)