Jump to content

Connect Leeroopedia MCP: Equip your AI agents to search best practices, build plans, verify code, diagnose failures, and look up hyperparameter defaults.

Implementation:Anthropics Anthropic sdk python APIResponse

From Leeroopedia
Knowledge Sources
Domains SDK_Infrastructure, Response_Parsing
Last Updated 2026-02-15 12:00 GMT

Overview

This page documents the modern API response wrapper classes -- BaseAPIResponse, APIResponse, AsyncAPIResponse, and their binary/streaming variants -- that parse httpx responses into typed Python objects with full support for sync, async, streaming, and binary content.

Description

The _response.py module implements the current (non-legacy) response processing system. BaseAPIResponse provides the core _parse() method that inspects the cast_to type to determine the decoding strategy: BaseModel for JSON, Stream/AsyncStream for SSE, JSONLDecoder for JSONL, and primitive types for text/bytes. APIResponse (sync) and AsyncAPIResponse (async) extend it with parse(), read(), text(), json(), and iterator methods.

The module also provides higher-order wrapper functions (to_raw_response_wrapper, to_streamed_response_wrapper, and their async/custom variants) that inject special headers to control how the base client processes responses. ResponseContextManager and AsyncResponseContextManager ensure proper cleanup of streamed responses. Specialized subclasses BinaryAPIResponse, AsyncBinaryAPIResponse, StreamedBinaryAPIResponse, and AsyncStreamedBinaryAPIResponse provide file-writing helpers for binary content.

Usage

This is the current response processing system that converts raw HTTP responses into the typed objects users receive from API calls. End users interact with it via .with_raw_response (returns APIResponse) and .with_streaming_response (returns StreamedBinaryAPIResponse) accessors on the client.

Code Reference

Source Location

  • Repository: Anthropic SDK Python
  • File: src/anthropic/_response.py
  • Lines: 1-872
  • Key classes: BaseAPIResponse (line 49), APIResponse (line 298), AsyncAPIResponse (line 407), BinaryAPIResponse (line 514), AsyncBinaryAPIResponse (line 539)

Signature

class BaseAPIResponse(Generic[R]):
    _cast_to: type[R]
    _client: BaseClient[Any, Any]
    _parsed_by_type: dict[type[Any], Any]
    _is_sse_stream: bool
    _stream_cls: type[Stream[Any]] | type[AsyncStream[Any]] | None
    _options: FinalRequestOptions
    http_response: httpx.Response
    retries_taken: int

    def __init__(
        self,
        *,
        raw: httpx.Response,
        cast_to: type[R],
        client: BaseClient[Any, Any],
        stream: bool,
        stream_cls: type[Stream[Any]] | type[AsyncStream[Any]] | None,
        options: FinalRequestOptions,
        retries_taken: int = 0,
    ) -> None

class APIResponse(BaseAPIResponse[R]):
    def parse(self, *, to: type[_T] | None = None) -> R | _T
    def read(self) -> bytes
    def text(self) -> str
    def json(self) -> object
    def close(self) -> None
    def iter_bytes(self, chunk_size: int | None = None) -> Iterator[bytes]
    def iter_text(self, chunk_size: int | None = None) -> Iterator[str]
    def iter_lines(self) -> Iterator[str]

class AsyncAPIResponse(BaseAPIResponse[R]):
    async def parse(self, *, to: type[_T] | None = None) -> R | _T
    async def read(self) -> bytes
    async def text(self) -> str
    async def json(self) -> object
    async def close(self) -> None
    async def iter_bytes(self, chunk_size: int | None = None) -> AsyncIterator[bytes]
    async def iter_text(self, chunk_size: int | None = None) -> AsyncIterator[str]
    async def iter_lines(self) -> AsyncIterator[str]

Import

from anthropic import APIResponse, AsyncAPIResponse
from anthropic._response import (
    BaseAPIResponse,
    BinaryAPIResponse,
    AsyncBinaryAPIResponse,
    to_raw_response_wrapper,
    to_streamed_response_wrapper,
)

I/O Contract

Inputs

Name Type Required Description
raw httpx.Response Yes The raw HTTP response from httpx
cast_to type[R] Yes The target type to parse the response into
client BaseClient[Any, Any] Yes Reference to the parent client for response data processing
stream bool Yes Whether this response is a server-sent events stream
stream_cls type[AsyncStream] | None Yes The stream class to use when stream=True
options FinalRequestOptions Yes The request options that produced this response
retries_taken int No (default: 0) Number of retries made before this response

Outputs

Name Type Description
parse() _T Parsed response object; type depends on cast_to
read() bytes Binary response content
text() str Decoded response content as string
json() object Decoded JSON response content
headers httpx.Headers Response headers
status_code int HTTP status code
request_id None The request-id response header value
elapsed datetime.timedelta Time taken for the request/response cycle

Response Properties

BaseAPIResponse exposes several convenience properties: 

@property
def headers(self) -> httpx.Headers
@property
def http_request(self) -> httpx.Request
@property
def status_code(self) -> int
@property
def url(self) -> httpx.URL
@property
def method(self) -> str
@property
def http_version(self) -> str
@property
def elapsed(self) -> datetime.timedelta
@property
def is_closed(self) -> bool

Parse Type Dispatch

The internal _parse() method handles these target types:

  • JSONLDecoder / AsyncJSONLDecoder -- Creates a line-by-line JSON decoder over the response bytes
  • Stream / AsyncStream -- Wraps the response as a server-sent events stream (when _is_sse_stream=True)
  • NoneType -- Returns None
  • str -- Returns response.text
  • bytes -- Returns response.content
  • int / float / bool -- Parses from response text
  • httpx.Response -- Returns the raw httpx response
  • BaseModel subclass -- Parses JSON and constructs via construct_type() or validate_type()

Wrapper Functions

def to_raw_response_wrapper(func: Callable[P, R]) -> Callable[P, APIResponse[R]]:
    """Wraps a sync API method to return APIResponse instead of the parsed type."""

def async_to_raw_response_wrapper(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[AsyncAPIResponse[R]]]:
    """Wraps an async API method to return AsyncAPIResponse."""

def to_streamed_response_wrapper(func: Callable[P, R]) -> Callable[P, APIResponse[R]]:
    """Wraps a sync API method to return a streaming APIResponse."""

def async_to_streamed_response_wrapper(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[AsyncAPIResponse[R]]]:
    """Wraps an async API method to return a streaming AsyncAPIResponse."""

Usage Examples

Raw Response Access

import anthropic

client = anthropic.Anthropic()

# Get the raw APIResponse wrapper
response = client.messages.with_raw_response.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)

print(response.status_code)       # 200
print(response.request_id)        # "req_xxx"
print(response.headers)           # httpx.Headers({...})

# Parse into the typed Message object
message = response.parse()
print(message.content)

Streaming Response

import anthropic

client = anthropic.Anthropic()

with client.messages.with_streaming_response.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
) as response:
    for chunk in response.iter_bytes():
        print(chunk)

Custom Parse Type

from anthropic import BaseModel

class MyModel(BaseModel):
    foo: str

# Parse the response into a custom type
obj = response.parse(to=MyModel)
print(obj.foo)

Dependencies

  • httpx -- Underlying HTTP response object and headers
  • anyio -- Async utilities for the async response variant
  • pydantic -- Type checking for BaseModel during parse dispatch

Related Pages

Replaces

Used By

Uses

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment
Retrieved from "https://leeroopedia.com/index.php?title=Implementation:Anthropics_Anthropic_sdk_python_APIResponse&oldid=4926"