How to Use Prompt Caching in Claude API: Complete 2026 Guide with Code Examples

{
  "type": "text",
  "text": "Your long context to cache...",
  "cache_control": {"type": "ephemeral"}
}

# BAD: Regenerating tools list on each request invalidates all caches
def get_response(user_message: str):
    tools = generate_dynamic_tools()  # This kills your cache!
    return client.messages.create(
        tools=tools,
        system=system_prompt,
        messages=[{"role": "user", "content": user_message}]
    )

# GOOD: Use stable tool definitions
STATIC_TOOLS = [...]  # Define once, never change

def get_response(user_message: str):
    return client.messages.create(
        tools=STATIC_TOOLS,  # Same every time = cache preserved
        system=system_prompt,
        messages=[{"role": "user", "content": user_message}]
    )

Request 1 (t=0min): Cache created, TTL set to 5 minutes
Request 2 (t=2min): Cache hit, TTL reset to 5 minutes from now
Request 3 (t=6min): Cache hit, TTL reset to 5 minutes from now
Request 4 (t=10min): Cache hit, TTL reset to 5 minutes from now
...

"cache_control": {
  "type": "ephemeral",
  "ttl": "1h"
}

# With a single breakpoint, the system automatically finds matches
system=[
    {"type": "text", "text": "Instructions..."},  # Could match here
    {"type": "text", "text": "Context part 1..."},  # Or here
    {"type": "text", "text": "Context part 2...", "cache_control": {"type": "ephemeral"}}  # Explicit breakpoint
]

Cost per request = 50,500 tokens × $2/MTok = $0.101
Daily cost = 1,000 × $0.101 = $101/day
Monthly cost = 30 × $101 = $3,030/month

First request (cache write):
  50,000 × $2.50/MTok + 500 × $2/MTok = $0.125 + $0.001 = $0.126

Subsequent 999 requests (cache read):
  50,000 × $0.10/MTok + 500 × $2/MTok = $0.005 + $0.001 = $0.006 each
  Total: 999 × $0.006 = $5.99

Daily total = $0.126 + $5.99 = $6.12/day
Monthly total = 30 × $6.12 = $184/month

pip install anthropic

import anthropic
from typing import Optional, Dict, Any
import logging

# Configure logging for cache monitoring
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def create_cached_message(
    client: anthropic.Anthropic,
    system_content: str,
    user_message: str,
    model: str = "claude-sonnet-5-5",
    max_tokens: int = 1024
) -> Dict[str, Any]:
    """
    Send a message with prompt caching enabled.

    Args:
        client: Anthropic client instance
        system_content: Long system prompt to cache (should be > min tokens)
        user_message: User's current message
        model: Model to use
        max_tokens: Maximum tokens in response

    Returns:
        Dict with content and detailed usage statistics
    """
    try:
        response = client.messages.create(
            model=model,
            max_tokens=max_tokens,
            system=[
                {
                    "type": "text",
                    "text": "You are a helpful assistant specialized in answering questions.",
                },
                {
                    "type": "text",
                    "text": system_content,
                    "cache_control": {"type": "ephemeral"}
                }
            ],
            messages=[
                {"role": "user", "content": user_message}
            ]
        )

        # Extract usage metrics
        usage = response.usage
        cache_read = getattr(usage, 'cache_read_input_tokens', 0) or 0
        cache_write = getattr(usage, 'cache_creation_input_tokens', 0) or 0

        # Log cache performance
        if cache_read > 0:
            logger.info(f"Cache HIT: {cache_read} tokens read from cache")
        elif cache_write > 0:
            logger.info(f"Cache MISS: {cache_write} tokens written to cache")
        else:
            logger.warning("No caching occurred - check token count")

        return {
            "content": response.content[0].text if response.content else "",
            "usage": {
                "cache_read": cache_read,
                "cache_write": cache_write,
                "input": usage.input_tokens,
                "output": usage.output_tokens
            },
            "model": response.model,
            "stop_reason": response.stop_reason
        }

    except anthropic.APIError as e:
        logger.error(f"API error: {e}")
        raise

# Production usage example
if __name__ == "__main__":
    client = anthropic.Anthropic()  # Uses ANTHROPIC_API_KEY env var

    # Simulated large document (in production, load from file/database)
    product_documentation = """
    # Product Manual - Version 3.2

    ## Chapter 1: Getting Started
    This comprehensive guide covers all aspects of the product...
    [Imagine 50,000 tokens of documentation here]

    ## Chapter 2: Installation
    Follow these steps to install the product correctly...

    ## Chapter 3: Configuration
    Configuration options and best practices...

    ## Chapter 4: Troubleshooting
    Common issues and their solutions...
    """ * 100  # Repeat to simulate large document

    # First request - creates cache
    print("Making first request (cache write expected)...")
    result1 = create_cached_message(
        client,
        product_documentation,
        "What are the main features of this product?"
    )
    print(f"Usage: {result1['usage']}")

    # Second request - should hit cache
    print("\nMaking second request (cache hit expected)...")
    result2 = create_cached_message(
        client,
        product_documentation,
        "How do I install the product on Windows?"
    )
    print(f"Usage: {result2['usage']}")

    # Third request - should still hit cache
    print("\nMaking third request (cache hit expected)...")
    result3 = create_cached_message(
        client,
        product_documentation,
        "What are the system requirements?"
    )
    print(f"Usage: {result3['usage']}")

npm install @anthropic-ai/sdk

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

// Type definitions for cache usage
interface CacheUsage {
  cacheRead: number;
  cacheWrite: number;
  input: number;
  output: number;
}

interface CachedResponse {
  content: string;
  usage: CacheUsage;
  model: string;
  stopReason: string | null;
}

// Configuration type
interface CacheConfig {
  model?: string;
  maxTokens?: number;
  ttl?: '5m' | '1h';
}

async function createCachedMessage(
  client: Anthropic,
  systemContent: string,
  userMessage: string,
  config: CacheConfig = {}
): Promise<CachedResponse> {
  const {
    model = 'claude-sonnet-5-5',
    maxTokens = 1024,
    ttl = '5m'
  } = config;

  const cacheControl = ttl === '1h'
    ? { type: 'ephemeral' as const, ttl: '1h' as const }
    : { type: 'ephemeral' as const };

  try {
    const response = await client.messages.create({
      model,
      max_tokens: maxTokens,
      system: [
        {
          type: 'text',
          text: 'You are a helpful assistant.',
        },
        {
          type: 'text',
          text: systemContent,
          cache_control: cacheControl
        }
      ],
      messages: [
        { role: 'user', content: userMessage }
      ]
    });

    const usage = response.usage;
    const cacheRead = (usage as any).cache_read_input_tokens || 0;
    const cacheWrite = (usage as any).cache_creation_input_tokens || 0;

    // Log cache status
    if (cacheRead > 0) {
      console.log(`[CACHE HIT] ${cacheRead} tokens read from cache`);
    } else if (cacheWrite > 0) {
      console.log(`[CACHE MISS] ${cacheWrite} tokens written to cache`);
    } else {
      console.warn('[WARNING] No caching occurred');
    }

    return {
      content: response.content[0].type === 'text'
        ? response.content[0].text
        : '',
      usage: {
        cacheRead,
        cacheWrite,
        input: usage.input_tokens,
        output: usage.output_tokens
      },
      model: response.model,
      stopReason: response.stop_reason
    };
  } catch (error) {
    if (error instanceof Anthropic.APIError) {
      console.error(`API Error: ${error.message}`);
    }
    throw error;
  }
}

// Cache monitoring class for production use
class CachePerformanceMonitor {
  private metrics: {
    requests: number;
    cacheHits: number;
    cacheMisses: number;
    totalCacheRead: number;
    totalCacheWrite: number;
    totalRegularTokens: number;
  };

  constructor() {
    this.metrics = {
      requests: 0,
      cacheHits: 0,
      cacheMisses: 0,
      totalCacheRead: 0,
      totalCacheWrite: 0,
      totalRegularTokens: 0
    };
  }

  recordRequest(usage: CacheUsage): void {
    this.metrics.requests++;
    this.metrics.totalCacheRead += usage.cacheRead;
    this.metrics.totalCacheWrite += usage.cacheWrite;
    this.metrics.totalRegularTokens += usage.input;

    if (usage.cacheRead > 0) {
      this.metrics.cacheHits++;
    } else if (usage.cacheWrite > 0) {
      this.metrics.cacheMisses++;
    }
  }

  getReport(): object {
    const hitRate = this.metrics.requests > 0
      ? (this.metrics.cacheHits / this.metrics.requests * 100).toFixed(1)
      : 0;

    // Calculate cost savings (using Sonnet 5.5 pricing as example)
    const baseRate = 2; // $2 per MTok
    const writeRate = 2.5;
    const readRate = 0.10;

    const withoutCaching =
      (this.metrics.totalCacheRead + this.metrics.totalCacheWrite) * baseRate / 1_000_000;
    const withCaching =
      (this.metrics.totalCacheWrite * writeRate + this.metrics.totalCacheRead * readRate) / 1_000_000;
    const savings = withoutCaching - withCaching;

    return {
      totalRequests: this.metrics.requests,
      cacheHits: this.metrics.cacheHits,
      cacheMisses: this.metrics.cacheMisses,
      hitRate: `${hitRate}%`,
      tokensReadFromCache: this.metrics.totalCacheRead,
      tokensWrittenToCache: this.metrics.totalCacheWrite,
      estimatedSavings: `$${savings.toFixed(4)}`
    };
  }

  reset(): void {
    this.metrics = {
      requests: 0,
      cacheHits: 0,
      cacheMisses: 0,
      totalCacheRead: 0,
      totalCacheWrite: 0,
      totalRegularTokens: 0
    };
  }
}

// Example usage
async function main() {
  const client = new Anthropic();
  const monitor = new CachePerformanceMonitor();

  const largeDocument = `
    [Your large document content here - needs to be at least 1024 tokens]
    This content will be cached after the first request.
    Include all your static context, documentation, or examples here.
  `.repeat(50);

  console.log('Testing prompt caching...\n');

  // Make several requests
  for (let i = 1; i <= 5; i++) {
    console.log(`Request ${i}:`);
    const result = await createCachedMessage(
      client,
      largeDocument,
      `Question ${i}: What can you help me with?`
    );
    monitor.recordRequest(result.usage);
    console.log(`Response length: ${result.content.length} chars\n`);
  }

  console.log('Performance Report:');
  console.log(monitor.getReport());
}

main().catch(console.error);

import time
from dataclasses import dataclass, field
from typing import List, Dict
from datetime import datetime

@dataclass
class CacheMetrics:
    """Tracks cache performance over time."""
    timestamp: datetime = field(default_factory=datetime.now)
    cache_read: int = 0
    cache_write: int = 0
    regular_tokens: int = 0
    latency_ms: float = 0

class CacheMonitor:
    """Production-grade cache monitoring."""

    def __init__(self, model: str = "claude-sonnet-5-5"):
        self.model = model
        self.history: List[CacheMetrics] = []
        self._pricing = self._get_pricing(model)

    def _get_pricing(self, model: str) -> Dict[str, float]:
        """Get pricing for the model (per million tokens)."""
        pricing_map = {
            "claude-sonnet-5-5": {"base": 2.0, "write": 2.5, "read": 0.10},
            "claude-opus-5-5": {"base": 4.0, "write": 5.0, "read": 0.20},
            "claude-haiku-5-5": {"base": 0.10, "write": 0.125, "read": 0.01},  # prompts up to 100K tokens
        }
        return pricing_map.get(model, pricing_map["claude-sonnet-5-5"])

    def record(self, usage: dict, latency_ms: float):
        """Record a request's cache metrics."""
        self.history.append(CacheMetrics(
            timestamp=datetime.now(),
            cache_read=usage.get("cache_read", 0),
            cache_write=usage.get("cache_write", 0),
            regular_tokens=usage.get("input", 0),
            latency_ms=latency_ms
        ))

    def get_summary(self) -> Dict:
        """Generate comprehensive performance summary."""
        if not self.history:
            return {"error": "No data recorded"}

        total_requests = len(self.history)
        cache_hits = sum(1 for m in self.history if m.cache_read > 0)
        cache_misses = sum(1 for m in self.history if m.cache_write > 0)

        total_read = sum(m.cache_read for m in self.history)
        total_write = sum(m.cache_write for m in self.history)

        # Calculate costs
        cost_without_cache = (total_read + total_write) * self._pricing["base"] / 1_000_000
        cost_with_cache = (
            total_write * self._pricing["write"] +
            total_read * self._pricing["read"]
        ) / 1_000_000

        avg_latency = sum(m.latency_ms for m in self.history) / total_requests

        return {
            "total_requests": total_requests,
            "cache_hits": cache_hits,
            "cache_misses": cache_misses,
            "hit_rate": f"{cache_hits/total_requests*100:.1f}%",
            "total_tokens_cached": total_read + total_write,
            "cost_without_caching": f"${cost_without_cache:.4f}",
            "cost_with_caching": f"${cost_with_cache:.4f}",
            "savings": f"${cost_without_cache - cost_with_cache:.4f}",
            "avg_latency_ms": f"{avg_latency:.0f}ms"
        }

from typing import List, Dict

def create_conversation_message(
    client: anthropic.Anthropic,
    system_prompt: str,
    conversation_history: List[Dict],
    new_message: str
) -> Dict:
    """
    Handle multi-turn conversations with optimal caching.

    Strategy:
    1. Cache the system prompt (rarely changes)
    2. Cache up to the second-to-last turn (conversation history)
    3. Let the final turn be processed fresh (new user message)
    """

    # Build messages with strategic cache control
    messages = []

    for i, msg in enumerate(conversation_history):
        content = msg["content"]

        # Add cache control to the second-to-last message
        # This allows the conversation prefix to be reused
        if i == len(conversation_history) - 2 and isinstance(content, str):
            content = [{
                "type": "text",
                "text": content,
                "cache_control": {"type": "ephemeral"}
            }]

        messages.append({
            "role": msg["role"],
            "content": content
        })

    # Add the new user message with cache control for future turns
    messages.append({
        "role": "user",
        "content": [{
            "type": "text",
            "text": new_message,
            "cache_control": {"type": "ephemeral"}
        }]
    })

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1024,
        system=[{
            "type": "text",
            "text": system_prompt,
            "cache_control": {"type": "ephemeral"}  # System always cached
        }],
        messages=messages
    )

    return {
        "response": response.content[0].text,
        "usage": {
            "cache_read": getattr(response.usage, 'cache_read_input_tokens', 0),
            "cache_write": getattr(response.usage, 'cache_creation_input_tokens', 0),
        }
    }

# Usage example
conversation = [
    {"role": "user", "content": "Hi, I need help with my order"},
    {"role": "assistant", "content": "Hello! I'd be happy to help..."},
    {"role": "user", "content": "It hasn't arrived yet"},
    {"role": "assistant", "content": "I understand that's frustrating..."},
]

result = create_conversation_message(
    client,
    "You are a helpful customer support agent...",
    conversation,
    "Can you check the tracking?"
)

# Define your tools once and reuse
TOOLS = [
    {
        "name": "search_products",
        "description": "Search the product catalog by keyword, category, or price range",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Search keywords"
                },
                "category": {
                    "type": "string",
                    "enum": ["electronics", "clothing", "home", "sports"]
                },
                "max_price": {
                    "type": "number",
                    "description": "Maximum price filter"
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "get_order_status",
        "description": "Look up the status of a customer order",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "The order ID to look up"
                }
            },
            "required": ["order_id"]
        }
    },
    {
        "name": "process_return",
        "description": "Initiate a return for an order",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string"},
                "reason": {"type": "string"},
                "items": {
                    "type": "array",
                    "items": {"type": "string"}
                }
            },
            "required": ["order_id", "reason"]
        },
        # Add cache_control to the LAST tool only
        "cache_control": {"type": "ephemeral"}
    }
]

def tool_use_request(user_message: str):
    """Make a tool-use request with cached tools."""
    return client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1024,
        tools=TOOLS,  # Tools are now cached
        messages=[{"role": "user", "content": user_message}]
    )

def rag_query_with_tiered_cache(
    client: anthropic.Anthropic,
    base_instructions: str,
    retrieved_documents: str,
    user_query: str
) -> str:
    """
    RAG query with two-level caching:

    Level 1: Base instructions (rarely change, shared across all users)
    Level 2: Retrieved documents (change per session/topic)

    This allows:
    - Instructions to be cached long-term
    - Documents to be cached per-session
    - User queries to vary freely
    """

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=2048,
        system=[
            {
                "type": "text",
                "text": base_instructions,
                "cache_control": {"type": "ephemeral"}  # Cache breakpoint 1
            },
            {
                "type": "text",
                "text": f"""
                ## Retrieved Documents

                The following documents are relevant to the user's question:

                {retrieved_documents}

                Use these documents to provide accurate, well-sourced answers.
                """,
                "cache_control": {"type": "ephemeral"}  # Cache breakpoint 2
            }
        ],
        messages=[
            {"role": "user", "content": user_query}
        ]
    )

    return response.content[0].text

# Example with stable instructions and variable documents
BASE_INSTRUCTIONS = """
You are an expert research assistant. Your role is to:
1. Answer questions using ONLY the provided documents
2. Cite specific documents when making claims
3. Acknowledge when information is not available
4. Never make up information

Format your responses clearly with headers and bullet points.
"""

# Different sessions can share the base instructions cache
# but have their own document caches
session_1_docs = "Document 1: ...\nDocument 2: ..."
session_2_docs = "Document A: ...\nDocument B: ..."

answer_1 = rag_query_with_tiered_cache(client, BASE_INSTRUCTIONS, session_1_docs, "Question about doc 1")
answer_2 = rag_query_with_tiered_cache(client, BASE_INSTRUCTIONS, session_2_docs, "Question about doc A")

N × 100% = 125% + (N-1) × 10%
100N = 125 + 10N - 10
90N = 115
N = 1.28

import anthropic

client = anthropic.Anthropic()

def check_token_count(content: str, model: str = "claude-sonnet-5-5") -> dict:
    """Check if content meets minimum cache threshold."""
    # Note: This is a simplified check. Use the actual tokenizer for precision.
    estimated_tokens = len(content) / 4  # Rough estimate

    thresholds = {
        "claude-sonnet-5-5": 512,
        "claude-opus-5-5": 512,
        "claude-haiku-5-5": 512,
        "claude-haiku-4-5": 4096,
    }

    min_tokens = thresholds.get(model, 1024)

    return {
        "estimated_tokens": int(estimated_tokens),
        "min_required": min_tokens,
        "can_cache": estimated_tokens >= min_tokens
    }

# Usage
result = check_token_count(your_system_prompt)
print(f"Token check: {result}")

def analyze_cache_response(response) -> dict:
    """Analyze cache behavior from response."""
    usage = response.usage
    cache_read = getattr(usage, 'cache_read_input_tokens', 0) or 0
    cache_write = getattr(usage, 'cache_creation_input_tokens', 0) or 0

    status = "UNKNOWN"
    if cache_read > 0:
        status = "CACHE_HIT"
    elif cache_write > 0:
        status = "CACHE_MISS_NEW_ENTRY"
    else:
        status = "NO_CACHING_OCCURRED"

    return {
        "status": status,
        "cache_read_tokens": cache_read,
        "cache_write_tokens": cache_write,
        "regular_input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens
    }

import hashlib
import json

def content_fingerprint(content: str) -> str:
    """Generate a fingerprint to verify content identity."""
    # Normalize whitespace and encoding
    normalized = ' '.join(content.split())
    return hashlib.sha256(normalized.encode('utf-8')).hexdigest()[:16]

# Log fingerprint with each request
def cached_request_with_fingerprint(client, system_prompt, user_message):
    fingerprint = content_fingerprint(system_prompt)
    print(f"System prompt fingerprint: {fingerprint}")

    response = client.messages.create(...)
    cache_info = analyze_cache_response(response)
    print(f"Cache status: {cache_info['status']}")

    return response

from datetime import datetime
import threading

class RequestTimer:
    """Track time between requests to detect TTL issues."""

    def __init__(self):
        self._last_request = None
        self._lock = threading.Lock()

    def check_and_update(self) -> dict:
        with self._lock:
            now = datetime.now()
            result = {
                "current_time": now.isoformat(),
                "warning": None
            }

            if self._last_request:
                delta = (now - self._last_request).total_seconds()
                result["seconds_since_last"] = delta

                if delta > 300:
                    result["warning"] = "TTL likely expired (>5 min since last request)"
                elif delta > 270:
                    result["warning"] = "Approaching TTL limit (>4.5 min)"

            self._last_request = now
            return result

timer = RequestTimer()

# Use before each request
timing_info = timer.check_and_update()
if timing_info.get("warning"):
    print(f"WARNING: {timing_info['warning']}")