Anthropic API: Messages, Tool Use, Streaming and Caching Reference Guide

# Python SDK
pip install anthropic           # ≥ 0.40

# Node / TS SDK
npm install @anthropic-ai/sdk

# env — picked up automatically
export ANTHROPIC_API_KEY=sk-ant-...

# Bedrock / Vertex variants ship in extras
pip install "anthropic[bedrock]"   # AWS Bedrock
pip install "anthropic[vertex]"    # GCP Vertex AI

from anthropic import Anthropic

client = Anthropic()

tools = [{
    "name": "get_weather",
    "description": "Current weather for a city.",
    "input_schema": {
        "type": "object",
        "properties": {"city": {"type": "string"}},
        "required": ["city"],
    },
}]

resp = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "Weather in Mumbai?"}],
)

# Loop until the model stops requesting tools
while resp.stop_reason == "tool_use":
    tool_use = next(b for b in resp.content if b.type == "tool_use")
    result = {"temp_c": 31, "city": tool_use.input["city"]}

    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=tools,
        messages=[
            {"role": "user", "content": "Weather in Mumbai?"},
            {"role": "assistant", "content": resp.content},
            {"role": "user", "content": [{
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": str(result),
            }]},
        ],
    )

print(resp.content[0].text)

from anthropic import Anthropic

client = Anthropic()

# Context-manager form — auto-closes the SSE connection
with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Haiku about Mumbai monsoons."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    final = stream.get_final_message()
    print(f"\n[done · {final.usage.input_tokens}+{final.usage.output_tokens} tokens]")

from anthropic import Anthropic

client = Anthropic()

# Mark expensive prefix content as cacheable.
# Anthropic auto-matches on subsequent calls and bills cached input at 10%.
LONG_DOC = open("policy_handbook.md").read()  # ~50k tokens

resp = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=512,
    system=[
        {"type": "text", "text": "You answer using only the handbook below."},
        {
            "type": "text",
            "text": LONG_DOC,
            "cache_control": {"type": "ephemeral"},  # 5-min TTL
        },
    ],
    messages=[{"role": "user", "content": "What's the leave policy?"}],
)

# Watch usage to confirm a cache hit
print(resp.usage.cache_creation_input_tokens)   # > 0 on first call
print(resp.usage.cache_read_input_tokens)       # > 0 on subsequent calls

from anthropic import Anthropic

client = Anthropic()

tools = [{
    "name": "lookup_order",
    "description": "Find an order by ID.",
    "input_schema": {
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
    },
}]

def run_tool(name: str, args: dict) -> str:
    if name == "lookup_order":
        return f"Order {args['order_id']}: shipped, ETA Tue."
    return "unknown tool"

messages = [{"role": "user", "content": "Where is order ORD-71?"}]

while True:
    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=tools,
        messages=messages,
    )
    messages.append({"role": "assistant", "content": resp.content})

    if resp.stop_reason != "tool_use":
        print(next(b.text for b in resp.content if b.type == "text"))
        break

    tool_results = [
        {
            "type": "tool_result",
            "tool_use_id": b.id,
            "content": run_tool(b.name, b.input),
        }
        for b in resp.content if b.type == "tool_use"
    ]
    messages.append({"role": "user", "content": tool_results})