AnthropicBedrock Class — anthropic-sdk-python Architecture

Architecture documentation for the AnthropicBedrock class in _client.py from the anthropic-sdk-python codebase.

Class python AnthropicClient

Entity Profile

AnthropicClient→ AnthropicBedrock Class — anthropic-sdk-python Architecture

Dependency Diagram

graph TD
  52b957ca_f6e2_8cc2_8b0b_1f00048d1b4f["AnthropicBedrock"]
  4b46ed95_ff71_ea5d_7534_6929dc929bdb["SyncAPIClient"]
  52b957ca_f6e2_8cc2_8b0b_1f00048d1b4f -->|extends| 4b46ed95_ff71_ea5d_7534_6929dc929bdb
  f30eca5a_2fa0_68c1_64aa_b52331a2dcc8["NotGiven"]
  52b957ca_f6e2_8cc2_8b0b_1f00048d1b4f -->|extends| f30eca5a_2fa0_68c1_64aa_b52331a2dcc8
  cbdd42fd_75f9_7cc6_3f3b_b8d939e4c157["_client.py"]
  52b957ca_f6e2_8cc2_8b0b_1f00048d1b4f -->|defined in| cbdd42fd_75f9_7cc6_3f3b_b8d939e4c157
  13292190_c332_c947_2781_45e82e05e90d["__init__()"]
  52b957ca_f6e2_8cc2_8b0b_1f00048d1b4f -->|method| 13292190_c332_c947_2781_45e82e05e90d
  33d04623_5f2b_aedc_ba7c_98df640a6e86["_make_sse_decoder()"]
  52b957ca_f6e2_8cc2_8b0b_1f00048d1b4f -->|method| 33d04623_5f2b_aedc_ba7c_98df640a6e86
  edd748c3_bd8e_a7a2_8c66_e4824029d1bf["_prepare_options()"]
  52b957ca_f6e2_8cc2_8b0b_1f00048d1b4f -->|method| edd748c3_bd8e_a7a2_8c66_e4824029d1bf
  1a84c61a_f43c_16d5_287c_446adc194df6["_prepare_request()"]
  52b957ca_f6e2_8cc2_8b0b_1f00048d1b4f -->|method| 1a84c61a_f43c_16d5_287c_446adc194df6
  7d88f9ba_782d_b203_6007_9c1dee5c62ea["copy()"]
  52b957ca_f6e2_8cc2_8b0b_1f00048d1b4f -->|method| 7d88f9ba_782d_b203_6007_9c1dee5c62ea

Relationship Graph

Source Code

src/anthropic/lib/bedrock/_client.py lines 131–270

class AnthropicBedrock(BaseBedrockClient[httpx.Client, Stream[Any]], SyncAPIClient):
    messages: Messages
    completions: Completions
    beta: Beta

    def __init__(
        self,
        aws_secret_key: str | None = None,
        aws_access_key: str | None = None,
        aws_region: str | None = None,
        aws_profile: str | None = None,
        aws_session_token: str | None = None,
        base_url: str | httpx.URL | None = None,
        timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
        max_retries: int = DEFAULT_MAX_RETRIES,
        default_headers: Mapping[str, str] | None = None,
        default_query: Mapping[str, object] | None = None,
        # Configure a custom httpx client. See the [httpx documentation](https://www.python-httpx.org/api/#client) for more details.
        http_client: httpx.Client | None = None,
        # Enable or disable schema validation for data returned by the API.
        # When enabled an error APIResponseValidationError is raised
        # if the API responds with invalid data for the expected schema.
        #
        # This parameter may be removed or changed in the future.
        # If you rely on this feature, please open a GitHub issue
        # outlining your use-case to help us decide if it should be
        # part of our public interface in the future.
        _strict_response_validation: bool = False,
    ) -> None:
        self.aws_secret_key = aws_secret_key

        self.aws_access_key = aws_access_key

        self.aws_region = _infer_region() if aws_region is None else aws_region
        self.aws_profile = aws_profile

        self.aws_session_token = aws_session_token

        if base_url is None:
            base_url = os.environ.get("ANTHROPIC_BEDROCK_BASE_URL")
        if base_url is None:
            base_url = f"https://bedrock-runtime.{self.aws_region}.amazonaws.com"

        super().__init__(
            version=__version__,
            base_url=base_url,
            timeout=timeout,
            max_retries=max_retries,
            custom_headers=default_headers,
            custom_query=default_query,
            http_client=http_client,
            _strict_response_validation=_strict_response_validation,
        )

        self.beta = Beta(self)
        self.messages = Messages(self)
        self.completions = Completions(self)

    @override
    def _make_sse_decoder(self) -> AWSEventStreamDecoder:
        return AWSEventStreamDecoder()

    @override
    def _prepare_options(self, options: FinalRequestOptions) -> FinalRequestOptions:
        return _prepare_options(options)

    @override
    def _prepare_request(self, request: httpx.Request) -> None:
        from ._auth import get_auth_headers

        data = request.read().decode()

        headers = get_auth_headers(
            method=request.method,
            url=str(request.url),
            headers=request.headers,
            aws_access_key=self.aws_access_key,
            aws_secret_key=self.aws_secret_key,
            aws_session_token=self.aws_session_token,
            region=self.aws_region or "us-east-1",
            profile=self.aws_profile,