BaseClient Class — anthropic-sdk-python Architecture

Architecture documentation for the BaseClient class in _base_client.py from the anthropic-sdk-python codebase.

Class python AnthropicClient

Entity Profile

AnthropicClient→ BaseClient Class — anthropic-sdk-python Architecture

Dependency Diagram

graph TD
  842ce0be_ea56_b939_960f_13eb61230daf["BaseClient"]
  f30eca5a_2fa0_68c1_64aa_b52331a2dcc8["NotGiven"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|extends| f30eca5a_2fa0_68c1_64aa_b52331a2dcc8
  bab3bcb9_c063_9bc5_9713_fd02b9e2ca7b["ModelBuilderProtocol"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|extends| bab3bcb9_c063_9bc5_9713_fd02b9e2ca7b
  31e60ad8_cac8_652d_176d_4f7cf7dda1ad["_base_client.py"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|defined in| 31e60ad8_cac8_652d_176d_4f7cf7dda1ad
  eb7eebdb_2b36_d8b3_8894_5676fc49202f["__init__()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| eb7eebdb_2b36_d8b3_8894_5676fc49202f
  1ca77086_160d_5755_68b6_1a9bcd328911["_enforce_trailing_slash()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| 1ca77086_160d_5755_68b6_1a9bcd328911
  942c9737_25f6_9ab5_d69c_764830bc92b4["_make_status_error_from_response()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| 942c9737_25f6_9ab5_d69c_764830bc92b4
  4eb2e348_281f_08ee_df0f_a3cb13e8d3ce["_make_status_error()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| 4eb2e348_281f_08ee_df0f_a3cb13e8d3ce
  56ae8c4c_37ad_68f4_21ce_35640dd2a264["_build_headers()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| 56ae8c4c_37ad_68f4_21ce_35640dd2a264
  4d73bd62_047b_924c_524a_7c9b60a2ca98["_prepare_url()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| 4d73bd62_047b_924c_524a_7c9b60a2ca98
  b3eb540e_233b_934a_74d4_1bcc5ca67272["_make_sse_decoder()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| b3eb540e_233b_934a_74d4_1bcc5ca67272
  c1255cc8_bc9d_85e2_d7e7_d99e63ebb389["_build_request()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| c1255cc8_bc9d_85e2_d7e7_d99e63ebb389
  0a2c9ce8_a659_23ba_622c_074027fb8d75["_serialize_multipartform()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| 0a2c9ce8_a659_23ba_622c_074027fb8d75
  b55917f0_4e1e_bd62_1265_81021799d2f5["_maybe_override_cast_to()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| b55917f0_4e1e_bd62_1265_81021799d2f5
  20e6b824_c656_dbe4_65e2_12429002de16["_should_stream_response_body()"]
  842ce0be_ea56_b939_960f_13eb61230daf -->|method| 20e6b824_c656_dbe4_65e2_12429002de16

Relationship Graph

Source Code

src/anthropic/_base_client.py lines 368–835

class BaseClient(Generic[_HttpxClientT, _DefaultStreamT]):
    _client: _HttpxClientT
    _version: str
    _base_url: URL
    max_retries: int
    timeout: Union[float, Timeout, None]
    _strict_response_validation: bool
    _idempotency_header: str | None
    _default_stream_cls: type[_DefaultStreamT] | None = None

    def __init__(
        self,
        *,
        version: str,
        base_url: str | URL,
        _strict_response_validation: bool,
        max_retries: int = DEFAULT_MAX_RETRIES,
        timeout: float | Timeout | None = DEFAULT_TIMEOUT,
        custom_headers: Mapping[str, str] | None = None,
        custom_query: Mapping[str, object] | None = None,
    ) -> None:
        self._version = version
        self._base_url = self._enforce_trailing_slash(URL(base_url))
        self.max_retries = max_retries
        self.timeout = timeout
        self._custom_headers = custom_headers or {}
        self._custom_query = custom_query or {}
        self._strict_response_validation = _strict_response_validation
        self._idempotency_header = None
        self._platform: Platform | None = None

        if max_retries is None:  # pyright: ignore[reportUnnecessaryComparison]
            raise TypeError(
                "max_retries cannot be None. If you want to disable retries, pass `0`; if you want unlimited retries, pass `math.inf` or a very high number; if you want the default behavior, pass `anthropic.DEFAULT_MAX_RETRIES`"
            )

    def _enforce_trailing_slash(self, url: URL) -> URL:
        if url.raw_path.endswith(b"/"):
            return url
        return url.copy_with(raw_path=url.raw_path + b"/")

    def _make_status_error_from_response(
        self,
        response: httpx.Response,
    ) -> APIStatusError:
        if response.is_closed and not response.is_stream_consumed:
            # We can't read the response body as it has been closed
            # before it was read. This can happen if an event hook
            # raises a status error.
            body = None
            err_msg = f"Error code: {response.status_code}"
        else:
            err_text = response.text.strip()
            body = err_text

            try:
                body = json.loads(err_text)
                err_msg = f"Error code: {response.status_code} - {body}"
            except Exception:
                err_msg = err_text or f"Error code: {response.status_code}"

        return self._make_status_error(err_msg, body=body, response=response)

    def _make_status_error(
        self,
        err_msg: str,
        *,
        body: object,
        response: httpx.Response,
    ) -> _exceptions.APIStatusError:
        raise NotImplementedError()

    def _build_headers(self, options: FinalRequestOptions, *, retries_taken: int = 0) -> httpx.Headers:
        custom_headers = options.headers or {}
        headers_dict = _merge_mappings(
            {
                "x-stainless-timeout": str(options.timeout.read)
                if isinstance(options.timeout, Timeout)
                else str(options.timeout),
                **self.default_headers,
            },