Replay-Driven Testing for Claude API: A Production Pattern for Recording and Replaying Responses

Pattern 1: Messages API cassettes in Python

The simplest case: a non-streaming Messages call. The implementation below patches Anthropic.messages.create.

# tests/cassette.py
"""Record and replay Claude Messages API responses.
Calls the real API once when no recording exists; replays from disk afterward.
"""
 
import json
import hashlib
from pathlib import Path
from typing import Any
from contextlib import contextmanager
from unittest.mock import patch
from anthropic import Anthropic
from anthropic.types import Message
 
CASSETTE_DIR = Path(__file__).parent / "cassettes"
 
def _request_key(messages: list[dict], model: str, tools: list | None = None) -> str:
    """Deterministic key from request shape — distinguishes recordings inside a cassette."""
    payload = json.dumps(
        {"messages": messages, "model": model, "tools": tools or []},
        sort_keys=True,
        ensure_ascii=False,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]
 
@contextmanager
def cassette(name: str, record_mode: str = "once"):
    """Use as:
        with cassette("test_basic_response"):
            client = Anthropic()
            res = client.messages.create(...)
 
    record_mode:
        "once":         Record if missing, otherwise replay only.
        "new_episodes": Append new recordings to an existing cassette.
        "none":         Replay only — fail loudly if no recording exists.
    """
    CASSETTE_DIR.mkdir(parents=True, exist_ok=True)
    cassette_path = CASSETTE_DIR / f"{name}.json"
 
    recordings: dict[str, Any] = (
        json.loads(cassette_path.read_text("utf-8"))
        if cassette_path.exists() else {}
    )
    new_recordings: dict[str, Any] = {}
    real_create = Anthropic().messages.create
 
    def patched_create(self, **kwargs):
        messages = kwargs.get("messages", [])
        model = kwargs.get("model", "")
        tools = kwargs.get("tools")
        key = _request_key(messages, model, tools)
 
        if key in recordings:
            return Message.model_validate(recordings[key])
 
        if record_mode == "none":
            raise AssertionError(
                f"Cassette {name} has no recording for key={key}. "
                f"Re-record with record_mode='once'."
            )
 
        response = real_create(**kwargs)
        new_recordings[key] = response.model_dump(mode="json")
        return response
 
    with patch.object(Anthropic.messages.__class__, "create", patched_create):
        try:
            yield
        finally:
            if new_recordings and record_mode != "none":
                recordings.update(new_recordings)
                cassette_path.write_text(
                    json.dumps(recordings, ensure_ascii=False, indent=2),
                    encoding="utf-8",
                )

A test using the cassette hits the real API on the first run and replays from disk afterward.

# tests/test_summarizer.py
"""Verify regression behavior against a recorded Claude response."""
 
from src.summarizer import summarize_long_text
from tests.cassette import cassette
 
def test_summarizer_truncates_long_input():
    """Summarizer should trim long input and preserve the lead entity name."""
    long_text = "Anthropic announced in 2026..." * 100
 
    with cassette("summarizer_long_input"):
        result = summarize_long_text(long_text)
 
    # Test the application logic, not Claude's wording.
    assert len(result) < len(long_text) / 2
    assert "Anthropic" in result

The crucial mental model: the test verifies your application's behavior on a fixed Claude response, not the response itself. Claude phrasing the same input differently next month no longer matters because the test runs against the recording.

Pattern 2: Streaming (SSE) cassettes

Streaming responses are sequences of events, so JSONL — one event per line — is the natural format. Replay simply iterates the events in order; you don't need to reproduce the original timing for most tests.

# tests/streaming_cassette.py
"""Record and replay Claude streaming events as JSONL."""
 
import json
from pathlib import Path
from contextlib import contextmanager
from typing import Iterator
from anthropic import Anthropic
from anthropic.types import RawMessageStreamEvent
 
STREAM_CASSETTE_DIR = Path(__file__).parent / "cassettes" / "streams"
 
class CassetteStream:
    """Mock that mirrors the real MessageStreamManager interface."""
 
    def __init__(self, events: list[dict]):
        self._events = events
 
    def __enter__(self):
        return self
 
    def __exit__(self, *args):
        return False
 
    def __iter__(self) -> Iterator[RawMessageStreamEvent]:
        for ev in self._events:
            yield RawMessageStreamEvent.model_validate(ev)
 
@contextmanager
def stream_cassette(name: str, record_mode: str = "once"):
    """Cassette flavored for the streaming API."""
    STREAM_CASSETTE_DIR.mkdir(parents=True, exist_ok=True)
    path = STREAM_CASSETTE_DIR / f"{name}.jsonl"
 
    events = (
        [json.loads(line) for line in path.read_text("utf-8").splitlines()]
        if path.exists() else []
    )
    captured: list[dict] = []
    real_stream = Anthropic().messages.stream
 
    def patched_stream(self, **kwargs):
        if events:
            return CassetteStream(events)
 
        if record_mode == "none":
            raise AssertionError(f"Stream cassette {name} not found.")
 
        original = real_stream(**kwargs)
 
        class RecordingStream:
            def __enter__(_self):
                _self._inner = original.__enter__()
                return _self
 
            def __exit__(_self, *a):
                return _self._inner.__exit__(*a)
 
            def __iter__(_self):
                for ev in _self._inner:
                    captured.append(ev.model_dump(mode="json"))
                    yield ev
 
        return RecordingStream()
 
    from unittest.mock import patch
 
    with patch.object(Anthropic.messages.__class__, "stream", patched_stream):
        try:
            yield
        finally:
            if captured:
                with path.open("w", encoding="utf-8") as f:
                    for ev in captured:
                        f.write(json.dumps(ev, ensure_ascii=False) + "\n")

This buys you a reliable way to test typing-style UIs, token counting, mid-stream cancel handling, and the tricky case of text_delta and input_json_delta blocks interleaving inside a single response.

Pattern 3: Multi-turn Tool Use cassettes

Tool Use is where cassettes shine. A single test can record the entire round trip — Claude asks for a tool, your code runs it, the result goes back, Claude composes the final answer.

# tests/test_weather_agent.py
"""End-to-end check that an agent handles Japanese place names through a tool call."""
 
from src.agents.weather_agent import WeatherAgent
from tests.cassette import cassette
 
def fake_get_weather(location: str) -> dict:
    """Fixed-value tool. If your goal is to test the agent, not the tool,
    avoid hitting external services from the test."""
    return {"location": location, "temp_c": 18, "condition": "clear"}
 
def test_weather_agent_handles_japanese_location():
    """The agent should route a Japanese-language query through the weather tool
    and weave the result into a coherent reply."""
    with cassette("weather_agent_kyoto"):
        agent = WeatherAgent(tool_impl=fake_get_weather)
        reply = agent.ask("What is the weather in Kyoto?")
 
    assert "clear" in reply.lower()
    assert "Kyoto" in reply
    assert agent.tool_call_count == 1

Both round trips end up in one cassette file. CI runs replay both calls without ever touching the network.

Why hashing the message body is the right matcher

Cassette projects argue endlessly about request matching. I shipped a tape-deck-style matcher first (replay in recording order). It broke any time we added retries or parallel calls. The hash-based matcher I describe above survived both refactors and concurrency, for three reasons.

It is order-independent: retry loops and parallel fan-out keep working. It supports incremental recording: a new test added to an existing cassette appends a new entry rather than reordering everything. And it is debuggable: you can cat the JSON file and see which key maps to which response.

The catch is that any non-deterministic content in the request — timestamps, UUIDs, current dates — produces a fresh hash on every run, which never matches a recording. The fix is either a normalization step inside _request_key or a discipline of passing fixed values from your tests. I prefer the latter; explicit beats clever.

CI integration and the "no-record guard"

The single biggest risk of cassette-based testing is recording on CI. If recording runs in CI, three bad things happen at once: production API keys leak into the build environment, your bill creeps up, and tests become non-deterministic again because they re-record every commit.

The guard pattern: force record_mode="none" whenever CI=true is set. Recording is something developers do locally, behind an explicit flag.

# tests/conftest.py
"""Global pytest config — disable recording on CI."""
 
import os
import pytest
 
def pytest_addoption(parser):
    parser.addoption(
        "--record",
        action="store_true",
        default=False,
        help="Enable cassette recording (local development only)",
    )
 
@pytest.fixture
def record_mode(request):
    """Tests receive this fixture and pass it to cassette()."""
    if os.getenv("CI"):
        return "none"
    return "once" if request.config.getoption("--record") else "none"

# Test usage
def test_with_record_guard(record_mode):
    with cassette("my_test", record_mode=record_mode):
        # Run once locally with --record to create the cassette,
        # commit it, and CI will replay it forever after.
        ...

After this is in place, the workflow for adding a new LLM test is: write the test, run pytest --record tests/test_new.py once, commit the generated cassette. CI runs that test for free thereafter.

Replaying production incidents

Cassettes are not just for tests. Designed correctly, the same machinery lets you reproduce a production incident on your laptop in five minutes.

In the system I run, every production request is persisted to Cloudflare R2 with the trace ID as part of the key — requests/{trace_id}.json. When something goes wrong, the workflow has three steps.

Step 1: pull the request/response pair from R2 by trace ID. Step 2: convert it into cassette JSON shape. Step 3: drop it into tests/cassettes/incidents/, write a regression test that mirrors the failing call, and run pytest --record=none.

# scripts/replay_incident.py
"""Pull a production request/response pair from R2 and save it as a cassette."""
 
import json
import sys
from pathlib import Path
import boto3
from tests.cassette import _request_key
 
INCIDENT_CASSETTE_DIR = Path("tests/cassettes/incidents")
R2_BUCKET = "claude-prod-traces"
 
def replay(trace_id: str) -> Path:
    s3 = boto3.client(
        "s3",
        endpoint_url="https://<account>.r2.cloudflarestorage.com",
    )
    obj = s3.get_object(Bucket=R2_BUCKET, Key=f"requests/{trace_id}.json")
    pair = json.loads(obj["Body"].read())  # {"request": ..., "response": ...}
 
    key = _request_key(
        pair["request"]["messages"],
        pair["request"]["model"],
        pair["request"].get("tools"),
    )
    INCIDENT_CASSETTE_DIR.mkdir(parents=True, exist_ok=True)
    cassette_path = INCIDENT_CASSETTE_DIR / f"{trace_id}.json"
    cassette_path.write_text(
        json.dumps({key: pair["response"]}, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(f"OK Cassette written to {cassette_path}")
    return cassette_path
 
if __name__ == "__main__":
    replay(sys.argv[1])

Once this script is in your toolbox, "I can't reproduce it" stops being an excuse. The Slack ping that says "we have a bad response in production" turns into a test case in minutes.

Common pitfalls

Three problems hit teams within days of adopting cassettes. Plan for them upfront.

Cassettes bloat the repo. Even with one cassette per test, streaming cassettes get into the hundreds of lines. A hundred tests is a few megabytes. The fix is either Git LFS for tests/cassettes/, or gzip compression with a .json.gz extension. I prefer the latter — swap json.dump for gzip.open and the storage cost drops by an order of magnitude.

Cassettes leak secrets and PII. If a test's system prompt or messages contain customer data, that data ends up verbatim in the cassette. Committing it is a data leak waiting to happen. Add a sanitization layer between hashing the request and writing the response — mask anything that looks like an API key, an email address, or a UUID. A monthly CI scan that greps every cassette for those patterns is a cheap last-line defense.

SDK upgrades break old cassettes. When Anthropic extends a response type, Message.model_validate on an old cassette can fail because of missing fields. Two strategies help: record the SDK version inside the cassette's _meta field and warn loudly on mismatch; or replace model_validate with model_construct to relax validation. I take the first route and re-record the affected cassettes during the two or three big SDK upgrades that land each year.

Implementing the same in TypeScript / Node.js

The structure ports cleanly to TypeScript. Wrap the Anthropic client with a Proxy that intercepts messages.create and reads or writes the cassette. The Proxy keeps the wrapper thin and avoids the trap of having to maintain a parallel mock of the SDK surface.

// tests/cassette.ts
/**
 * Cassette-based Claude API testing for TypeScript / Node.js.
 * Designed to drop into a Vitest suite.
 */
import fs from "node:fs";
import path from "node:path";
import crypto from "node:crypto";
import Anthropic from "@anthropic-ai/sdk";
import type { Message, MessageCreateParams } from "@anthropic-ai/sdk/resources/messages";
 
const CASSETTE_DIR = path.join(__dirname, "cassettes");
 
function requestKey(params: MessageCreateParams): string {
  const payload = JSON.stringify({
    messages: params.messages,
    model: params.model,
    tools: params.tools ?? [],
  });
  return crypto.createHash("sha256").update(payload).digest("hex").slice(0, 16);
}
 
export async function withCassette<T>(
  name: string,
  fn: (client: Anthropic) => Promise<T>,
  recordMode: "once" | "none" = process.env.CI ? "none" : "once",
): Promise<T> {
  fs.mkdirSync(CASSETTE_DIR, { recursive: true });
  const cassettePath = path.join(CASSETTE_DIR, `${name}.json`);
  const recordings: Record<string, Message> = fs.existsSync(cassettePath)
    ? JSON.parse(fs.readFileSync(cassettePath, "utf-8"))
    : {};
  const newRecordings: Record<string, Message> = {};
 
  const realClient = new Anthropic();
  const proxiedClient = new Proxy(realClient, {
    get(target, prop, receiver) {
      if (prop === "messages") {
        return new Proxy(target.messages, {
          get(messagesTarget, messagesProp) {
            if (messagesProp === "create") {
              return async (params: MessageCreateParams) => {
                const key = requestKey(params);
                if (recordings[key]) return recordings[key];
                if (recordMode === "none") {
                  throw new Error(
                    `Cassette ${name} has no recording for key=${key}`,
                  );
                }
                const res = await messagesTarget.create(params);
                newRecordings[key] = res as Message;
                return res;
              };
            }
            return Reflect.get(messagesTarget, messagesProp);
          },
        });
      }
      return Reflect.get(target, prop, receiver);
    },
  });
 
  try {
    return await fn(proxiedClient);
  } finally {
    if (Object.keys(newRecordings).length > 0 && recordMode !== "none") {
      fs.writeFileSync(
        cassettePath,
        JSON.stringify({ ...recordings, ...newRecordings }, null, 2),
      );
    }
  }
}

For streaming, wrap the AsyncIterator returned by client.messages.stream() with a recording iterator that captures every event before yielding it.

How this composes with other production patterns

Cassette testing is most valuable when it sits inside a wider production playbook. Combine it with the retry logic in Claude API Production Rate Limits and Billing Guide, and your retry behavior itself becomes deterministic — recordings can include failed responses to drive the retry path.

Pair it with the patterns in Claude API Streaming Resilience for Production and you can verify your reconnect-after-disconnect flow against a recorded stream that ends mid-message.

For longer-term quality, the LLM Evaluation Pipeline Guide plugs neatly above the cassette layer — feed cassette responses into your evaluator to keep regression-checking the evaluator itself.

Running cassettes across a team

The moment more than one engineer maintains the same cassette, merge conflicts become real. The two practices that worked for my team: split into one file per test, and always sort the JSON keys before saving. With sorted keys and small files, two engineers re-recording the same test produce diffs that are trivial to merge.

Add cassette review to the PR checklist. Reviewers should open the generated JSON and confirm there is no unexpected customer data, no embarrassing model output, and that tool_use inputs match the intent of the test. Spending two minutes on this saves dramas later.

For long-lived projects, schedule a monthly cassette audit script that scans every recording for likely API keys, email addresses, and credit-card-shaped sequences. CI catches what humans miss.

Avoiding races under parallel test runners

Run tests under pytest-xdist and the cassette directory becomes a contention point. One-cassette-per-test eliminates write conflicts; the only remaining race is parent directory creation. Make mkdir idempotent (parents=True, exist_ok=True) and write through a temp file with an atomic rename.

def _atomic_write_json(path: Path, data: dict) -> None:
    """Safe write under concurrent runners."""
    tmp = path.with_suffix(path.suffix + ".tmp")
    with tmp.open("w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2, sort_keys=True)
    tmp.replace(path)  # atomic on POSIX

This is fully atomic on Linux and macOS. On Windows, ensure the temp file is on the same filesystem as the target — os.replace is the cross-platform escape hatch.

Catching prompt regressions

Cassettes also catch the "I tweaked a prompt and something far away broke" failure mode. Keep the existing cassettes in place, change your prompt, then run with record_mode="new_episodes" to record the new responses under fresh keys. Diff the two and you have a precise picture of how the change rippled through every test case.

In my team's workflow, every PR that changes a system prompt produces an automatic "cassette diff report" in CI, and reviewers must look at the diff before approving. We have caught subtle regressions this way — for instance, a "make the assistant slightly less formal" change that quietly switched code-comment language from English to a mix of English and the user's locale. Without the diff, that would have shipped.

When cassettes are the wrong tool

Three situations call for a different approach.

You are evaluating model upgrades. When you bump from one Claude version to another and want to know how response quality changed, replaying old cassettes tells you nothing. Use a real-API evaluation harness — see the LLM Evaluation Pipeline Guide.

You need to verify mid-stream cancellation behavior. Cassettes replay all recorded events. They cannot model what happens when the server is asked to stop generating mid-message. Pair cassettes with a small set of low-frequency live integration tests for these scenarios.

You want to be the first to notice new SDK fields. Cassettes are frozen snapshots and won't reveal that Claude has added a new stop_reason value. Schedule quarterly re-recording of a representative subset of tests with record_mode="new_episodes" to surface those changes.

Use cassettes inside their sweet spot — making the bulk of your tests deterministic and free — and run a thin layer of live tests for the cases above. The combination is what gives you both signal and stability.

Designing for replay from day one

The single biggest mistake I see teams make is treating cassettes as something they will add "after we ship." By then the application is full of code paths that record incidental state — wall-clock timestamps in the prompt, current process IDs in the system message, request UUIDs threaded into tool input. Each of those breaks the request matcher, and each requires either a refactor or a brittle normalization step.

If you are starting fresh, design with replay in mind. Pass timestamps and IDs in through explicit parameters rather than reading them from the environment. Keep the system prompt static and put dynamic context in the user message in a structured form. Make the seed for any random sampling (temperature, top_p, top_k) part of the test fixture rather than a global default. None of these requires extra code in production — it is a handful of decisions about where values come from.

The same principle applies to tool definitions. If a tool's schema is generated from runtime configuration, regenerating it can change the JSON byte-for-byte even when the semantic schema is identical. The cassette key changes, the cassette misses, and the test re-records on its next run — silently. Pin the tool schema definitions to a stable serialization format (sorted keys, explicit type fields) and the problem disappears.

This level of discipline pays off the moment you start chasing a bug. A clean, deterministic request makes the cassette diff readable, the failure reproducible, and the fix verifiable in one cycle.

Observability is the other half of replay

Cassettes give you reproducibility. Observability gives you the trace ID that points to the right cassette to reproduce. Both are necessary; either alone is incomplete.

The minimum bar is propagating a trace_id through every Claude request, capturing the full request and response pair against that ID, and storing the pair somewhere you can fetch from outside production. I use Cloudflare R2 because it is cheap and addressable from anywhere; S3, GCS, or even a logging pipeline that lets you export raw bytes works just as well.

Tag each pair with the model name, the SDK version, the deployment commit SHA, and the customer or workspace ID (with PII stripped). When an incident report comes in, those tags collapse "search the logs for an hour" into a single trace ID lookup. From there, the replay_incident.py script in the previous section turns the pair into a cassette and you have a deterministic test in front of you.

The discipline that makes this scale is treating the production trace store as the source of truth and the cassette directory as derived. Never edit a cassette by hand — re-derive it from the trace. Never lose the trace store — back it up at the same cadence as your production database. The combination is what lets you say, with confidence, "I can reproduce any production failure from the past 90 days on my laptop."

Deciding what not to test with cassettes

Even inside the sweet spot, there is a discipline question: which tests deserve a cassette? Recording everything is cheap in CPU but expensive in maintenance. Each cassette is a contract you have to update when the underlying behavior changes intentionally.

The test suites I keep on cassettes are: the highest-value end-to-end flows (one or two per major user journey), the regression tests for every fixed bug (one cassette per bug, named after the issue ID), and the canonical examples from the documentation that I want to keep working forever. Anything else either runs as a fast unit test against a dummy Anthropic shim, or runs nightly against the live API.

This trichotomy — unit tests with shims, cassette-backed integration tests, nightly live tests — is the layered approach that buys you both speed and signal. Cassettes are the middle layer. They are not the whole pyramid, but they are the layer that most teams skip and most regret skipping.

Next step

Pick one slow integration test in your repo. Add tests/cassette.py, wrap the test in with cassette("my_test"):, and run pytest --record once. The next CI run for that test will be free and instant. Once you have felt the difference on a single test, make "new LLM tests must use cassettes" a team rule. The compounding payoff arrives within a sprint or two: deterministic CI, reproducible production failures, and a quiet test suite that finally tells you the truth about your code.