Cut Anthropic API Costs 90% with Prompt Caching 2026

# Python — use uv for fast, reproducible installs
uv add anthropic

# Or pip
pip install anthropic --break-system-packages

# Node.js / TypeScript
npm install @anthropic-ai/sdk

python -c "import anthropic; print(anthropic.__version__)"
# Expected: 0.40.0 or higher

import anthropic

client = anthropic.Anthropic()

# Large system prompt — must be 1,024+ tokens to cache
SYSTEM_PROMPT = """
You are a senior software engineer specializing in distributed systems.
You help teams design, debug, and optimize production infrastructure.

[... your full 2,000+ token system prompt here ...]
"""

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"},  # Mark this prefix for caching
        }
    ],
    messages=[
        {"role": "user", "content": "How do I reduce p99 latency in a gRPC service?"}
    ],
)

print(response.content[0].text)

# Check cache metrics
usage = response.usage
print(f"Cache write tokens:  {usage.cache_creation_input_tokens}")
print(f"Cache read tokens:   {usage.cache_read_input_tokens}")
print(f"Standard input:      {usage.input_tokens}")

tools = [
    {
        "name": "run_query",
        "description": "Execute a SQL query against the production database and return results as JSON.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Valid SQL SELECT statement"},
                "database": {"type": "string", "enum": ["prod", "replica", "analytics"]},
                "timeout_ms": {"type": "integer", "description": "Query timeout in milliseconds. Max 30000."},
            },
            "required": ["query", "database"],
        },
    },
    # ... more tools
]

# Add cache_control to the LAST tool in the list
# The cache prefix covers everything up to and including this breakpoint
tools[-1]["cache_control"] = {"type": "ephemeral"}

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "Show me the top 10 users by revenue this month."}],
)

import anthropic

client = anthropic.Anthropic()

# Load your document once — e.g., a 50-page technical spec
with open("technical_spec.txt", "r") as f:
    document_content = f.read()

def ask_about_document(question: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        system=[
            {
                "type": "text",
                "text": "You are a technical documentation assistant. Answer questions accurately based only on the provided document.",
                "cache_control": {"type": "ephemeral"},  # Cache the instruction
            },
        ],
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": f"<document>\n{document_content}\n</document>",
                        "cache_control": {"type": "ephemeral"},  # Cache the document
                    },
                    {
                        "type": "text",
                        "text": question,  # Only this varies per call — not cached
                    },
                ],
            }
        ],
    )
    return response.content[0].text

# First call — writes cache for both system prompt and document
answer1 = ask_about_document("What are the rate limits for the /v1/messages endpoint?")

# Second call — both prefixes hit cache; only the question token is billed at standard rate
answer2 = ask_about_document("What authentication methods are supported?")

import anthropic

client = anthropic.Anthropic()

def chat_with_caching(conversation_history: list, new_user_message: str) -> tuple[str, list]:
    """
    Send a multi-turn chat message with the history prefix cached.
    Returns (assistant_reply, updated_history).
    """
    # Mark the last assistant message in history as the cache breakpoint
    # Everything before this is a stable prefix
    cached_history = []
    for i, msg in enumerate(conversation_history):
        if i == len(conversation_history) - 1 and msg["role"] == "assistant":
            # Cache breakpoint at the end of existing history
            cached_history.append({
                "role": msg["role"],
                "content": [
                    {
                        "type": "text",
                        "text": msg["content"],
                        "cache_control": {"type": "ephemeral"},
                    }
                ],
            })
        else:
            cached_history.append(msg)

    messages = cached_history + [{"role": "user", "content": new_user_message}]

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system="You are a helpful assistant.",
        messages=messages,
    )

    assistant_reply = response.content[0].text

    # Append the new turn to history for the next call
    updated_history = conversation_history + [
        {"role": "user", "content": new_user_message},
        {"role": "assistant", "content": assistant_reply},
    ]

    return assistant_reply, updated_history


# Usage
history = []
reply, history = chat_with_caching(history, "Explain CAP theorem in simple terms.")
reply, history = chat_with_caching(history, "How does Cassandra trade off consistency?")
reply, history = chat_with_caching(history, "What about CockroachDB?")
# By the third call, the growing history is served from cache at 10% of input token cost

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

const SYSTEM_PROMPT = `
You are a senior backend engineer specializing in API design and performance.
[... your 1,024+ token system prompt ...]
`;

async function callWithCaching(userMessage: string): Promise<string> {
  const response = await client.messages.create({
    model: "claude-sonnet-4-20250514",
    max_tokens: 1024,
    system: [
      {
        type: "text",
        text: SYSTEM_PROMPT,
        cache_control: { type: "ephemeral" },
      },
    ],
    messages: [{ role: "user", content: userMessage }],
  });

  // Log cache metrics for monitoring
  const { usage } = response;
  console.log({
    cacheWrite: usage.cache_creation_input_tokens,  // Tokens written to cache
    cacheRead: usage.cache_read_input_tokens,        // Tokens loaded from cache
    standardInput: usage.input_tokens,               // Non-cached input tokens
  });

  const block = response.content[0];
  if (block.type !== "text") throw new Error("Unexpected response type");
  return block.text;
}

// First call — cache miss, writes the system prompt prefix
await callWithCaching("What HTTP status code should I return for rate limiting?");

// Second call — cache hit, system prompt loads from cache
await callWithCaching("How do I implement exponential backoff correctly?");

import anthropic

client = anthropic.Anthropic()

SYSTEM = "You are a helpful assistant. " + ("word " * 600)  # ~1,024+ tokens

def test_call(label: str):
    resp = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=50,
        system=[{"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}}],
        messages=[{"role": "user", "content": "Say hello."}],
    )
    u = resp.usage
    print(f"{label}:")
    print(f"  cache_write={u.cache_creation_input_tokens} cache_read={u.cache_read_input_tokens} input={u.input_tokens}")

test_call("Call 1 — should write cache")
test_call("Call 2 — should read cache")
test_call("Call 3 — should read cache")