Token counting & usage

count_tokens.py# ▶ NEEDS API KEY + NETWORK. Real Anthropic SDK — will not run offline.
# pip install anthropic ; export ANTHROPIC_API_KEY=sk-ant-...
import anthropic

client = anthropic.Anthropic()

messages = [{"role": "user", "content": "Summarize the theory of relativity."}]

# Pre-flight: same args as messages.create, but no generation happens.
count = client.messages.count_tokens(
    model="claude-opus-4-8",
    system="You are a concise physics tutor.",
    messages=messages,
)
print("input tokens:", count.input_tokens)   # e.g. 24

# A pre-flight budget/window check before you spend on a real call:
MODEL_CONTEXT_WINDOW = 1_000_000      # Opus 4.8 context window
MY_INPUT_BUDGET = 50_000              # your own per-request cap
if count.input_tokens > MY_INPUT_BUDGET:
    raise SystemExit("Prompt over budget — trim it before sending.")
if count.input_tokens > MODEL_CONTEXT_WINDOW:
    raise SystemExit("Prompt exceeds the model context window.")

# Only now do you pay for generation:
resp = client.messages.create(
    model="claude-opus-4-8", max_tokens=1024,
    system="You are a concise physics tutor.", messages=messages,
)

read_usage.py# ▶ NEEDS API KEY + NETWORK. Real Anthropic SDK — will not run offline.
import anthropic

client = anthropic.Anthropic()

resp = client.messages.create(
    model="claude-opus-4-8", max_tokens=256,
    messages=[{"role": "user", "content": "Name three prime numbers."}],
)

u = resp.usage
print("input_tokens:                ", u.input_tokens)
print("output_tokens:               ", u.output_tokens)
print("cache_creation_input_tokens: ", u.cache_creation_input_tokens)
print("cache_read_input_tokens:     ", u.cache_read_input_tokens)

# Full prompt size = uncached input + cache write + cache read
full_prompt = (u.input_tokens
               + (u.cache_creation_input_tokens or 0)
               + (u.cache_read_input_tokens or 0))
print("full prompt tokens:          ", full_prompt)

# Log the request id alongside usage so you can trace a call with Anthropic.
print("request id:", resp._request_id)

cost_from_usage.py# (runs offline) — pure stdlib, no SDK, no network. `python cost_from_usage.py`
# Prices are $ per 1,000,000 tokens and DRIFT — always re-check current pricing.
PRICES = {
    # model: (input_per_mtok, output_per_mtok)
    "claude-opus-4-8":   (5.00, 25.00),
    "claude-sonnet-4-6": (3.00, 15.00),
    "claude-haiku-4-5":  (1.00,  5.00),
}

CACHE_READ_MULTIPLIER = 0.1   # cache reads bill at ~0.1x the input rate

def cost_from_usage(model, usage):
    """usage: dict with input_tokens / output_tokens and optional cache fields.
    Returns the dollar cost of one call, rounded to 6 decimals."""
    in_price, out_price = PRICES[model]
    inp   = usage.get("input_tokens", 0)
    out_  = usage.get("output_tokens", 0)
    creat = usage.get("cache_creation_input_tokens", 0)   # billed ~1.25x input
    read  = usage.get("cache_read_input_tokens", 0)       # billed ~0.1x input

    dollars = (
        inp   * in_price
        + out_  * out_price
        + creat * in_price * 1.25
        + read  * in_price * CACHE_READ_MULTIPLIER
    ) / 1_000_000
    return round(dollars, 6)

# A fake usage receipt — no API call needed.
fake_usage = {
    "input_tokens": 1_000,
    "output_tokens": 500,
    "cache_creation_input_tokens": 0,
    "cache_read_input_tokens": 8_000,   # served cheaply from cache
}
cost = cost_from_usage("claude-opus-4-8", fake_usage)
print("call cost: $" + format(cost, ".6f"))
# 1000*5 + 500*25 + 8000*5*0.1 = 5000 + 12500 + 4000 = 21500 microdollars
print("that's", 1_000 + 500 + 8_000, "tokens touched")

usage_logger.py# ▶ NEEDS API KEY + NETWORK. Real Anthropic SDK — will not run offline.
# (The cost_from_usage helper it calls IS the offline one from recipe 4.)
import anthropic
from cost_from_usage import cost_from_usage

client = anthropic.Anthropic()
LEDGER = []   # in real life: a DB row or a metrics counter per call

def tracked_create(**kwargs):
    resp = client.messages.create(**kwargs)
    u = resp.usage
    usage_dict = {
        "input_tokens": u.input_tokens,
        "output_tokens": u.output_tokens,
        "cache_creation_input_tokens": u.cache_creation_input_tokens or 0,
        "cache_read_input_tokens": u.cache_read_input_tokens or 0,
    }
    cost = cost_from_usage(kwargs["model"], usage_dict)
    LEDGER.append({
        "request_id": resp._request_id,
        "model": kwargs["model"],
        "usage": usage_dict,
        "cost_usd": cost,
    })
    return resp

tracked_create(
    model="claude-opus-4-8", max_tokens=256,
    messages=[{"role": "user", "content": "Hello"}],
)
print("calls logged:", len(LEDGER))
print("total spend: $" + format(sum(row["cost_usd"] for row in LEDGER), ".6f"))

rate_limits.py# ▶ NEEDS API KEY + NETWORK. Real Anthropic SDK — will not run offline.
import anthropic

client = anthropic.Anthropic()

# .with_raw_response exposes the HTTP headers alongside the parsed message.
raw = client.messages.with_raw_response.create(
    model="claude-opus-4-8", max_tokens=64,
    messages=[{"role": "user", "content": "ping"}],
)
h = raw.headers
print("requests remaining:", h.get("anthropic-ratelimit-requests-remaining"))
print("input tokens remaining:", h.get("anthropic-ratelimit-input-tokens-remaining"))
message = raw.parse()   # the Message you'd normally get from .create(...)
print("output tokens:", message.usage.output_tokens)

# The SDK auto-retries 429/5xx, but you can still catch a give-up:
try:
    client.messages.create(
        model="claude-opus-4-8", max_tokens=64,
        messages=[{"role": "user", "content": "ping"}],
    )
except anthropic.RateLimitError as e:
    retry_after = int(e.response.headers.get("retry-after", "60"))
    print(f"Rate limited; retry after {retry_after}s.")

paginate.py# ▶ NEEDS API KEY + NETWORK. Real Anthropic SDK — will not run offline.
import anthropic

client = anthropic.Anthropic()

# Iterating the list result auto-fetches EVERY page — not just the first.
total_batches = 0
for b in client.messages.batches.list(limit=20):
    total_batches += 1
    print(b.id, b.processing_status)
print("batches seen across all pages:", total_batches)

# Same pattern for the Files API (beta):
for f in client.beta.files.list(betas=["files-api-2025-04-14"]):
    print(f.id, f.filename, f.size_bytes)

# .data is only the CURRENT page — do NOT treat it as the full set:
first_page = client.messages.batches.list(limit=20)
print("this is ONLY page 1:", len(first_page.data), "items")
if first_page.has_next_page():
    print("...more pages exist; iterate the list to get them all")