Claude API in Python cheat sheet

import anthropic

# Reads ANTHROPIC_API_KEY from the environment.
client = anthropic.Anthropic()

message = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=1024,
    system="Answer in one short paragraph.",
    messages=[{"role": "user", "content": "What is a token?"}],
)

# content[0] can be a thinking block, so pick the text blocks.
text = "".join(b.text for b in message.content if b.type == "text")
print(text)
print(message.stop_reason, message.usage.input_tokens, message.usage.output_tokens)
print(message._request_id)

import anthropic

client = anthropic.Anthropic()
history: list[anthropic.types.MessageParam] = []


def ask(prompt: str) -> str:
    history.append({"role": "user", "content": prompt})
    reply = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1024,
        messages=history,  # the API is stateless: resend everything
    )
    # Send Claude's blocks back unchanged so thinking blocks stay valid.
    history.append({"role": "assistant", "content": reply.content})
    return "".join(b.text for b in reply.content if b.type == "text")


print(ask("My name is Sam. Pick a number from 1 to 10."))
print(ask("What is my name, and what number did you pick?"))

import anthropic

client = anthropic.Anthropic()

stream = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=1024,
    stream=True,  # returns raw events, no accumulated Message
    messages=[{"role": "user", "content": "Say hello in three languages."}],
)

for event in stream:
    if event.type == "message_start":
        print("input tokens:", event.message.usage.input_tokens)
    elif event.type == "content_block_delta":
        delta = event.delta
        if delta.type == "text_delta":
            print(delta.text, end="", flush=True)
        elif delta.type == "input_json_delta":  # partial JSON for a tool_use block
            print(delta.partial_json, end="")
        elif delta.type == "thinking_delta":
            print(delta.thinking, end="")
    elif event.type == "message_delta":  # usage here is cumulative
        print(f"\nstop_reason={event.delta.stop_reason} output={event.usage.output_tokens}")
    elif event.type == "message_stop":
        print("done")

import asyncio

import anthropic

client = anthropic.AsyncAnthropic()


async def main() -> None:
    async with client.messages.stream(
        model="claude-sonnet-5-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Say hello there!"}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
        message = await stream.get_final_message()
    print(f"\n{message.usage.output_tokens} output tokens")


asyncio.run(main())

import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

client = anthropic.Anthropic()
questions = ["What is a vector database?", "What is RAG?", "What is MCP?"]

batch = client.messages.batches.create(
    requests=[
        Request(
            custom_id=f"q-{i}",  # 1 to 64 chars: letters, digits, - and _
            params=MessageCreateParamsNonStreaming(
                model="claude-haiku-5-5",
                max_tokens=300,
                messages=[{"role": "user", "content": question}],
            ),
        )
        for i, question in enumerate(questions)
    ]
)
print(batch.id, batch.processing_status, batch.expires_at)  # in_progress, expires in 24h

import sys
import time

import anthropic

client = anthropic.Anthropic()
batch_id = sys.argv[1]  # msgbatch_...

while client.messages.batches.retrieve(batch_id).processing_status != "ended":
    time.sleep(60)  # poll; client.messages.batches.cancel(batch_id) stops it early

for item in client.messages.batches.results(batch_id):  # order is not guaranteed
    outcome = item.result
    if outcome.type == "succeeded":
        text = "".join(b.text for b in outcome.message.content if b.type == "text")
        print(item.custom_id, text[:80])
    else:  # errored, canceled or expired: none of these are billed
        print(item.custom_id, outcome.type)

from pathlib import Path

import anthropic

client = anthropic.Anthropic()

# Out of beta: client.files, no beta header. 500 MB per file.
uploaded = client.files.upload(file=Path("report.pdf"), expires_in_seconds=86400)
print(uploaded.id, uploaded.size_bytes, uploaded.expires_at)

response = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=512,
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "document", "source": {"type": "file", "file_id": uploaded.id}},
                {"type": "text", "text": "List the three biggest risks."},
            ],
        }
    ],
)
print("".join(b.text for b in response.content if b.type == "text"))
client.files.delete(uploaded.id)

import base64
from pathlib import Path

import anthropic

client = anthropic.Anthropic()
data = base64.standard_b64encode(Path("chart.png").read_bytes()).decode()

response = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=512,
    messages=[
        {
            "role": "user",
            "content": [
                # Image first, then the question. Or use {"type": "url", "url": "https://..."}.
                {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": data}},
                {"type": "text", "text": "What trend does this chart show?"},
            ],
        }
    ],
)
print("".join(b.text for b in response.content if b.type == "text"))
print("input tokens:", response.usage.input_tokens)

import base64
from pathlib import Path

import anthropic

client = anthropic.Anthropic()
pdf = base64.standard_b64encode(Path("contract.pdf").read_bytes()).decode()

response = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "document",  # PDF first, then the question
                    "source": {"type": "base64", "media_type": "application/pdf", "data": pdf},
                    "cache_control": {"type": "ephemeral"},  # reuse it across questions
                },
                {"type": "text", "text": "What are the termination terms? Cite page numbers."},
            ],
        }
    ],
)
print("".join(b.text for b in response.content if b.type == "text"))

import anthropic

client = anthropic.Anthropic()

# Free to call, but it has its own requests-per-minute limit.
count = client.messages.count_tokens(
    model="claude-haiku-5-5",
    system="You are a careful reviewer.",
    messages=[{"role": "user", "content": "Review this paragraph for clarity."}],
)
print(count.input_tokens)

# Recount per model: Claude 4.7 and later use a newer tokenizer.
if count.input_tokens > 100_000:
    raise SystemExit("Prompt too long for the Haiku 5.5 low price band")

import anthropic

# USD per million tokens: (input, output, cache-read multiplier). Checked Oct 28, 2026.
PRICES = {
    "claude-fable-5-1": (10.00, 50.00, 0.025),
    "claude-opus-5-5": (4.00, 20.00, 0.05),
    "claude-sonnet-5-5": (2.00, 10.00, 0.05),
    "claude-haiku-5-5": (0.10, 0.50, 0.10),  # 5x above 100,000 prompt tokens
}


def cost_usd(model, usage, batch=False):
    p_in, p_out, read_mult = PRICES[model]
    split = usage.cache_creation  # may be None
    w5 = split.ephemeral_5m_input_tokens if split else usage.cache_creation_input_tokens or 0
    w1 = split.ephemeral_1h_input_tokens if split else 0
    read = usage.cache_read_input_tokens or 0
    if model == "claude-haiku-5-5" and usage.input_tokens + read + w5 + w1 > 100_000:
        p_in, p_out = p_in * 5, p_out * 5
    total = usage.input_tokens * p_in + w5 * p_in * 1.25 + w1 * p_in * 2
    total += read * p_in * read_mult + usage.output_tokens * p_out
    total *= (0.5 if batch else 1) * (1.1 if usage.inference_geo == "us" else 1)
    return total / 1_000_000


client = anthropic.Anthropic()
model = "claude-sonnet-5-5"
msg = client.messages.create(
    model=model, max_tokens=256, messages=[{"role": "user", "content": "Say hi."}]
)
print(f"${cost_usd(model, msg.usage):.6f}")

import anthropic

# Defaults: max_retries=2, timeout=600s (10 minutes).
client = anthropic.Anthropic(max_retries=4, timeout=60.0)

try:
    message = client.with_options(max_retries=1).messages.create(
        model="claude-haiku-5-5",
        max_tokens=256,
        messages=[{"role": "user", "content": "Say hello."}],
    )
    print(message._request_id)
except anthropic.RateLimitError as e:  # 429
    print("slow down", e.response.headers.get("retry-after"), e.request_id)
except anthropic.OverloadedError as e:  # 529
    print("API overloaded", e.request_id)
except anthropic.APITimeoutError:  # subclass of APIConnectionError, so catch it first
    print("timed out")
except anthropic.APIConnectionError as e:
    print("network problem", e.__cause__)
except anthropic.APIStatusError as e:  # any other 4xx or 5xx
    print(e.status_code, e.request_id)

import os

from anthropic import AnthropicBedrockMantle, AnthropicFoundry, AnthropicVertex

# pip install "anthropic[bedrock]" "anthropic[vertex]"; Foundry ships in the base package.
bedrock = AnthropicBedrockMantle(aws_region="us-east-1")
vertex = AnthropicVertex(project_id=os.environ["GOOGLE_CLOUD_PROJECT"], region="global")
foundry = AnthropicFoundry(
    api_key=os.environ["ANTHROPIC_FOUNDRY_API_KEY"],
    resource=os.environ["FOUNDRY_RESOURCE"],
)

# Same messages.create() on every client; only the model ID differs on Bedrock.
for client, model in [
    (bedrock, "anthropic.claude-sonnet-5-5"),
    (vertex, "claude-sonnet-5-5"),
    (foundry, "claude-sonnet-5-5"),  # your Foundry deployment name
]:
    reply = client.messages.create(
        model=model,
        max_tokens=256,
        messages=[{"role": "user", "content": "Hello!"}],
    )
    print(model, reply.stop_reason)