Cloud Provider Deep Dive

pip install "anthropic[bedrock]"   # adds boto3 automatically

from anthropic import AnthropicBedrock
import anthropic

# ── Chunk 1: Build the client ─────────────────────────────────────
# AnthropicBedrock reads AWS credentials from the standard chain:
#   1. AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY env vars (dev)
#   2. ~/.aws/credentials file (CLI / SSO)
#   3. EC2/ECS/Lambda instance role (production — no keys in code)
client = AnthropicBedrock(
    aws_region="us-west-2",   # must match the region where you requested model access
)

# ── Chunk 2: Cross-region inference profile model ID ─────────────
# The "us." prefix means Bedrock may route to any US region for
# higher throughput. Without it, you're locked to one region and
# may hit tighter rate limits.
BEDROCK_MODEL = "us.anthropic.claude-sonnet-4-6-20250514-v1:0"

# ── Chunk 3: Make the call ────────────────────────────────────────
# GOTCHA: max_tokens is REQUIRED on Bedrock — the default (None)
# that works on the direct API raises a ValidationException here.
try:
    response = client.messages.create(
        model=BEDROCK_MODEL,
        max_tokens=1024,
        messages=[{"role": "user", "content": "Explain AWS Bedrock in one sentence."}],
    )
    print(response.content[0].text)
    print(f"Tokens: {response.usage.input_tokens} in / {response.usage.output_tokens} out")

except anthropic.APIStatusError as e:
    # Common errors:
    # 403 AccessDeniedException  → IAM policy missing bedrock:InvokeModel
    # 404 ResourceNotFoundException → wrong model ID format or model not enabled
    # 429 ThrottlingException    → rate limit hit; add exponential backoff
    print(f"Bedrock error {e.status_code}: {e.message}")

import boto3, os

# ── Bedrock Guardrails via boto3 (raw Bedrock client) ─────────────
# Use when you need guardrails but prefer boto3 over the Anthropic SDK
bedrock_rt = boto3.client("bedrock-runtime", region_name="us-west-2")

body = {
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 512,
    "messages": [{"role": "user", "content": "List medications for diabetes."}],
}
response = bedrock_rt.invoke_model(
    modelId="us.anthropic.claude-sonnet-4-6-20250514-v1:0",
    body=__import__("json").dumps(body),
    # Attach guardrail — created in Bedrock console
    guardrailIdentifier=os.environ["BEDROCK_GUARDRAIL_ID"],
    guardrailVersion="DRAFT",   # or a published version number
)
result = __import__("json").loads(response["body"].read())
print(result["content"][0]["text"])

# ── Bedrock Knowledge Bases — built-in RAG ────────────────────────
# retrieve_and_generate does: embed query → search KB → augment prompt
# → generate response.  No vector DB code on your side.
agent_rt = boto3.client("bedrock-agent-runtime", region_name="us-west-2")

kb_response = agent_rt.retrieve_and_generate(
    input={"text": "What are the Q4 revenue figures?"},
    retrieveAndGenerateConfiguration={
        "type": "KNOWLEDGE_BASE",
        "knowledgeBaseConfiguration": {
            "knowledgeBaseId": os.environ["BEDROCK_KB_ID"],
            "modelArn": "arn:aws:bedrock:us-west-2::foundation-model/anthropic.claude-sonnet-4-6-v1:0",
        },
    },
)
print(kb_response["output"]["text"])
print("Sources:", [r["location"] for r in kb_response.get("citations", [])])

from anthropic import AnthropicVertex
import anthropic, os

# ── Chunk 1: Install ──────────────────────────────────────────────
# pip install "anthropic[vertex]"
# This adds the google-auth library for ADC credential resolution.

# ── Chunk 2: Build the client ─────────────────────────────────────
# project_id: your GCP project ID (not number, not name — the slug)
# region: must be a region where Claude is available on Vertex
# Credentials are resolved automatically via ADC — no key in code.
client = AnthropicVertex(
    project_id=os.environ["GCP_PROJECT"],   # e.g. "my-project-123"
    region=os.environ.get("GCP_REGION", "us-east5"),
)

# ── Chunk 3: Vertex model ID format ──────────────────────────────
# The date stamp (@20250514) pins to a specific model checkpoint.
# Vertex exposes both "claude-sonnet-4-6" (resolves to latest) and
# "claude-sonnet-4-6@20250514" (pinned). Use pinned in production.
VERTEX_MODEL = "claude-sonnet-4-6@20250514"

# ── Chunk 4: Make the call ────────────────────────────────────────
# GOTCHA: system prompts on Vertex are passed the same way as the
# direct API — as a separate `system` parameter, NOT as a message
# with role "system". This is identical to the direct API behavior.
try:
    response = client.messages.create(
        model=VERTEX_MODEL,
        max_tokens=1024,
        system="You are a concise assistant.",
        messages=[{"role": "user", "content": "What is Vertex AI?"}],
    )
    print(response.content[0].text)

except anthropic.APIStatusError as e:
    # 403 → service account missing aiplatform.user role
    # 404 → model not enabled in Model Garden for this project/region
    print(f"Vertex error {e.status_code}: {e.message}")

from azure.ai.inference import ChatCompletionsClient
from azure.ai.inference.models import SystemMessage, UserMessage, AssistantMessage
from azure.core.credentials import AzureKeyCredential
from azure.core.exceptions import HttpResponseError
import os

# ── Chunk 1: Build the client (API key auth — dev/testing) ────────
# endpoint: the URL from your Azure AI Foundry deployment
# credential: wraps the API key from the Azure portal
# For production, replace AzureKeyCredential with ManagedIdentityCredential
# (see the "Managed Identity" section below)
client = ChatCompletionsClient(
    endpoint=os.environ["AZURE_AI_ENDPOINT"],
    credential=AzureKeyCredential(os.environ["AZURE_AI_KEY"]),
)

# ── Chunk 2: Make a call ──────────────────────────────────────────
# model: the DEPLOYMENT NAME you chose in Azure AI Foundry
# (not the Anthropic model ID — you named it during deployment)
# Messages use Azure's typed message classes or plain dicts — both work.
try:
    response = client.complete(
        model=os.environ.get("AZURE_DEPLOYMENT_NAME", "claude-sonnet-4-6"),
        messages=[
            SystemMessage(content="You are a concise assistant."),
            UserMessage(content="What is Azure AI Foundry?"),
        ],
        max_tokens=512,
    )
    # GOTCHA: response structure differs from Anthropic SDK
    # Not response.content[0].text — it's response.choices[0].message.content
    print(response.choices[0].message.content)
    print(f"Finish reason: {response.choices[0].finish_reason}")
    print(f"Tokens: {response.usage.prompt_tokens} in / {response.usage.completion_tokens} out")

except HttpResponseError as e:
    # 401 → wrong API key
    # 404 → deployment name doesn't match what you created
    # 429 → rate limit; Azure returns Retry-After header
    print(f"Azure error {e.status_code}: {e.message}")

# pip install azure-identity
from azure.ai.inference import ChatCompletionsClient
from azure.identity import DefaultAzureCredential
import os

# DefaultAzureCredential tries (in order):
#   1. AZURE_CLIENT_ID / SECRET / TENANT env vars (CI/CD)
#   2. Azure CLI (local dev: `az login`)
#   3. Managed Identity (Azure App Service / Container Apps / AKS)
# No key in code — zero secrets to rotate or accidentally commit.
credential = DefaultAzureCredential()

client = ChatCompletionsClient(
    endpoint=os.environ["AZURE_AI_ENDPOINT"],
    credential=credential,
)
# The rest of the code is identical to the API-key example above

"""
claude_client.py — provider-agnostic Claude wrapper
Switch providers by setting CLAUDE_PROVIDER=direct|bedrock|vertex|azure
"""
from __future__ import annotations
import os
from enum import Enum
from typing import Any


class Provider(str, Enum):
    DIRECT  = "direct"
    BEDROCK = "bedrock"
    VERTEX  = "vertex"
    AZURE   = "azure"


class ClaudeClient:
    """
    Uniform interface to Claude regardless of which cloud hosts it.
    Usage:
        client = ClaudeClient()              # reads CLAUDE_PROVIDER from env
        client = ClaudeClient(Provider.BEDROCK)  # explicit
        answer = client.chat("Hello!")
    """

    def __init__(self, provider: Provider | None = None):
        if provider is None:
            raw = os.environ.get("CLAUDE_PROVIDER", "direct").lower()
            provider = Provider(raw)
        self.provider = provider
        self._backend: Any = self._build_backend()

    # ── Private: build the right backend client ───────────────────
    def _build_backend(self) -> Any:
        if self.provider == Provider.DIRECT:
            from anthropic import Anthropic
            return Anthropic()

        elif self.provider == Provider.BEDROCK:
            from anthropic import AnthropicBedrock
            return AnthropicBedrock(
                aws_region=os.environ.get("AWS_REGION", "us-west-2")
            )

        elif self.provider == Provider.VERTEX:
            from anthropic import AnthropicVertex
            return AnthropicVertex(
                project_id=os.environ["GCP_PROJECT"],
                region=os.environ.get("GCP_REGION", "us-east5"),
            )

        elif self.provider == Provider.AZURE:
            from azure.ai.inference import ChatCompletionsClient
            from azure.core.credentials import AzureKeyCredential
            return ChatCompletionsClient(
                endpoint=os.environ["AZURE_AI_ENDPOINT"],
                credential=AzureKeyCredential(os.environ["AZURE_AI_KEY"]),
            )

        raise ValueError(f"Unknown provider: {self.provider}")

    # ── Private: resolve the right model ID for this provider ─────
    def _model_id(self) -> str:
        if self.provider == Provider.DIRECT:
            return os.environ.get("CLAUDE_MODEL", "claude-sonnet-4-6")
        elif self.provider == Provider.BEDROCK:
            return os.environ.get(
                "BEDROCK_MODEL_ID",
                "us.anthropic.claude-sonnet-4-6-20250514-v1:0"
            )
        elif self.provider == Provider.VERTEX:
            return os.environ.get("VERTEX_MODEL_ID", "claude-sonnet-4-6@20250514")
        elif self.provider == Provider.AZURE:
            return os.environ.get("AZURE_DEPLOYMENT_NAME", "claude-sonnet-4-6")
        return "claude-sonnet-4-6"

    # ── Public: the only method your application code calls ───────
    def chat(self, message: str, system: str | None = None, max_tokens: int = 1024) -> str:
        model = self._model_id()

        if self.provider == Provider.AZURE:
            from azure.ai.inference.models import SystemMessage, UserMessage
            msgs = []
            if system:
                msgs.append(SystemMessage(content=system))
            msgs.append(UserMessage(content=message))
            resp = self._backend.complete(model=model, messages=msgs, max_tokens=max_tokens)
            return resp.choices[0].message.content

        else:  # direct, bedrock, vertex — all use Anthropic SDK format
            kwargs: dict = dict(model=model, max_tokens=max_tokens,
                                messages=[{"role": "user", "content": message}])
            if system:
                kwargs["system"] = system
            resp = self._backend.messages.create(**kwargs)
            return resp.content[0].text


# ── Usage ─────────────────────────────────────────────────────────
if __name__ == "__main__":
    import sys
    provider_name = sys.argv[1] if len(sys.argv) > 1 else None
    p = Provider(provider_name) if provider_name else None
    client = ClaudeClient(p)
    print(f"Provider: {client.provider.value}")
    answer = client.chat(
        "In one sentence, what is the main benefit of using Claude through a cloud provider?",
        system="You are concise."
    )
    print(f"Answer: {answer}")

mkdir cloud-provider-lab && cd cloud-provider-lab
python -m venv venv
# macOS/Linux:
source venv/bin/activate
# Windows:
# venv\Scripts\activate

# Install all providers — only the ones you configure will be used
pip install "anthropic[bedrock,vertex]" azure-ai-inference azure-core azure-identity python-dotenv