Cassetter

Rust-powered HTTP cassette recorder. Safe by default.

Available for Python (this README), Node/TypeScript (ts/), Go (go/), and Rust (crates/cassetter/). Every SDK uses the same cassette format.

Why?

VCR.py works, but has fundamental problems:

• Unsafe by default - doesn't filter sensitive headers, tokens, or API keys
• Unsafe YAML - uses yaml.load() with an unsafe loader that can execute arbitrary Python code from cassette files
• Slow - pure Python YAML parsing, matching, and serialization
• Fragile - relies on undocumented internals that break on minor version bumps
• Poor readability - JSON bodies stored as escaped strings in YAML

cassetter fixes all of this with a Rust core (PyO3) for speed, safe-by-default security filtering, and secure YAML parsing.

Install

uv add cassetter

Go

Use the Go module when you need HTTP, gRPC, or WebSocket cassette recording:

go get github.com/Kludex/cassetter/go@v0.1.0

It reads and writes the same structured YAML and TOML formats, including VCR.py YAML migration. It supports all four gRPC call patterns and WebSocket text and binary messages with YAML cassettes. It also provides inspect, diff, scrub, and convert commands. See the Go documentation for complete examples and record modes.

Rust

Use the Rust crate when the code under test sends requests through reqwest or a generated tonic client:

cargo add cassetter reqwest

The Rust SDK uses explicit transport injection. It supports async HTTP and unary gRPC recording, exact protobuf request matching, concurrent use, safe filtering, and explicit finalization. See the Rust documentation for complete examples and current streaming limits.

Quick start

With pytest (recommended)

Mark tests with @pytest.mark.vcr:

import httpx
import pytest


@pytest.mark.vcr
async def test_api_call():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/users")
    assert response.status_code == 200

First run records real HTTP interactions. Subsequent runs replay from the cassette file - no network needed.

If you need direct access to the cassette (e.g. to inspect recorded interactions), request the fixture explicitly:

from cassetter import Cassette


@pytest.mark.vcr
async def test_with_cassette(cassette: Cassette):
    ...
    assert len(cassette.interactions) == 1
    assert cassette.interactions[0].request.method == "GET"

Recorded objects (HttpRequest, HttpResponse, HttpInteraction, Body, and the gRPC and WebSocket equivalents) are immutable value objects: they compare by value, and attribute assignment raises AttributeError. Use replace() to derive a modified copy:

recorded = cassette.interactions[0].request
probe = recorded.replace(uri="https://api.example.com/other")

assert probe != recorded
assert probe.method == recorded.method

To rewrite what gets recorded, use the before_record_request and before_record_response hooks, which receive mutable RawRequest and RawResponse dataclasses.

With the context manager

from cassetter import use_cassette

with use_cassette("tests/cassettes/my_test.yaml", record_mode="once"):
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/users")

With a reusable configuration

Cassetter holds the options shared by a group of cassettes, so they are declared once instead of on every call:

from cassetter import Cassetter

recorder = Cassetter(
    cassette_library_dir="tests/cassettes",
    record_mode="none",
    filter_headers=["x-gateway-apikey"],
    before_record_request=my_request_hook,
)

with recorder.use_cassette("openai.yaml"):
    ...

# override any option for a single cassette
with recorder.use_cassette("openai.yaml", record_mode="all"):
    ...

It takes every option use_cassette() takes, plus cassette_library_dir - the directory cassette names are resolved against. The object is frozen and callable, so recorder("openai.yaml") works too.

The vcr_config fixture accepts a Cassetter, so one object can configure both the pytest suite and direct use_cassette() calls:

@pytest.fixture(scope="module")
def vcr_config() -> Cassetter:
    return recorder

Record modes

A session that can record only replays interactions it hasn't played yet, and never replays one it just recorded. A repeated request, like the next turn of a conversation posting to the same URL, reaches the server instead of getting an earlier response back.

Set via CLI: pytest --record-mode=none

Safe by default

Sensitive data is filtered at write time - cassettes never contain secrets. These headers are stripped automatically:

authorization, cookie, set-cookie, x-api-key, api-key, x-auth-token, proxy-authorization, www-authenticate, x-goog-api-key, x-amz-security-token

Query params like api_key, access_token, token, client_secret are replaced with [FILTERED].

JSON body fields like password, access_token, refresh_token, client_secret are scrubbed.

Filtering applies to every protocol: HTTP headers, query params, and bodies; gRPC request/response metadata and the json_debug payload; WebSocket handshake headers and text/JSON frame bodies. Binary protobuf bodies are stored as-is - they cannot be pattern-scrubbed.

Customize filtering:

from cassetter import use_cassette

with use_cassette(
    "cassette.yaml",
    filter_headers=["x-custom-secret"],
    body_scrub_patterns=["my_secret_field"],
    filter_replacement="***REDACTED***",
):
    ...

These add to the built-in lists rather than standing in for them, so naming one more header to scrub never starts recording the ones above.

Cassette format

Cassettes can be stored as YAML (default) or TOML. The format is detected by file extension (.yaml / .yml for YAML, .toml for TOML).

YAML (default)

JSON bodies are stored as structured YAML - not escaped strings:

version: 1
interactions:
  - request:
      method: POST
      uri: https://api.openai.com/v1/chat/completions
      headers:
        content-type:
          - application/json
      body:
        type: json
        content:
          model: gpt-4o
          messages:
            - role: user
              content: Hello!
    response:
      status: 200
      headers:
        content-type:
          - application/json
      body:
        type: json
        content:
          id: chatcmpl-abc123
          choices:
            - message:
                role: assistant
                content: Hi there!
    recorded_at: '2026-02-20T10:30:01Z'

TOML

Use .toml extension for TOML cassettes. Body content is stored as a JSON string since TOML cannot represent null values or heterogeneous arrays:

with use_cassette("cassette.toml"):
    ...

TOML loads ~2.8x faster than YAML and produces ~12% smaller files (saves are slower).

Request matching

Default: match on method + URI. Configurable:

from cassetter import use_cassette

with use_cassette(
    "cassette.yaml",
    match_on=["method", "uri", "json_body"],
    ignore_json_paths=["request_id", "timestamp"],
):
    ...

Available matchers: method, uri, headers, body, json_body.

Supported libraries

By default, interceptors are auto-detected based on which libraries are installed. To limit interception to specific libraries:

with use_cassette("cassette.yaml", intercept=["httpx", "aiohttp"]):
    ...

Concurrency

Multiple cassettes can run concurrently in the same process - each use_cassette context gets its own isolated cassette via contextvars.ContextVar. This works out of the box with asyncio.gather, anyio.create_task_group, and any framework that creates async tasks (e.g. Pydantic Evals with max_concurrency > 1).

async def task_a():
    with use_cassette("cassettes/a.yaml", record_mode="none"):
        async with httpx.AsyncClient() as client:
            return await client.get("https://api.example.com/data")


async def task_b():
    with use_cassette("cassettes/b.yaml", record_mode="none"):
        async with httpx.AsyncClient() as client:
            return await client.get("https://api.example.com/data")


# Each task uses its own cassette - no cross-contamination
results = await asyncio.gather(task_a(), task_b())

Nested cassettes work too - the inner cassette overrides the outer, and the outer is restored when the inner exits.

Threads

For ThreadPoolExecutor, each thread has its own context by default (no cassette). To propagate the current cassette into a thread, use contextvars.copy_context():

with use_cassette("cassette.yaml", record_mode="none"):
    ctx = contextvars.copy_context()

    def work():
        with httpx.Client() as client:
            return client.get("https://api.example.com/data")

    with ThreadPoolExecutor() as pool:
        future = pool.submit(ctx.run, work)
        result = future.result()

Without copy_context(), threads see no active cassette and requests pass through to the real server.

gRPC support

Install the gRPC extra:

uv add "cassetter[grpc]"

Record and replay gRPC calls by adding "grpc" to the interceptor list:

with use_cassette("cassette.yaml", intercept=["grpc"]):
    channel = grpc.aio.insecure_channel("localhost:50051")
    stub = my_service_pb2_grpc.MyServiceStub(channel)
    response = await stub.Echo(my_service_pb2.EchoRequest(message="hello"))

All four gRPC call patterns are supported: unary-unary, server streaming, client streaming, and bidirectional streaming. Request and response bodies are stored as binary in the cassette, with an optional json_debug section for human-readable protobuf representation (when google.protobuf is available):

grpc_interactions:
  - request:
      method: /mypackage.MyService/Echo
      metadata: {}
      body:
        type: binary
        content: 0a0568656c6c6f
    response:
      status_code: 0
      status_message: OK
      metadata: {}
      body:
        type: binary
        content: 0a0568656c6c6f
    json_debug:
      request:
        message: hello
      response:
        message: hello

Streaming responses use length-prefixed binary encoding - multiple response chunks are stored in a single body field and decoded back into individual messages on replay.

WebSocket support

Install the WebSocket extra:

uv add "cassetter[websockets]"

Record and replay WebSocket connections:

with use_cassette("cassette.yaml", intercept=["websockets"]):
    async with websockets.connect("wss://ws.example.com/stream") as ws:
        await ws.send('{"subscribe": "ticker"}')
        data = await ws.recv()

WebSocket interactions record each frame with direction, type, and timing offset:

ws_interactions:
  - uri: wss://ws.example.com/stream
    headers: {}
    frames:
      - direction: send
        frame_type: text
        body:
          type: text
          content: '{"subscribe": "ticker"}'
        offset_ms: 0
      - direction: recv
        frame_type: text
        body:
          type: json
          content:
            price: 42.5
        offset_ms: 120

On replay, recv() returns recorded frames in order without a real connection, then raises the recorded close status. Older cassettes without a close frame use ConnectionClosedOK. send() is a no-op. Text, binary, and close frames are supported, and both async with websockets.connect(...) and ws = await websockets.connect(...) work.

Streaming / SSE support

SSE (Server-Sent Events) responses - used by OpenAI, Anthropic, Groq, and other LLM APIs for streaming - work out of the box. The full response body is recorded as readable text in the cassette:

response:
  status: 200
  headers:
    content-type:
      - text/event-stream
  body:
    type: text
    content: |+
      data: {"id":"chatcmpl-abc","choices":[{"delta":{"role":"assistant"}}]}

      data: {"id":"chatcmpl-abc","choices":[{"delta":{"content":"Hello"}}]}

      data: [DONE]

On replay, the buffered body is returned to the client SDK, which parses SSE events from it. This matches how VCR.py handles streaming - chunk boundaries aren't preserved, but SSE parsers split on \n\n boundaries regardless of how bytes are delivered.

Request filtering

Ignore hosts

Bypass the cassette entirely for requests to specific hosts. Matched requests pass through to the real server - no recording, no replay:

with use_cassette(
    "cassette.yaml",
    ignore_hosts=["*.googleapis.com", "accounts.google.com"],
):
    ...

Patterns use fnmatch syntax (* matches any sequence of characters). Combine with ignore_localhost for full control:

with use_cassette(
    "cassette.yaml",
    ignore_localhost=True,
    ignore_hosts=["*.googleapis.com"],
):
    ...

Before record request hook

Use a callback that runs before each request is recorded or replayed. Return the (possibly modified) RawRequest. Raise SkipRecording to let the request pass through live:

from cassetter import RawRequest, SkipRecording, use_cassette


def my_hook(request: RawRequest) -> RawRequest:
    if not request.uri.startswith("https://api.mycompany.com"):
        raise SkipRecording
    # Strip auth header before recording
    request.headers.pop("authorization", None)
    return request


with use_cassette("cassette.yaml", before_record_request=my_hook):
    ...

Before record response hook

Modify or discard responses before they are recorded. Return the (possibly modified) RawResponse. Raise SkipRecording to skip recording the interaction:

from cassetter import RawResponse, SkipRecording, use_cassette


def my_hook(response: RawResponse) -> RawResponse:
    if response.status >= 500:
        raise SkipRecording  # don't record server errors
    # Strip a volatile header
    response.headers.pop("x-request-id", None)
    return response


with use_cassette("cassette.yaml", before_record_response=my_hook):
    ...

Both hooks work with the pytest plugin via vcr_config:

@pytest.fixture(scope="module")
def vcr_config():
    return {
        "ignore_hosts": ["*.googleapis.com"],
    }

Cassette expiry

Force re-recording when cassettes get stale:

with use_cassette("cassette.yaml", max_age="30d", on_expiry="rerecord"):
    ...

max_age accepts durations like "24h", "7d", "4w". on_expiry controls the behavior:

Also configurable via pytest:

# pyproject.toml
[tool.pytest.ini_options]
vcr_max_age = "30d"
vcr_on_expiry = "warn"

Or per-test:

@pytest.mark.vcr(max_age="7d", on_expiry="fail")
async def test_fresh_data(): ...

Unplayed interactions

A test that stops sending a recorded request still passes on replay. Fail it instead:

# pyproject.toml
[tool.pytest.ini_options]
vcr_on_unplayed = "fail"  # or "warn"; the default is "ignore"

Passing tests on replay-only cassettes then fail at teardown with the unplayed interactions by protocol. Set it per module with on_unplayed in vcr_config, or per test with @pytest.mark.vcr(on_unplayed=...).

Orphan detection

Find cassette files that no test uses:

pytest --vcr-check-orphans=tests/cassettes/

Performance

Cassetter's Rust core is faster than vcrpy (compared against vcrpy with libyaml, its fastest configuration). Matching is the number a test suite actually feels - it runs once per request, while load and save run once per test:

                    cassetter    vcrpy        speedup
  10 interactions
  load              205 us       471 us       2.3x
  match             0.9 us       12.6 us      13.7x
  save              252 us       456 us       1.8x

  1000 interactions
  load              18.1 ms      52.8 ms      2.9x
  match             0.8 us       1.22 ms      1573.7x
  save              6.5 ms       42.5 ms      6.5x

Match cost is constant in cassette size: the method+URI index is built once and cached on the cassette, and matching runs against the interactions Rust already owns rather than copying them across the FFI boundary per request. vcrpy's linear scan is why its match column grows with N and cassetter's does not.

Absolute timings are machine dependent; the speedup ratios are the portable part. Load speedup also depends on cassette shape: many small interactions (as above) is the hardest case for the parser, while cassettes with large bodies (e.g. LLM/SSE responses) load proportionally faster.

TOML cassettes (.toml) load ~2.8x faster than YAML and produce ~12% smaller files (at the cost of slower saves):

  1000 interactions
                      YAML         TOML
  save                10.7 ms      18.0 ms
  load                53 ms        18.6 ms
  size                768 KB       675 KB

Run uv run python benchmarks/bench.py and uv run python benchmarks/bench_formats.py to reproduce.

YAML safety

vcrpy uses yaml.load() with an unsafe loader (CLoader/Loader) that can execute arbitrary Python via !!python/object tags. A malicious cassette file could run code when loaded.

cassetter parses YAML in Rust with serde-saphyr - no Python object construction, no unsafe code, panic-free on malformed input, and hard budgets against alias-expansion attacks (billion laughs). Only data types are supported.

Migrating from pytest-recording / VCR.py

cassetter is designed as a drop-in replacement. Most projects can migrate with minimal changes.

Cassette files

Existing VCR cassettes work as-is - cassetter reads both VCR format and its own format. On the next re-record, cassettes are written in cassetter's format with structured JSON bodies instead of escaped strings.

To bulk-convert existing cassettes to a different format, use the CLI:

# Convert a single file
cassetter convert cassette.yaml cassette.toml

# Convert all cassettes in a directory (in-place, changing extension)
cassetter convert tests/cassettes/ toml

# Rewrite in-place keeping the same format (VCR -> cassetter migration)
cassetter convert tests/cassettes/ yaml --force

# Convert to a separate output directory
cassetter convert tests/cassettes/ output/ --to toml

Conversion applies the default security filtering (headers, query params, body fields), so secrets recorded by VCR.py are removed on the way through. Pass --no-scrub to skip it.

pytest plugin

cassetter uses the same @pytest.mark.vcr marker, vcr_config fixture, and --record-mode CLI flag. Key differences:

Architecture

Everything that defines a cassette - the file format (YAML and TOML), request matching, security filtering, and body processing - lives in one Rust crate. Each language gets a thin binding over it rather than its own reimplementation, so the parts that matter cannot drift apart.

crates/
  cassetter-core/     pure Rust: format, matching, security, body processing
  cassetter/          async Rust SDK   -> reqwest and tonic
  cassetter-python/   PyO3 bindings    -> cassetter._core
  cassetter-node/     napi-rs bindings -> cassetter.node
src/cassetter/        Python package (interceptors, pytest plugin, CLI)
ts/                   Node package (fetch interception)
conformance/          shared fixtures every SDK must reproduce

A binding only implements what is genuinely language-specific: HTTP library interception and the idiomatic record/replay wrapper.

Adding a language

Add a crate under crates/ that wraps cassetter-core, then make it pass conformance/. That suite pins the cassette format across bindings - see its README for the contract.

Development

Requires Rust 1.88+, Python 3.10+, and Node 18+ for the Node binding.

git clone https://github.com/Kludex/cassetter.git
cd cassetter

# Rust core
cargo test --workspace

# Python
uv sync
uv run maturin develop
uv run pytest

# Node
cd ts && npm install && npm run build:native && npm test

Changes to matching, security, the cassette format, or body handling belong in crates/cassetter-core so every binding gets them at once.

Use the release checklist to publish every SDK from one version tag.