How Claude's prompt caching reduces API costs

Claude prompt caching cost savings come from a simple idea: stop paying to reprocess the same system prompts, few-shot examples, and long documents on every request. Anthropic’s caching lets you mark prefix portions of your prompt as reusable, then charges a fraction of the normal input token price when those prefixes match a recent cache entry. This guide walks through enabling caching, structuring your requests for maximum hit rates, and verifying the savings in your usage data.

Step 1: Understand the pricing model

Before you change any code, know what you’re optimizing. As of this writing, cached input tokens cost roughly 10% of standard input tokens, while cache writes carry a small premium over standard input. The exact ratios shift, so check Anthropic’s current pricing page. The key insight: a cache hit saves ~90% on the cached portion, but a cache miss (write) costs slightly more than a normal request. You want high reuse on large, stable prefixes.

Typical candidates for caching:

• System prompts and developer instructions
• Few-shot examples (10–50 examples)
• Long reference documents (style guides, schemas, policy texts)
• Conversation history that stays fixed across turns

Avoid caching:

• User-specific data that changes per request
• Short prompts where the write premium outweighs the hit savings
• Highly variable content that rarely repeats

Step 2: Upgrade to a supported SDK version

Prompt caching requires Anthropic SDK version 0.25.0 or later (Python) or 0.12.0 or later (TypeScript). Older versions silently ignore the cache control fields.

# Python
pip install --upgrade anthropic>=0.25.0

# TypeScript / Node
npm install @anthropic-ai/sdk@latest

Verify the version in your environment:

import anthropic
print(anthropic.__version__)

Step 3: Mark cacheable prefixes with cache_control

The cache_control field goes on individual content blocks within a message. Use {"type": "ephemeral"} for a 5-minute TTL (the only option currently). You can mark multiple blocks across the system prompt and the first user message.

import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "You are a senior code reviewer. Follow the style guide below exactly.",
            "cache_control": {"type": "ephemeral"}
        },
        {
            "type": "text",
            "text": STYLE_GUIDE,  # 50k token document
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": "Review this PR for style violations:\n\n" + pr_diff,
                    # Do NOT cache the PR diff — it changes every request
                }
            ]
        }
    ]
)

Key rules:

• Only text and document block types support cache_control
• Cache control applies to the prefix of the conversation. The cached portion must be identical (byte-for-byte) across requests
• You can mark multiple blocks; they concatenate into a single cache key
• The cache key includes model version, so switching models invalidates the cache

Step 4: Structure requests for cache hits

The cache key is the concatenation of all cached blocks in order, from the system prompt through the user message. A single character change — whitespace, a comma, a newline — breaks the hit. Follow these patterns:

Keep cached content in constants or versioned files. Load the style guide, system prompt, and few-shot examples from source-controlled files, not string literals that drift across deploys.

# constants.py
SYSTEM_PROMPT = """You are a senior code reviewer. Follow the style guide below exactly."""

with open("style_guide.md", "r") as f:
    STYLE_GUIDE = f.read()

FEW_SHOT_EXAMPLES = [
    {"input": "...", "output": "..."},
    # ...
]

Put all cached blocks first. The cache covers a contiguous prefix. If you interleave cached and non-cached blocks, only the leading cached segment counts.

# Good: all cached blocks first
messages = [
    {"role": "user", "content": [
        {"type": "text", "text": SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"}},
        {"type": "text", "text": STYLE_GUIDE, "cache_control": {"type": "ephemeral"}},
        {"type": "text", "text": "Now review this code:"},  # not cached
        {"type": "text", "text": user_code},
    ]}
]

# Bad: non-cached block breaks the prefix
messages = [
    {"role": "user", "content": [
        {"type": "text", "text": SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"}},
        {"type": "text", "text": "Current date: 2024-01-15"},  # changes daily — kills cache
        {"type": "text", "text": STYLE_GUIDE, "cache_control": {"type": "ephemeral"}},
    ]}
]

Pin the model version. Cache keys include the model string. "claude-3-5-sonnet-20241022" and "claude-3-5-sonnet-latest" are different keys. Use explicit versions in production.

Step 5: Read the cache metrics from the response

Every response includes usage.cache_creation_input_tokens and usage.cache_read_input_tokens. Log these. They are your ground truth.

response = client.messages.create(...)

usage = response.usage
print(f"Cache write (creation): {usage.cache_creation_input_tokens} tokens")
print(f"Cache read (hits):      {usage.cache_read_input_tokens} tokens")
print(f"Regular input tokens:   {usage.input_tokens} tokens")
print(f"Output tokens:          {usage.output_tokens} tokens")

Interpretation:

• cache_creation_input_tokens > 0 on the first request (or after TTL expiry) — you paid the write premium
• cache_read_input_tokens > 0 on subsequent requests — you got the 90% discount
• Both zero means caching isn’t working (check SDK version, block types, model pinning)

Step 6: Build a cache-aware request wrapper

Wrap the logic so your application code stays clean and you get consistent logging.

# claude_cached.py
import os
import anthropic
from dataclasses import dataclass
from typing import Optional

@dataclass
class CacheMetrics:
    cache_write_tokens: int
    cache_read_tokens: int
    regular_input_tokens: int
    output_tokens: int

    @property
    def hit_rate(self) -> float:
        total_cached = self.cache_write_tokens + self.cache_read_tokens
        if total_cached == 0:
            return 0.0
        return self.cache_read_tokens / total_cached

    def estimated_savings_usd(self, input_price_per_mtok: float, cache_read_discount: float = 0.9) -> float:
        """Rough estimate: cached read tokens saved ~90% vs regular input."""
        return (self.cache_read_tokens / 1_000_000) * input_price_per_mtok * cache_read_discount


class CachedClaudeClient:
    def __init__(self, api_key: Optional[str] = None, model: str = "claude-3-5-sonnet-20241022"):
        self.client = anthropic.Anthropic(api_key=api_key or os.environ["ANTHROPIC_API_KEY"])
        self.model = model

    def create_cached(
        self,
        *,
        system_cached: list[str],
        user_cached: list[str],
        user_uncached: list[str],
        max_tokens: int = 4096,
        temperature: float = 0.0,
    ) -> tuple[str, CacheMetrics]:
        """
        system_cached: list of strings for system prompt blocks (all cached)
        user_cached: list of strings for user message prefix blocks (all cached)
        user_uncached: list of strings for user message suffix blocks (not cached)
        """
        system_blocks = [
            {"type": "text", "text": text, "cache_control": {"type": "ephemeral"}}
            for text in system_cached
        ]

        user_blocks = [
            {"type": "text", "text": text, "cache_control": {"type": "ephemeral"}}
            for text in user_cached
        ] + [
            {"type": "text", "text": text}
            for text in user_uncached
        ]

        response = self.client.messages.create(
            model=self.model,
            max_tokens=max_tokens,
            temperature=temperature,
            system=system_blocks,
            messages=[{"role": "user", "content": user_blocks}],
        )

        usage = response.usage
        metrics = CacheMetrics(
            cache_write_tokens=usage.cache_creation_input_tokens,
            cache_read_tokens=usage.cache_read_input_tokens,
            regular_input_tokens=usage.input_tokens,
            output_tokens=usage.output_tokens,
        )

        return response.content[0].text, metrics

Usage:

from claude_cached import CachedClaudeClient

client = CachedClaudeClient()

# First request — cache write
answer, metrics = client.create_cached(
    system_cached=[SYSTEM_PROMPT, STYLE_GUIDE],
    user_cached=[FEW_SHOT_PROMPT],
    user_uncached=[f"Review this PR:\n{pr_diff}"],
)
print(f"First request hit rate: {metrics.hit_rate:.1%}")
print(f"Estimated savings: ${metrics.estimated_savings_usd(3.00):.4f}")

# Second request (same cached prefix) — cache read
answer2, metrics2 = client.create_cached(
    system_cached=[SYSTEM_PROMPT, STYLE_GUIDE],
    user_cached=[FEW_SHOT_PROMPT],
    user_uncached=[f"Review this PR:\n{pr_diff_2}"],
)
print(f"Second request hit rate: {metrics2.hit_rate:.1%}")

Step 7: Handle cache misses and TTL expiry

The ephemeral cache lasts 5 minutes of inactivity. After that, the next request pays the write premium again. Design for this:

• Batch related requests within the 5-minute window when possible
• Accept the write cost on cold starts — it’s still cheaper than sending the full prompt uncached every time
• Monitor the write/read ratio. A healthy workload shows one write followed by many reads.

def log_cache_health(metrics: CacheMetrics, request_id: str):
    if metrics.cache_write_tokens > 0 and metrics.cache_read_tokens == 0:
        print(f"[{request_id}] COLD START: wrote {metrics.cache_write_tokens} tokens to cache")
    elif metrics.cache_read_tokens > 0:
        print(f"[{request_id}] CACHE HIT: read {metrics.cache_read_tokens} tokens, "
              f"hit rate {metrics.hit_rate:.1%}")
    else:
        print(f"[{request_id}] NO CACHE: check SDK version and cache_control placement")

Step 8: Verify savings in your billing data

Response metrics are per-request. To prove the impact on your bill, correlate with Anthropic’s usage export or your own aggregated logs.

Quick sanity check: Run a load test — 100 requests with identical cached prefixes, then 100 with varied prefixes. Compare total input tokens billed.

# load_test.py
import time
from claude_cached import CachedClaudeClient

client = CachedClaudeClient()

def run_batch(label: str, count: int, vary_uncached: bool):
    total_write = 0
    total_read = 0
    total_regular = 0

    for i in range(count):
        uncached = [f"Request {i}"] if vary_uncached else ["Fixed request"]
        _, metrics = client.create_cached(
            system_cached=[SYSTEM_PROMPT, STYLE_GUIDE],
            user_cached=[FEW_SHOT_PROMPT],
            user_uncached=uncached,
        )
        total_write += metrics.cache_write_tokens
        total_read += metrics.cache_read_tokens
        total_regular += metrics.regular_input_tokens
        time.sleep(0.1)  # be nice to rate limits

    print(f"{label}: write={total_write}, read={total_read}, regular={total_regular}")

run_batch("Cached prefix (100 requests)", 100, vary_uncached=False)
run_batch("Varied prefix (100 requests)", 100, vary_uncached=True)

Expected output pattern:

Cached prefix (100 requests): write=52000, read=5148000, regular=10000
Varied prefix (100 requests): write=5200000, read=0, regular=10000

The first batch writes once, reads 99 times. The second batch writes every request because the cache key changes. That difference is your savings.

Step 9: Avoid common pitfalls

Step 10: Integrate with your observability stack

Ship the metrics to your time-series database (Datadog, Prometheus, CloudWatch) alongside latency and error rates. Alert on:

• Cache hit rate dropping below 80% for a sustained period (indicates prompt drift or TTL issues)
• Sudden spike in cache_creation_input_tokens (new deployment changed a cached block)
• Ratio of cache write tokens to total input tokens exceeding 20% (too many cold starts)

Example Datadog metric submission:

from ddtrace import tracer
import os

def emit_cache_metrics(metrics: CacheMetrics, tags: dict):
    tags = {**tags, "model": os.getenv("CLAUDE_MODEL", "unknown")}
    tracer.dogstatsd.gauge("claude.cache.write_tokens", metrics.cache_write_tokens, tags=tags)
    tracer.dogstatsd.gauge("claude.cache.read_tokens", metrics.cache_read_tokens, tags=tags)
    tracer.dogstatsd.gauge("claude.cache.hit_rate", metrics.hit_rate, tags=tags)
    tracer.dogstatsd.gauge("claude.cache.estimated_savings_usd", 
                           metrics.estimated_savings_usd(3.00), tags=tags)

Call this after every request in your wrapper.

Verification checklist

Before declaring success, confirm each item:

• SDK version ≥0.25.0 (Python) or ≥0.12.0 (TypeScript)
• Model pinned to exact version string (e.g., claude-3-5-sonnet-20241022)
• All cache_control blocks are text or document type
• Cached blocks form a contiguous prefix in system + first user message
• First request shows cache_creation_input_tokens > 0
• Second request (within 5 min, same prefix) shows cache_read_input_tokens > 0
• Hit rate in load test exceeds 90% for repeated prefixes
• Billing export shows reduced input token charges vs. pre-caching baseline

Where n4n.ai fits

If you route traffic through n4n.ai’s OpenAI-compatible endpoint, the same cache_control fields pass through to Anthropic unchanged. You still read cache_creation_input_tokens and cache_read_input_tokens from the response usage object. The gateway adds per-token metering and automatic fallback across providers, but the caching mechanics remain identical to calling Anthropic directly.