From 09458501d11d3edc029dca991f36846252207268 Mon Sep 17 00:00:00 2001 From: AP Date: Wed, 21 Oct 2026 01:07:56 +0530 Subject: [PATCH 1/2] fix(streaming): wrap mid-stream transport errors as APIError subclasses _streaming.py iterated the response with no error handling, so a read timeout or dropped connection mid-stream escaped as a raw httpx2 exception. Translate it the same way the initial send does: httpx2.TimeoutException -> APITimeoutError, other transport errors -> APIConnectionError, so callers see the documented error taxonomy. --- src/anthropic/_streaming.py | 25 +++++++++++++++++++--- tests/test_streaming.py | 41 ++++++++++++++++++++++++++++++++++++- 2 files changed, 62 insertions(+), 4 deletions(-) diff --git a/src/anthropic/_streaming.py b/src/anthropic/_streaming.py index ae718ce10..c0bafd750 100644 --- a/src/anthropic/_streaming.py +++ b/src/anthropic/_streaming.py @@ -13,6 +13,7 @@ from ._utils import extract_type_var_from_base from ._compat import cached_property +from ._exceptions import AnthropicError, APITimeoutError, APIConnectionError if TYPE_CHECKING: from ._client import Anthropic, AsyncAnthropic @@ -68,7 +69,16 @@ 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 Exception as err: + if isinstance(err, AnthropicError): + # SDK-originated errors already carry their own type; don't wrap. + raise + + raise APIConnectionError(request=self.response.request) from err @staticmethod def raw_events(response: httpx2.Response) -> Iterator[ServerSentEvent]: @@ -196,8 +206,17 @@ 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 Exception as err: + if isinstance(err, AnthropicError): + # SDK-originated errors already carry their own type; don't wrap. + raise + + raise APIConnectionError(request=self.response.request) from err @staticmethod def raw_events(response: httpx2.Response) -> AsyncIterator[ServerSentEvent]: diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 5efd07a67..da9f58494 100644 --- a/tests/test_streaming.py +++ b/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,45 @@ 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) + + 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) From dffaba4b2698d74a09d8a212e9417761188603ed Mon Sep 17 00:00:00 2001 From: AP Date: Wed, 21 Oct 2026 01:08:04 +0530 Subject: [PATCH 2/2] fix(streaming): narrow mid-stream error wrapping to httpx transport errors The broad `except Exception` in `_iter_events` also caught decoder failures, so e.g. a `UnicodeDecodeError` from a malformed SSE line (or an exception from a custom SSEBytesDecoder) was relabeled as `APIConnectionError`. Scope the non-timeout handler to `httpx2.TransportError` so decoder errors keep their original identity, and add a regression test for it. --- src/anthropic/_streaming.py | 14 +++----------- tests/test_streaming.py | 19 +++++++++++++++++++ 2 files changed, 22 insertions(+), 11 deletions(-) diff --git a/src/anthropic/_streaming.py b/src/anthropic/_streaming.py index c0bafd750..0364140c4 100644 --- a/src/anthropic/_streaming.py +++ b/src/anthropic/_streaming.py @@ -13,7 +13,7 @@ from ._utils import extract_type_var_from_base from ._compat import cached_property -from ._exceptions import AnthropicError, APITimeoutError, APIConnectionError +from ._exceptions import APITimeoutError, APIConnectionError if TYPE_CHECKING: from ._client import Anthropic, AsyncAnthropic @@ -73,11 +73,7 @@ def _iter_events(self) -> Iterator[ServerSentEvent]: yield from self._decoder.iter_bytes(self.response.iter_bytes()) except httpx2.TimeoutException as err: raise APITimeoutError(request=self.response.request) from err - except Exception as err: - if isinstance(err, AnthropicError): - # SDK-originated errors already carry their own type; don't wrap. - raise - + except httpx2.TransportError as err: raise APIConnectionError(request=self.response.request) from err @staticmethod @@ -211,11 +207,7 @@ async def _iter_events(self) -> AsyncIterator[ServerSentEvent]: yield sse except httpx2.TimeoutException as err: raise APITimeoutError(request=self.response.request) from err - except Exception as err: - if isinstance(err, AnthropicError): - # SDK-originated errors already carry their own type; don't wrap. - raise - + except httpx2.TransportError as err: raise APIConnectionError(request=self.response.request) from err @staticmethod diff --git a/tests/test_streaming.py b/tests/test_streaming.py index da9f58494..3b0b18acc 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -387,6 +387,25 @@ def body() -> Iterator[bytes]: 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)