cupo 0.0.1

Usage limits, feature gates, and metering for AI products. Drop-in for FastAPI and Next.js.

Cupo

Usage limits, feature gates, and metering for AI products. Drop-in for FastAPI and Next.js.

⚠️ Status: design phase / request for comments. This README describes the API we intend to build. We're validating the design before writing the code — if this would (or wouldn't) solve a problem for you, please open an issue or comment. Brutal honesty welcome.

---

The problem

Billing platforms (Stripe, Lago, Metronome, Orb) answer "how much do I charge?" — they count usage and generate invoices after the fact.

None of them answer the question your code asks a hundred times per second: "is this customer allowed to do this, right now?"

So every AI SaaS ends up hand-rolling the same thing:

• A usage table with counters that break under concurrent requests
• Plan limits scattered across if customer.plan == "pro" checks
• Token counting glued onto LLM calls (and silently wrong for streaming)
• A cron job that resets counters monthly (usually in the wrong timezone)
• No warning to the customer before they hit the wall

Cupo is that layer, done once, done right, and open source.

What it looks like

from cupo import Cupo

cupo = Cupo()  # embedded mode: uses your existing Postgres, no server needed

@app.post("/chat")
@cupo.protect(feature="ai_chat")            # blocks with 429 + upgrade info if over limit
async def chat(req: ChatRequest, customer: Customer):
    ...

Or with explicit control:

res = cupo.check(customer.id, feature="ai_chat", units=1)   # atomic check-and-consume
if not res.allowed:
    return JSONResponse(429, {"error": "plan_limit", "resets_at": res.resets_at,
                              "upgrade_url": res.upgrade_url})

reply = anthropic.messages.create(...)

cupo.track(customer.id, feature="ai_chat",
           tokens=reply.usage.output_tokens,
           idempotency_key=req.id)          # safe to retry, never double-counts

Plans as code

Plans live in a versioned YAML file in your repo — reviewable in a PR, not hidden in a dashboard:

# cupo.yaml
features:
  ai_chat:    { unit: message }
  ai_tokens:  { unit: token }
  pdf_export: { unit: export }

plans:
  free:
    ai_chat:    { limit: 50,     window: month }
    ai_tokens:  { limit: 100_000, window: month }
    pdf_export: false

  pro:
    ai_chat:    { limit: 5_000,  window: month, on_limit: degrade }  # block | degrade | bill
    ai_tokens:  { limit: 10_000_000, window: month }
    pdf_export: true

  enterprise:
    ai_chat:    unlimited
    ai_tokens:  { limit: 100_000_000, window: month, on_limit: bill, overage_price: 0.50/1_000_000 }
    pdf_export: true

on_limit policies:

• block — deny the request (default)
• degrade — allow it, but flag res.degraded = True so you can route to a cheaper model
• bill — allow it and emit an overage event your billing system can invoice

Token-aware AI helpers

The part everyone gets wrong. Cupo ships thin wrappers around the Anthropic and OpenAI clients that meter tokens automatically — including streaming, where usage is only known when the stream ends:

from cupo.anthropic import metered

client = metered(anthropic.Anthropic(), cupo, feature="ai_tokens")

# streaming: tokens are tracked when the stream closes, with the request's idempotency key
with client.messages.stream(model="claude-sonnet-4-6", ...) as stream:
    for text in stream.text_stream:
        yield text

How it works

your app ──▶ Cupo SDK ──▶ counters (your Postgres, or Redis, or Cupo server)
                │
                ├─ local entitlement cache (checks add ~0 network latency)
                └─ async usage flush (batched, idempotent)

Three problems Cupo solves so you don't have to:

1. Atomicity. Two concurrent requests with one credit left: exactly one passes. Counters use atomic operations (INCR / row-level locks), never read-modify-write.
2. Latency. Entitlements are cached in-process and synced in the background. A check() is a dictionary lookup, not a network call. Trade-off: near the limit, a small overshoot is possible — this is configurable (strict: true forces a synchronous check for expensive features).
3. Idempotency. Every track() takes an idempotency key. Retries, at-least-once queues, and network flakiness never double-count.

Failure mode is yours to choose: fail_open (if Cupo is unreachable, allow the request — default, your product stays up) or fail_closed (deny — for features where overshoot costs you real money).

Deployment modes

Mode	What it needs	For
Embedded	Your existing Postgres (Supabase works)	Solo devs, single service
Server	Docker container + Redis	Multiple services / languages
Cloud (planned)	Nothing — hosted	Teams that want dashboards, analytics, SLA

Webhooks

The server emits events so you can warn customers before they hit the wall:

• usage.threshold — configurable (e.g. at 80% of any limit)
• usage.limit_reached
• usage.overage — with units and computed price, ready to forward to your billing

What Cupo is not

• Not a billing platform. It doesn't generate invoices or charge cards. It pairs with Stripe, Mercado Pago, Lago, or whatever you already use (plan sync integrations are on the roadmap).
• Not an API gateway. It runs inside your app, not in front of it.
• Not for enterprise contract management. If you have negotiated multi-year commits with drawdowns, you want Metronome or Orb.

FAQ

vs. Lago / Metronome / Orb? Those meter usage to bill it. Cupo meters usage to enforce it, in the request path, in real time. Different layer — Cupo can feed them.

vs. Stigg? Closest neighbor. Stigg is a hosted, dashboard-first entitlements platform aimed at teams. Cupo is open source, config-as-code, and designed so a solo developer is enforcing limits 15 minutes after pip install cupo.

vs. rolling my own? You can. Most people's version has the concurrency bug, misses streaming tokens, and has no idempotency. Ours has tests for all three.

Why should I trust the counters? Every counter mutation is an atomic operation with tests that hammer it concurrently. The test suite is part of the pitch — read it.

Roadmap

• v0.1 — Python SDK, embedded mode (Postgres), plans-as-code, FastAPI middleware, Anthropic/OpenAI metered wrappers, docs + 3 runnable examples
• v0.2 — Standalone server (Docker), TypeScript SDK, webhooks, Redis counters
• v0.3 — Stripe & Mercado Pago plan sync, usage dashboard, hosted cloud (free tier + flat self-serve pricing — no "talk to sales")

License

SDKs: MIT. Server: AGPL-3.0. Self-hosting is free forever; the hosted cloud is how the project sustains itself.

---

Would you use this? Open an issue titled feedback: and tell us — especially if the answer is no and why. If you've hand-rolled this layer before, we'd love 20 minutes of your war stories.