From 19d42ad0280b6a9593b067c074a3f0282e3703ae Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Mon, 7 Sep 2026 16:28:14 +0200 Subject: [PATCH] fix(streaming): raise the exception class matching an in-stream error's type An error event inside a stream is raised through the status-code dispatch with the stream's own 200 response, so it was always the bare APIStatusError, never OverloadedError, RateLimitError or the other subclasses. The class is now picked from the error type in the body; the status code and the response stay the real ones, and error.type keeps working as before. Unknown types still raise the bare class. --- src/anthropic/_streaming.py | 41 +++++++++++++++++++++-- tests/test_streaming.py | 65 ++++++++++++++++++++++++++++++++++++- 2 files changed, 103 insertions(+), 3 deletions(-) diff --git a/src/anthropic/_streaming.py b/src/anthropic/_streaming.py index 0b98f410c..1dfc5182d 100644 --- a/src/anthropic/_streaming.py +++ b/src/anthropic/_streaming.py @@ -10,12 +10,47 @@ import httpx2 from ._utils import is_dict, extract_type_var_from_base +from ._exceptions import ( + NotFoundError, + APIStatusError, + RateLimitError, + BadRequestError, + OverloadedError, + AuthenticationError, + InternalServerError, + PermissionDeniedError, +) if TYPE_CHECKING: from ._client import Anthropic, AsyncAnthropic from ._models import FinalRequestOptions +# An error event inside a stream arrives on the stream's own HTTP response, whose status is 200, +# so the status code cannot pick the exception class the way it does for a failed request. +# The error type in the body can; the status code and the response stay the real ones. +_STREAM_ERROR_TYPE_TO_CLASS: dict[str, type[APIStatusError]] = { + "invalid_request_error": BadRequestError, + "authentication_error": AuthenticationError, + "permission_error": PermissionDeniedError, + "not_found_error": NotFoundError, + "rate_limit_error": RateLimitError, + "overloaded_error": OverloadedError, + "api_error": InternalServerError, +} + + +def _make_stream_status_error( + client: Anthropic | AsyncAnthropic, err_msg: str, *, body: object, response: httpx2.Response +) -> APIStatusError: + error = body.get("error") if is_dict(body) else None + error_type = error.get("type") if is_dict(error) else None + error_class = _STREAM_ERROR_TYPE_TO_CLASS.get(error_type) if isinstance(error_type, str) else None + if error_class is not None: + return error_class(err_msg, response=response, body=body) + return client._make_status_error(err_msg, body=body, response=response) + + _T = TypeVar("_T") @@ -137,7 +172,8 @@ def __stream__(self) -> Iterator[_T]: except Exception: err_msg = sse.data or f"Error code: {response.status_code}" - raise self._client._make_status_error( + raise _make_stream_status_error( + self._client, err_msg, body=body, response=self.response, @@ -285,7 +321,8 @@ async def __stream__(self) -> AsyncIterator[_T]: except Exception: err_msg = sse.data or f"Error code: {response.status_code}" - raise self._client._make_status_error( + raise _make_stream_status_error( + self._client, err_msg, body=body, response=self.response, diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 42106c645..df4ab6266 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -7,7 +7,16 @@ from anthropic import Anthropic, AsyncAnthropic from anthropic._streaming import Stream, AsyncStream, ServerSentEvent -from anthropic._exceptions import APIStatusError +from anthropic._exceptions import ( + NotFoundError, + APIStatusError, + RateLimitError, + BadRequestError, + OverloadedError, + AuthenticationError, + InternalServerError, + PermissionDeniedError, +) _T = TypeVar("_T") @@ -238,6 +247,60 @@ def body() -> Iterator[bytes]: assert "Overloaded" in str(exc_info.value) +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +@pytest.mark.parametrize( + "error_type,error_class", + [ + ("invalid_request_error", BadRequestError), + ("authentication_error", AuthenticationError), + ("permission_error", PermissionDeniedError), + ("not_found_error", NotFoundError), + ("rate_limit_error", RateLimitError), + ("overloaded_error", OverloadedError), + ("api_error", InternalServerError), + ], +) +async def test_error_class_from_type( + sync: bool, + error_type: str, + error_class: type[APIStatusError], + client: Anthropic, + async_client: AsyncAnthropic, +) -> None: + def body() -> Iterator[bytes]: + yield b"event: error\n" + yield f'data: {{"type": "error", "error": {{"type": "{error_type}", "message": "boom"}}}}\n\n'.encode() + + iterator = make_stream_iterator(content=body(), sync=sync, client=client, async_client=async_client) + + with pytest.raises(error_class) as exc_info: + await iter_next(iterator) + + # the response is the stream's own, so its status is reported as is + assert exc_info.value.status_code == 200 + assert exc_info.value.type == error_type + assert "boom" in str(exc_info.value) + + +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +async def test_error_of_unknown_type_stays_generic( + sync: bool, + client: Anthropic, + async_client: AsyncAnthropic, +) -> None: + def body() -> Iterator[bytes]: + yield b"event: error\n" + yield b'data: {"type": "error", "error": {"type": "billing_error", "message": "boom"}}\n\n' + + iterator = make_stream_iterator(content=body(), sync=sync, client=client, async_client=async_client) + + with pytest.raises(APIStatusError) as exc_info: + await iter_next(iterator) + + assert type(exc_info.value) is APIStatusError + assert exc_info.value.type == "billing_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)