From 6fbf32915da01698c8805853be4ce029bcf5d950 Mon Sep 17 00:00:00 2001 From: adhavan18 Date: Tue, 8 Sep 2026 11:10:53 +0530 Subject: [PATCH 1/2] fix(streaming): wrap mid-stream transport errors as APITimeoutError/APIConnectionError _base_client wraps the initial send so httpx transport failures surface as APIError subclasses, but Stream/AsyncStream.__stream__ iterated the response with no handling at all. A read timeout or dropped connection mid-stream escaped as a raw httpx exception, so `except anthropic.APIError` around a streaming call missed the most common streaming failure and max_retries was never consulted for it. Wraps the iteration in the same pattern _base_client already uses: httpx2.TimeoutException -> APITimeoutError, an SDK-originated AnthropicError (the in-stream error-event case) re-raised as-is, anything else -> APIConnectionError. --- src/anthropic/_streaming.py | 13 +++++++++ tests/test_streaming.py | 54 ++++++++++++++++++++++++++++++++++++- 2 files changed, 66 insertions(+), 1 deletion(-) diff --git a/src/anthropic/_streaming.py b/src/anthropic/_streaming.py index 0b98f410c..39636fc62 100644 --- a/src/anthropic/_streaming.py +++ b/src/anthropic/_streaming.py @@ -10,6 +10,7 @@ import httpx2 from ._utils import is_dict, extract_type_var_from_base +from ._exceptions import AnthropicError, APITimeoutError, APIConnectionError if TYPE_CHECKING: from ._client import Anthropic, AsyncAnthropic @@ -142,6 +143,12 @@ def __stream__(self) -> Iterator[_T]: body=body, response=self.response, ) + except httpx2.TimeoutException as err: + raise APITimeoutError(request=response.request) from err + except Exception as err: + if isinstance(err, AnthropicError): + raise + 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 +297,12 @@ async def __stream__(self) -> AsyncIterator[_T]: body=body, response=self.response, ) + except httpx2.TimeoutException as err: + raise APITimeoutError(request=response.request) from err + except Exception as err: + if isinstance(err, AnthropicError): + raise + 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() diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 42106c645..d7c5169ad 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -5,12 +5,64 @@ import httpx2 import pytest -from anthropic import Anthropic, AsyncAnthropic +from anthropic import Anthropic, AsyncAnthropic, APITimeoutError 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 + @pytest.mark.asyncio @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) From 11a8046e55e7ef14b48878358c25948aa55b94f3 Mon Sep 17 00:00:00 2001 From: adhavan18 Date: Wed, 9 Sep 2026 07:30:13 +0530 Subject: [PATCH 2/2] fix(streaming): narrow the transport-error handler to httpx2.TransportError The except Exception handler wrapped the whole iteration body, not just the read, so a defect in our own event parsing or in process_data() came out mislabeled as APIConnectionError instead of the real exception. That's not just a wrong error message: _should_retry_exception walks __cause__ and stops at the APIConnectionError node, so it reports the (deterministic, unfixable-by-retrying) bug as retryable, and a caller doing the documented retry-on-APIConnectionError pattern would loop on it forever. httpx2.TransportError is the actual family raised on the read path, and httpx2.TimeoutException is already a subclass of it, so the existing `except httpx2.TimeoutException` above still wins on ordering. With the handler narrowed to that family, the AnthropicError re-raise guard is no longer reachable and drops out. Added a regression test that injects malformed JSON into an in-stream event and asserts the resulting exception is not APIConnectionError. --- src/anthropic/_streaming.py | 10 +++------- tests/test_streaming.py | 28 +++++++++++++++++++++++++++- 2 files changed, 30 insertions(+), 8 deletions(-) diff --git a/src/anthropic/_streaming.py b/src/anthropic/_streaming.py index 39636fc62..af06ece0d 100644 --- a/src/anthropic/_streaming.py +++ b/src/anthropic/_streaming.py @@ -10,7 +10,7 @@ import httpx2 from ._utils import is_dict, extract_type_var_from_base -from ._exceptions import AnthropicError, APITimeoutError, APIConnectionError +from ._exceptions import APITimeoutError, APIConnectionError if TYPE_CHECKING: from ._client import Anthropic, AsyncAnthropic @@ -145,9 +145,7 @@ def __stream__(self) -> Iterator[_T]: ) except httpx2.TimeoutException as err: raise APITimeoutError(request=response.request) from err - except Exception as err: - if isinstance(err, AnthropicError): - raise + 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 @@ -299,9 +297,7 @@ async def __stream__(self) -> AsyncIterator[_T]: ) except httpx2.TimeoutException as err: raise APITimeoutError(request=response.request) from err - except Exception as err: - if isinstance(err, AnthropicError): - raise + 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 diff --git a/tests/test_streaming.py b/tests/test_streaming.py index d7c5169ad..5208c320c 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -5,7 +5,7 @@ import httpx2 import pytest -from anthropic import Anthropic, AsyncAnthropic, APITimeoutError +from anthropic import Anthropic, AsyncAnthropic, APITimeoutError, APIConnectionError from anthropic._streaming import Stream, AsyncStream, ServerSentEvent from anthropic._exceptions import APIStatusError @@ -64,6 +64,32 @@ async def handler(request: httpx2.Request) -> httpx2.Response: 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"]) async def test_basic(sync: bool, client: Anthropic, async_client: AsyncAnthropic) -> None: