feat(client): add error type field to APIStatusError (#1587)

karpetrosyan · stainless-app[bot] · commit dd563c031c2a · 2026-03-26T20:51:40.000Z
diff --git a/src/anthropic/_exceptions.py b/src/anthropic/_exceptions.py
@@ -2,10 +2,14 @@
 
 from __future__ import annotations
 
+from typing import Union, cast
 from typing_extensions import Literal
 
 import httpx
 
+from ._utils import is_dict
+from .types.shared.error_type import ErrorType
+
 __all__ = [
     "BadRequestError",
     "AuthenticationError",
@@ -60,13 +64,20 @@ class APIStatusError(APIError):
     response: httpx.Response
     status_code: int
     request_id: str | None
+    type: ErrorType | None
 
     def __init__(self, message: str, *, response: httpx.Response, body: object | None) -> None:
         super().__init__(message, response.request, body=body)
         self.response = response
         self.status_code = response.status_code
         self.request_id = response.headers.get("request-id")
 
+        self.type = None
+        if is_dict(body):
+            error = body.get("error")
+            if is_dict(error):
+                self.type = cast(Union[ErrorType, None], error.get("type"))
+
 
 class APIConnectionError(APIError):
     def __init__(self, *, message: str = "Connection error.", request: httpx.Request) -> None:
diff --git a/tests/test_client.py b/tests/test_client.py
@@ -1113,6 +1113,23 @@ def test_follow_redirects_disabled(self, respx_mock: MockRouter, client: Anthrop
         assert exc_info.value.response.status_code == 302
         assert exc_info.value.response.headers["Location"] == f"{base_url}/redirected"
 
+    @pytest.mark.respx(base_url=base_url)
+    def test_status_error_type_field(self, respx_mock: MockRouter, client: Anthropic) -> None:
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx.Response(
+                400,
+                json={"type": "error", "error": {"type": "invalid_request_error", "message": "Bad request"}},
+            )
+        )
+        with pytest.raises(APIStatusError) as exc_info:
+            client.messages.create(
+                max_tokens=1024,
+                messages=[{"role": "user", "content": "Hello"}],
+                model="claude-opus-4-6",
+            )
+        assert exc_info.value.type == "invalid_request_error"
+        assert exc_info.value.status_code == 400
+
 
 class TestAsyncAnthropic:
     @pytest.mark.respx(base_url=base_url)
@@ -2125,3 +2142,20 @@ async def test_follow_redirects_disabled(self, respx_mock: MockRouter, async_cli
 
         assert exc_info.value.response.status_code == 302
         assert exc_info.value.response.headers["Location"] == f"{base_url}/redirected"
+
+    @pytest.mark.respx(base_url=base_url)
+    async def test_status_error_type_field(self, respx_mock: MockRouter, async_client: AsyncAnthropic) -> None:
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx.Response(
+                400,
+                json={"type": "error", "error": {"type": "invalid_request_error", "message": "Bad request"}},
+            )
+        )
+        with pytest.raises(APIStatusError) as exc_info:
+            await async_client.messages.create(
+                max_tokens=1024,
+                messages=[{"role": "user", "content": "Hello"}],
+                model="claude-opus-4-6",
+            )
+        assert exc_info.value.type == "invalid_request_error"
+        assert exc_info.value.status_code == 400
diff --git a/tests/test_streaming.py b/tests/test_streaming.py
@@ -1,12 +1,15 @@
 from __future__ import annotations
 
-from typing import Iterator, AsyncIterator
+from typing import TypeVar, Iterator, AsyncIterator
 
 import httpx
 import pytest
 
 from anthropic import Anthropic, AsyncAnthropic
 from anthropic._streaming import Stream, AsyncStream, ServerSentEvent
+from anthropic._exceptions import APIStatusError
+
+_T = TypeVar("_T")
 
 
 @pytest.mark.asyncio
@@ -216,6 +219,25 @@ def body() -> Iterator[bytes]:
     assert sse.json() == {"content": "известни"}
 
 
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+async def test_error_type(
+    sync: bool,
+    client: Anthropic,
+    async_client: AsyncAnthropic,
+) -> None:
+    def body() -> Iterator[bytes]:
+        yield b"event: error\n"
+        yield b'data: {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}\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 exc_info.value.type == "overloaded_error"
+    assert "Overloaded" in str(exc_info.value)
+
+
 def test_isinstance_check(client: Anthropic, async_client: AsyncAnthropic) -> None:
     async_stream = AsyncStream(cast_to=object, client=async_client, response=httpx.Response(200, content=b"foo"))
     assert isinstance(async_stream, AsyncStream)
@@ -229,7 +251,7 @@ async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]:
         yield chunk
 
 
-async def iter_next(iter: Iterator[ServerSentEvent] | AsyncIterator[ServerSentEvent]) -> ServerSentEvent:
+async def iter_next(iter: Iterator[_T] | AsyncIterator[_T]) -> _T:
     if isinstance(iter, AsyncIterator):
         return await iter.__anext__()
 
@@ -254,3 +276,27 @@ def make_event_iterator(
     return AsyncStream(
         cast_to=object, client=async_client, response=httpx.Response(200, content=to_aiter(content))
     )._iter_events()
+
+
+# Unlike make_event_iterator which only parses SSE events using _iter_events(),
+# this helper uses __stream__() to process the full stream pipeline including
+# parsing message objects and converting error events into raised exceptions.
+def make_stream_iterator(
+    content: Iterator[bytes],
+    *,
+    sync: bool,
+    client: Anthropic,
+    async_client: AsyncAnthropic,
+) -> AsyncIterator[object] | Iterator[object]:
+    if sync:
+        return Stream(
+            cast_to=object,
+            client=client,
+            response=httpx.Response(200, content=content, request=httpx.Request("GET", "https://example.com")),
+        ).__stream__()
+
+    return AsyncStream(
+        cast_to=object,
+        client=async_client,
+        response=httpx.Response(200, content=to_aiter(content), request=httpx.Request("GET", "https://example.com")),
+    ).__stream__()
