diff --git a/src/anthropic/_streaming.py b/src/anthropic/_streaming.py index 0b98f410c..af06ece0d 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 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() diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 42106c645..5208c320c 100644 --- a/tests/test_streaming.py +++ b/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"])