Implementing Usage-Based Billing for Claude API Services — Token Tracking, Price Conversion, and Stripe Metering from Scratch

When you're building a Claude API-powered service, the billing model decision is one that will shape every downstream technical choice.

Flat monthly pricing is simple, but heavy users eat your margins. Pure pay-as-you-go is fair, but it makes revenue unpredictable and increases churn. Usage-based billing threads the needle — but the implementation is genuinely complex.

Why Usage-Based Billing (and Why It's Harder Than It Looks)

There are two strong reasons to choose usage-based pricing for AI services.

Margin alignment: Claude API charges by token. Flat-rate plans create a structural risk where heavy users consume costs faster than their subscription covers. Tying your pricing to actual usage keeps your unit economics predictable.

User fairness: Users who interact with your service twice a month and users who interact 500 times shouldn't pay the same amount. The lighter user will churn. Usage-based pricing removes that churn driver.

The implementation challenge is the gap between API-level granularity (tokens) and business-level pricing (dollars, credits, or seats). You need an abstraction layer between them.

My preferred approach: a credit system. 1 credit = $0.01 internally. You apply a markup to the raw API cost and convert it to credits. Users purchase credits; API calls consume them. This gives you pricing flexibility without directly exposing your API cost structure.

Step 1: Token Measurement and Usage Logging

Claude API responses include a usage field with input and output token counts. This is your billing source of truth.

import anthropic
import logging
from dataclasses import dataclass
from datetime import datetime, UTC
import math
 
client = anthropic.Anthropic()
logger = logging.getLogger(__name__)
 
@dataclass
class UsageRecord:
    """Single API call usage data — record this in your database."""
    user_id: str
    model: str
    input_tokens: int
    output_tokens: int
    cache_creation_tokens: int
    cache_read_tokens: int
    timestamp: datetime
    request_id: str
 
# Pricing per 1M tokens (USD) — update when Anthropic adjusts pricing
MODEL_PRICING = {
    "claude-opus-4-6": {
        "input": 15.00,
        "output": 75.00,
        "cache_creation": 18.75,
        "cache_read": 1.50,
    },
    "claude-sonnet-4-6": {
        "input": 3.00,
        "output": 15.00,
        "cache_creation": 3.75,
        "cache_read": 0.30,
    },
    "claude-haiku-4-5-20251001": {
        "input": 0.80,
        "output": 4.00,
        "cache_creation": 1.00,
        "cache_read": 0.08,
    },
}
 
# Your markup factor — 2.5x means you charge 2.5x the raw API cost
MARKUP_FACTOR = 2.5
 
def calculate_billable_cents(record: UsageRecord) -> int:
    """
    Returns the amount to bill the user in USD cents (integer).
    Using integer cents avoids floating-point rounding errors.
    """
    pricing = MODEL_PRICING.get(record.model)
    if not pricing:
        raise ValueError(f"Unknown model pricing: {record.model}")
    
    input_cost = (record.input_tokens / 1_000_000) * pricing["input"]
    output_cost = (record.output_tokens / 1_000_000) * pricing["output"]
    cache_create = (record.cache_creation_tokens / 1_000_000) * pricing["cache_creation"]
    cache_read = (record.cache_read_tokens / 1_000_000) * pricing["cache_read"]
    
    raw_cost_usd = input_cost + output_cost + cache_create + cache_read
    billable_usd = raw_cost_usd * MARKUP_FACTOR
    
    # Round up to nearest cent (never round down — you absorb the loss)
    billable_cents = math.ceil(billable_usd * 100)
    
    logger.info(
        "usage_calculated",
        extra={
            "user_id": record.user_id,
            "model": record.model,
            "input_tokens": record.input_tokens,
            "output_tokens": record.output_tokens,
            "raw_cost_usd": round(raw_cost_usd, 6),
            "billable_cents": billable_cents,
        }
    )
    
    return billable_cents
 
def call_with_tracking(
    user_id: str,
    messages: list[dict],
    model: str = "claude-sonnet-4-6",
    system: str = "",
) -> tuple[str, UsageRecord]:
    """
    Calls Claude API and returns (response_text, UsageRecord).
    The caller is responsible for persisting the record and reporting to Stripe.
    """
    import uuid
    request_id = str(uuid.uuid4())
    
    kwargs: dict = {
        "model": model,
        "max_tokens": 4096,
        "messages": messages,
    }
    if system:
        kwargs["system"] = system
    
    response = client.messages.create(**kwargs)
    usage = response.usage
    
    record = UsageRecord(
        user_id=user_id,
        model=model,
        input_tokens=usage.input_tokens,
        output_tokens=usage.output_tokens,
        cache_creation_tokens=getattr(usage, 'cache_creation_input_tokens', 0),
        cache_read_tokens=getattr(usage, 'cache_read_input_tokens', 0),
        timestamp=datetime.now(UTC),
        request_id=request_id,
    )
    
    return response.content[0].text, record

The separation of concerns here matters: call_with_tracking produces a UsageRecord but doesn't persist it or touch Stripe. That's intentional — it makes unit testing straightforward and keeps the API layer decoupled from billing infrastructure.

Step 2: Database Schema for Usage Records

CREATE TABLE api_usage_records (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id VARCHAR(255) NOT NULL,
    request_id VARCHAR(255) UNIQUE NOT NULL,  -- idempotency key
    model VARCHAR(100) NOT NULL,
    input_tokens INTEGER NOT NULL DEFAULT 0,
    output_tokens INTEGER NOT NULL DEFAULT 0,
    cache_creation_tokens INTEGER NOT NULL DEFAULT 0,
    cache_read_tokens INTEGER NOT NULL DEFAULT 0,
    billable_cents INTEGER NOT NULL DEFAULT 0,  -- integer, never float
    stripe_reported BOOLEAN NOT NULL DEFAULT FALSE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
 
CREATE INDEX idx_usage_user_month 
    ON api_usage_records(user_id, created_at);
CREATE INDEX idx_usage_unreported 
    ON api_usage_records(stripe_reported, user_id) 
    WHERE stripe_reported = FALSE;
 
-- Monthly rollup view — query this for dashboards
CREATE VIEW monthly_usage_summary AS
SELECT
    user_id,
    DATE_TRUNC('month', created_at) AS billing_month,
    SUM(input_tokens) AS total_input_tokens,
    SUM(output_tokens) AS total_output_tokens,
    SUM(billable_cents) AS total_billable_cents,
    COUNT(*) AS request_count
FROM api_usage_records
GROUP BY user_id, DATE_TRUNC('month', created_at);

Key design decisions: billable_cents is stored as an integer (cents, not dollars) to avoid floating-point errors. request_id has a UNIQUE constraint to prevent double-recording if a retry somehow succeeds twice.

Step 3: Stripe Metered Billing Integration

import stripe
 
stripe.api_key = "sk_live_..."  # Load from secret manager in production
 
def report_usage_batch(
    stripe_subscription_item_id: str,
    user_id: str,
    db_conn,
) -> int:
    """
    Reports unreported usage records to Stripe.
    Safe to call multiple times — uses FOR UPDATE SKIP LOCKED to prevent
    concurrent duplicate reporting.
    
    Returns: number of records reported
    """
    with db_conn.cursor() as cur:
        cur.execute("""
            SELECT id, billable_cents
            FROM api_usage_records
            WHERE user_id = %s
              AND stripe_reported = FALSE
            ORDER BY created_at ASC
            LIMIT 100
            FOR UPDATE SKIP LOCKED
        """, (user_id,))
        
        rows = cur.fetchall()
        if not rows:
            return 0
        
        record_ids = [r[0] for r in rows]
        total_cents = sum(r[1] for r in rows)
        
        try:
            stripe.SubscriptionItem.create_usage_record(
                stripe_subscription_item_id,
                quantity=total_cents,  # 1 unit = $0.01
                timestamp=int(datetime.now(UTC).timestamp()),
                action="increment",  # Add to existing, don't replace
            )
            
            cur.execute("""
                UPDATE api_usage_records
                SET stripe_reported = TRUE
                WHERE id = ANY(%s)
            """, (record_ids,))
            
            db_conn.commit()
            return len(rows)
            
        except stripe.error.StripeError as e:
            db_conn.rollback()
            logger.error(
                "stripe_report_failed",
                extra={"user_id": user_id, "error": str(e)}
            )
            raise  # Re-raise; the batch job will retry
 
def create_metered_price() -> dict:
    """
    One-time setup: create a Stripe product and metered price.
    Run this once during service setup.
    """
    product = stripe.Product.create(
        name="AI API Service (Usage-Based)",
        description="Pay only for what you use — billed monthly",
    )
    
    price = stripe.Price.create(
        product=product.id,
        currency="usd",
        billing_scheme="per_unit",
        unit_amount=1,  # $0.01 per unit (1 cent)
        recurring={
            "interval": "month",
            "usage_type": "metered",
            "aggregate_usage": "sum",
        },
    )
    
    return {"product_id": product.id, "price_id": price.id}

Using action="increment" rather than action="set" is important for concurrent batch jobs. With set, two parallel jobs reporting partial amounts would overwrite each other. With increment, both additions are applied correctly.

Step 4: Plan Limits and Access Control

from enum import Enum
from typing import Optional
import redis
 
r = redis.Redis(host='localhost', port=6379, db=0)
 
class PlanTier(str, Enum):
    FREE = "free"
    STARTER = "starter"    # $9.80/month
    PRO = "pro"            # $38/month
    ENTERPRISE = "enterprise"
 
# Monthly spend caps per plan (in cents)
PLAN_CAPS_CENTS = {
    PlanTier.FREE: 500,         # $5 cap
    PlanTier.STARTER: 5_000,    # $50 cap
    PlanTier.PRO: 30_000,       # $300 cap
    PlanTier.ENTERPRISE: None,  # Unlimited
}
 
PLAN_ALLOWED_MODELS = {
    PlanTier.FREE: ["claude-haiku-4-5-20251001"],
    PlanTier.STARTER: ["claude-haiku-4-5-20251001", "claude-sonnet-4-6"],
    PlanTier.PRO: ["claude-haiku-4-5-20251001", "claude-sonnet-4-6", "claude-opus-4-6"],
    PlanTier.ENTERPRISE: ["claude-haiku-4-5-20251001", "claude-sonnet-4-6", "claude-opus-4-6"],
}
 
class PlanLimitExceeded(Exception):
    def __init__(self, cap: int, current: int, plan: PlanTier):
        self.cap = cap
        self.current = current
        super().__init__(f"Monthly limit of ${cap/100:.2f} reached (current: ${current/100:.2f})")
 
def get_monthly_spend_cents(user_id: str) -> int:
    """Fast Redis lookup for current month spend. Falls back to DB on cache miss."""
    now = datetime.now(UTC)
    key = f"spend:{user_id}:{now.year}:{now.month}"
    cached = r.get(key)
    return int(cached) if cached is not None else 0  # DB fallback omitted for brevity
 
def execute_with_billing(
    user_id: str,
    plan: PlanTier,
    model: str,
    messages: list[dict],
    system: str = "",
) -> tuple[str, int, Optional[str]]:
    """
    Full billing-aware API call.
    Returns (response_text, billable_cents, upgrade_nudge_or_None).
    
    Raises PlanLimitExceeded if monthly cap is reached.
    """
    if model not in PLAN_ALLOWED_MODELS[plan]:
        raise ValueError(f"Model {model} not available on {plan.value} plan")
    
    cap = PLAN_CAPS_CENTS.get(plan)
    if cap is not None:
        current = get_monthly_spend_cents(user_id)
        if current >= cap:
            raise PlanLimitExceeded(cap, current, plan)
    
    text, record = call_with_tracking(user_id, messages, model, system)
    cents = calculate_billable_cents(record)
    
    # Update Redis cache (persist to DB asynchronously)
    now = datetime.now(UTC)
    key = f"spend:{user_id}:{now.year}:{now.month}"
    new_total = r.incrby(key, cents)
    
    # Generate upgrade nudge if approaching limit
    nudge = None
    if cap is not None:
        ratio = new_total / cap
        if ratio >= 0.95:
            nudge = f"You've used {int(ratio*100)}% of your monthly limit. Upgrade to Pro for 6x more capacity."
        elif ratio >= 0.80:
            nudge = f"Heads up: {int(ratio*100)}% of monthly limit used."
    
    return text, cents, nudge

Three Pitfalls to Avoid

Double-reporting to Stripe: The FOR UPDATE SKIP LOCKED pattern in the batch job prevents this, but only if your background workers use it. Without row-level locking, two concurrent workers can both grab the same unreported rows and report them twice.

Hardcoded exchange rates: If you're billing in a currency different from USD (your API cost currency), a hardcoded rate is a margin liability. Fetch rates from an API (exchangerate.host, openexchangerates.org) and refresh them daily at minimum.

Redis-DB drift: Redis can lose data on restart or failure. A nightly reconciliation job that rebuilds Redis spend totals from the database keeps your cache trustworthy. Without it, the cap check becomes unreliable over time.

Implementation Order

Don't try to build everything at once. Here's the order that gets you to production fastest.

Start with Steps 1 and 2 — token measurement and DB recording — without any Stripe integration. This lets you accumulate real usage data and validate your pricing assumptions. Then add Step 4's plan limits as a safety valve before you open the service to real users. Finally, connect Stripe Metered Billing in Step 3 once you have confidence in your usage data.

Measurement first, billing infrastructure second. That's the sequence that ships.