fix(streaming): wrap mid-stream transport errors as APITimeoutError/APIConnectionError

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
@@ -142,6 +143,10 @@ def __stream__(self) -> Iterator[_T]:
                         body=body,
                         response=self.response,
                     )
+        except httpx2.TimeoutException as err:
+            raise APITimeoutError(request=response.request) from err
+        except httpx2.TransportError as err:
+            raise APIConnectionError(request=response.request) from err
         finally:
             # Ensure the response is closed even if the consumer doesn't read all data
             response.close()
@@ -290,6 +295,10 @@ async def __stream__(self) -> AsyncIterator[_T]:
                         body=body,
                         response=self.response,
                     )
+        except httpx2.TimeoutException as err:
+            raise APITimeoutError(request=response.request) from err
+        except httpx2.TransportError as err:
+            raise APIConnectionError(request=response.request) from err
         finally:
             # Ensure the response is closed even if the consumer doesn't read all data
             await response.aclose()

tests/test_streaming.py

@@ -5,12 +5,90 @@
 import httpx2
 import pytest
 
-from anthropic import Anthropic, AsyncAnthropic
+from anthropic import Anthropic, AsyncAnthropic, APITimeoutError, APIConnectionError
 from anthropic._streaming import Stream, AsyncStream, ServerSentEvent
 from anthropic._exceptions import APIStatusError
 
 _T = TypeVar("_T")
 
+_FIRST_EVENT = (
+    b'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1","type":"message",'
+    b'"role":"assistant","model":"claude-sonnet-4-6","content":[],"stop_reason":null,"stop_sequence":null,'
+    b'"usage":{"input_tokens":1,"output_tokens":0}}}\n\n'
+)
+
+
+class _DiesMidStream(httpx2.SyncByteStream):
+    def __iter__(self) -> Iterator[bytes]:
+        yield _FIRST_EVENT
+        raise httpx2.ReadTimeout("timed out while reading the stream")
+
+
+class _AsyncDiesMidStream(httpx2.AsyncByteStream):
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        yield _FIRST_EVENT
+        raise httpx2.ReadTimeout("timed out while reading the stream")
+
+
+def test_sync_stream_wraps_mid_stream_transport_error() -> None:
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, headers={"content-type": "text/event-stream"}, stream=_DiesMidStream())
+
+    client = Anthropic(
+        api_key="My API Key",
+        http_client=httpx2.Client(transport=httpx2.MockTransport(handler)),
+        max_retries=0,
+    )
+
+    with pytest.raises(APITimeoutError):
+        with client.messages.stream(model="claude-sonnet-4-6", max_tokens=8, messages=[{"role": "user", "content": "hi"}]) as s:
+            for _ in s:
+                pass
+
+
+async def test_async_stream_wraps_mid_stream_transport_error() -> None:
+    async def handler(request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, headers={"content-type": "text/event-stream"}, stream=_AsyncDiesMidStream())
+
+    client = AsyncAnthropic(
+        api_key="My API Key",
+        http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(handler)),
+        max_retries=0,
+    )
+
+    with pytest.raises(APITimeoutError):
+        async with client.messages.stream(
+            model="claude-sonnet-4-6", max_tokens=8, messages=[{"role": "user", "content": "hi"}]
+        ) as s:
+            async for _ in s:
+                pass
+
+
+def test_sync_stream_does_not_mislabel_a_parse_error_as_a_connection_error() -> None:
+    """A bug in parsing an in-stream event (malformed JSON here) is a defect in our own
+    code, not a transport failure — it must not come out the other end looking retryable."""
+
+    def body() -> Iterator[bytes]:
+        yield b"event: message_start\n"
+        yield b"data: {not valid json\n"
+        yield b"\n"
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, headers={"content-type": "text/event-stream"}, content=body())
+
+    client = Anthropic(
+        api_key="My API Key",
+        http_client=httpx2.Client(transport=httpx2.MockTransport(handler)),
+        max_retries=0,
+    )
+
+    with pytest.raises(Exception) as exc_info:
+        with client.messages.stream(model="claude-sonnet-4-6", max_tokens=8, messages=[{"role": "user", "content": "hi"}]) as s:
+            for _ in s:
+                pass
+
+    assert not isinstance(exc_info.value, APIConnectionError)
+
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])