from anthropic import Anthropic
from cachemux import cachemux, Recorder

recorder = Recorder()
client = cachemux(Anthropic(), recorder)   # the only change to your code

# ... use the client exactly as before ...

print(recorder.report(base_input_per_mtok=5.0))

cachemux report
  calls           : 3 (0 errored)
  hit rate        : 66.7%
  cache reads     : 400,000
  cache writes    : 200,000
  net saved       : 1.5500  (saving money)

prefix: anthropic / claude-opus-5
  stable prefix     : 1 blocks, ~1240 tokens
  volatile blocks   :
    [1] messages[0]: 3 variants, ~9 tokens uncacheable from here
  verdict: worth it - a write costs 1.25x, so it needs 1.28 reads; 2.0 expected

Try it with no API key: python examples/demo.py

Why

Every major provider caches your prompt prefix, and every one does it differently:

Same headline discount, three different cost models. So "should I cache this?" has a different answer per provider, and caching can lose money outright:

• A cache written and never read costs more than not caching.
• A rented Gemini cache accrues rent whether or not anyone reads it.
• Anthropic silently skips caching below the model minimum, with no error.

A hit-rate dashboard shows none of that.

What it does

• Measures — normalises every provider's cache token counts into one shape, so hit rate and real cost are comparable across providers.
• Analyses — splits each request into hashed blocks and names the exact block stopping your prefix from caching. A timestamp in your system prompt voids everything after it; cachemux points at it.
• Decides — computes the break-even (how many reads a cache needs to pay for its own write, or its rent) and compares it against your measured reuse.
• Acts — the default. Once the math clears, requests get caching set up on the way out: a cache_control breakpoint on Anthropic, a pinned prompt_cache_key on OpenAI, a rented explicit cache on Gemini — rented only when it beats the implicit caching Gemini already does for free. auto=False to only observe.
• Diagnoses — when the hit rate drops, diagnose() compares two traffic windows and names the block that rotated the cache key.

How it works

flowchart LR
    A[your code] --> P{PREPARE<br/>four gates}
    P -->|worth it| ACT[breakpoint / cache key / rental]
    P -->|not yet| S[SEND]
    ACT --> S
    S --> API[(provider API)]
    API --> O[OBSERVE + RECORD]
    O -.->|next request reads this| P

Every request is fingerprinted; every caching decision is made from what was actually measured, never a guess. If cachemux fails at any step, the call still goes through uncached — a caching layer must never take down an app, and there's a test for it.

The math, in one breath

An Anthropic cache write costs 1.25x base input; a read costs 0.1x:

reads needed = (1.25 − 0.1) / (1 − 0.1) = 1.28 reads to break even

cachemux acts only when your measured reuse clears that bar. No dollar prices are bundled — the decision runs on multipliers, and you supply your own base rate for money reports. A shipped price list goes stale silently; yours doesn't.

Design rules

1. Zero config. Import it and it works.
2. If cachemux fails, the call still goes through uncached.
3. No new abstraction to learn. Keep using your provider's SDK exactly as before.
4. No runtime dependencies.
5. No bundled price list.

Layout

src/cachemux/
├── providers/     one module per provider, same four functions each
│   ├── base.py        shared Usage / Applied types
│   ├── anthropic.py   breakpoints, usage fields, model minimums
│   ├── openai.py      prompt_cache_key routing
│   └── google.py      rent-or-not planning
├── analysis/      prefix.py (stability) + drift.py (what changed)
├── economics/     pricing.py (cost models, break-even)
└── core/          proxy.py (interception) + recorder.py (aggregation)

Adding a provider is one module in providers/ and one line in its registry. Nothing above that layer branches on which provider it's holding.

Status

Working: measurement, prefix analysis, break-even decisions, drift diagnosis, and automatic application (the default) on all three providers — live-verified against Gemini (rented explicit cache) and OpenAI (accepted prompt_cache_key).

Not measured yet: streaming. A create(stream=True) call returns before usage exists, so such calls are recorded but excluded from the hit rate rather than miscounted as misses.

Development

uv venv --python 3.11
uv pip install -e ".[dev]"
pytest -q
ruff check src tests

Licence

MIT