Libraries/tokenguard
tokenguardBudget

An over-budget call
never runs.

Cap spend before a call executes — block, downgrade, or degrade gracefully — and get cost attribution per feature and per user, for free. No proxy, no server: a decorator.

$ pip install cendor-tokenguardWatch it block, live ↓
support_bot.pyenforced BEFORE the call runs
from cendor.tokenguard import budget, track, report

@budget(usd=0.50, on_exceed="block")     # raises BEFORE an over-budget call
def answer(q):
    with track(feature="support", user_id="alice"):
        return client.chat.completions.create(model="gpt-4o", ...)

report(group_by=["feature", "user_id"])   # spend per tag — for free
report().assert_under(usd=0.05, feature="search")   # cost as a test assertion
Watch

See it in action.

tokenguard — the $400 Friday night
tokenguard — the $400 Friday night
How it enforces

Block, downgrade, truncate, raise — or your own policy.

@budget(usd=…, on_exceed=…) wraps any function. Pre-flight modes project the cost of the call about to run and act before a token is spent; post-flight modes account exactly and act on the next call.

"block"pre-flight

Projects the cost of the call about to run; raises BudgetExceeded before a token is spent.

policy_block.pytokenguard
@budget(usd=0.50, on_exceed="block")
def answer(q): ...
# over the cap → BudgetExceeded, raised BEFORE the call ($0 spent)
"downgrade"pre-flight

Reroutes to a cheaper model (e.g. gpt-4o → gpt-4o-mini) before the call.

policy_downgrade.pytokenguard
@budget(usd=0.50, on_exceed="downgrade")
def answer(q): ...
# over the cap → rerouted gpt-4o → gpt-4o-mini, pre-flight
"truncate"post-flight

Degrades gracefully once the cap is crossed.

policy_truncate.pytokenguard
@budget(usd=0.50, on_exceed="truncate")
def answer(q): ...
# past the cap → the work degrades instead of stopping
"raise"post-flight

Exact accounting; stops the NEXT call in a runaway loop.

policy_raise.pytokenguard
@budget(usd=0.50, on_exceed="raise")
def answer(q): ...
# exact accounting — the NEXT call in the loop is the one that stops
any callableeither

Your policy.

policy_custom.pytokenguard
def my_policy(ctx): ...

@budget(usd=0.50, on_exceed=my_policy)
def answer(q): ...
policy_block.pytokenguard
@budget(usd=0.50, on_exceed="block")
def answer(q): ...
# over the cap → BudgetExceeded, raised BEFORE the call ($0 spent)

Budgets nest — the tightest applicable cap wins; an inner downgrade never masks an outer hard cap. Attribution rides contextvars, so tags survive nested and async calls.

Try it

Watch it block, live.

A simulated agent loop under @budget(usd=0.50). Each iteration is one gpt-4o call costing $0.09. Run it past the cap and watch the mode decide what happens.

▸ each click simulates one gpt-4o call ($0.09) · enforcement happens before the call runs

on_exceed
spent $0.00 / $0.50 · 0 calls
Attribution for free

Every dollar gets a feature and a user.

Wrap work in track(feature=…, user_id=…) and every call inside carries those tags — attribution rides contextvars, so tags survive nested and async calls.

feature
user_id
calls
spend
support
alice
8
$0.31
search
bob
3
$0.04

unpriced_calls: 0

Every instrumented call is priced and tagged from ambient context — report(group_by=…) aggregates it; assert_under() turns cost into a CI assertion.

Durable spend, off the hot path. Point use_sink(…) at a SQLite / OTel / file sink to persist every priced row. The event bus runs subscribers inline, so a durable sink would add its write latency to every model call — on a long or high-throughput run, a latency cliff. Wrap it: use_sink(QueueSink(SQLiteSink("spend.db"))) moves that I/O to a background worker — write() enqueues and returns, a single worker drains in order, and flush() / close() block for durability at a checkpoint or shutdown. Durable logging, no per-call cost.
Honest limits

Where the edges are — by design.

Pre-flight is projection-based.

block and downgrade project cost from offline token estimates plus an output reserve — they're approximate. Post-flight raise is exact, but stops the NEXT call, not the one that breached.

Streams enforce on consumption.

A streamed call is accounted when the stream is drained, not when it starts; a loop launching many streams eagerly can outrun post-flight modes (pre-flight still applies).

Budgets are context-scoped.

contextvars flow through async naturally; a plain threading.Thread starts fresh — use copy_context() for worker threads.

Unpriced models can't hit a USD cap.

An unknown model warns (UnpricedModelWarning) and is counted in unpriced_calls; configure(on_unpriced="raise") makes block-mode reject such calls outright.

Get started

One decorator between you and a $400 surprise.

$ pip install cendor-tokenguard

tokenguard docs → · cost-assertion cookbook → · composes with the Cendor stack · see it live in Cendor Monitor →

Want the whole loop?

Compose tokenguard and the other six libraries into a governed agent — Agent, run, sessions, RAG, multi-agent — with the cendor-sdk. Same objects, re-exported; no migration.

Wiring an AI assistant?

npx @cendor/init / uvx cendor-init writes the rules files for you. Point it at the call-shape trap sheet, or connect the MCP server in agent mode.