AsyncAPIResponse Class — anthropic-sdk-python Architecture

Architecture documentation for the AsyncAPIResponse class in _response.py from the anthropic-sdk-python codebase.

Class python AnthropicClient

Entity Profile

AnthropicClient→ AsyncAPIResponse Class — anthropic-sdk-python Architecture

Dependency Diagram

graph TD
  8968fc97_d184_e3da_727e_c38832cb23a5["AsyncAPIResponse"]
  17ce5647_6f06_0676_a4a5_e378a3f57cb1["BaseModel"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|extends| 17ce5647_6f06_0676_a4a5_e378a3f57cb1
  2e9c15bb_de03_aa91_93be_0e05db0a552e["_response.py"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|defined in| 2e9c15bb_de03_aa91_93be_0e05db0a552e
  cccda700_bbad_a5a1_f704_10c90917eaf0["request_id()"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|method| cccda700_bbad_a5a1_f704_10c90917eaf0
  cd4d4a4e_4ea8_4346_3324_a03f797485c7["parse()"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|method| cd4d4a4e_4ea8_4346_3324_a03f797485c7
  45221a04_d986_f437_8477_9f996a4f0720["read()"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|method| 45221a04_d986_f437_8477_9f996a4f0720
  cca0138d_3228_e9b9_dd38_fa478f426c60["text()"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|method| cca0138d_3228_e9b9_dd38_fa478f426c60
  09cc6794_47ac_ad9c_1535_b3282c0e01d7["json()"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|method| 09cc6794_47ac_ad9c_1535_b3282c0e01d7
  db9bfc2d_3e40_db1c_5a0b_2e73be279e1d["close()"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|method| db9bfc2d_3e40_db1c_5a0b_2e73be279e1d
  681781bd_bf17_30fc_c2f4_93e275856355["iter_bytes()"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|method| 681781bd_bf17_30fc_c2f4_93e275856355
  99ccd304_738a_e2fa_3366_ae7b827454c8["iter_text()"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|method| 99ccd304_738a_e2fa_3366_ae7b827454c8
  4717afb4_191d_ba23_2887_060fe6e90bf4["iter_lines()"]
  8968fc97_d184_e3da_727e_c38832cb23a5 -->|method| 4717afb4_191d_ba23_2887_060fe6e90bf4

Relationship Graph

Source Code

src/anthropic/_response.py lines 407–511

class AsyncAPIResponse(BaseAPIResponse[R]):
    @property
    def request_id(self) -> str | None:
        return self.http_response.headers.get("request-id")  # type: ignore[no-any-return]

    @overload
    async def parse(self, *, to: type[_T]) -> _T: ...

    @overload
    async def parse(self) -> R: ...

    async def parse(self, *, to: type[_T] | None = None) -> R | _T:
        """Returns the rich python representation of this response's data.

        For lower-level control, see `.read()`, `.json()`, `.iter_bytes()`.

        You can customise the type that the response is parsed into through
        the `to` argument, e.g.

        ```py
        from anthropic import BaseModel


        class MyModel(BaseModel):
            foo: str


        obj = response.parse(to=MyModel)
        print(obj.foo)
        ```

        We support parsing:
          - `BaseModel`
          - `dict`
          - `list`
          - `Union`
          - `str`
          - `httpx.Response`
        """
        cache_key = to if to is not None else self._cast_to
        cached = self._parsed_by_type.get(cache_key)
        if cached is not None:
            return cached  # type: ignore[no-any-return]

        if not self._is_sse_stream:
            await self.read()

        parsed = self._parse(to=to)
        if is_given(self._options.post_parser):
            parsed = self._options.post_parser(parsed)

        if isinstance(parsed, BaseModel):
            add_request_id(parsed, self.request_id)

        self._parsed_by_type[cache_key] = parsed
        return cast(R, parsed)

    async def read(self) -> bytes:
        """Read and return the binary response content."""
        try:
            return await self.http_response.aread()
        except httpx.StreamConsumed as exc:
            # the default error raised by httpx isn't very
            # helpful in our case so we re-raise it with
            # a different error message
            raise StreamAlreadyConsumed() from exc

    async def text(self) -> str:
        """Read and decode the response content into a string."""
        await self.read()
        return self.http_response.text

    async def json(self) -> object:
        """Read and decode the JSON response content."""
        await self.read()
        return self.http_response.json()

    async def close(self) -> None:
        """Close the response and release the connection.

        Automatically called if the response body is read to completion.