noburn

Hard LLM spend enforcement for Python — pre-call budget checks and per-user metering.

Check a planned LLM call against project-, per-user-, and per-run budget caps before you make it, then record what it actually cost.

Install

pip install noburn

Requires Python ≥ 3.9. No third-party dependencies. Ships type hints (py.typed).

Quick start

import os
from noburn import NoburnGuard

guard = NoburnGuard(
    api_key=os.environ["NOBURN_API_KEY"],
    project_id=os.environ["NOBURN_PROJECT_ID"],
    budget_cap_usd=10.0,     # optional local fallback cap
    on_error="allow",        # "allow" (default, fail-open) | "block" (fail-closed)
)

decision = guard.check(
    model="gpt-4o",
    estimated_tokens_in=1000,
    estimated_tokens_out=300,
    end_user_id="user_123",  # optional — enforces that user's cap
)

if decision.blocked:
    raise RuntimeError(f"Blocked by noburn: {decision.block_reason}")

# ... make your LLM call ...

guard.record(
    model="gpt-4o",
    tokens_in=980,
    tokens_out=290,
    cost_usd=0.005,
    was_blocked=False,
    end_user_id="user_123",
)

record() is fire-and-forget — it returns immediately (delivery happens on a background daemon thread) and never raises, so it adds no latency to your call path.

Usage quotas

Give each plan an allowance ("free gets 20 requests a day") in the dashboard under Plan quotas, then pass the user's plan in context["plan"]:

decision = guard.check(
    model="gpt-4o",
    estimated_tokens_in=1000,
    estimated_tokens_out=300,
    end_user_id="user_123",
    context={"plan": "free"},
)

if decision.block_reason == "quota_exceeded" and decision.quota and decision.quota.upgrade_suggested:
    ...  # show your upgrade prompt

# Show the allowance before the user tries anything:
q = guard.remaining(end_user_id="user_123", plan="free")
if q:
    print(f"{q.remaining:g} of {q.limit:g} left, resets {q.resets_at}")

remaining() returns None when no quota applies or on any network error, and only raises NoburnAuthError. Windows are UTC (day 00:00, week Monday 00:00, month the 1st). A per-user cap set in the dashboard overrides the plan quota.

Per-customer budgets

Set each customer's cap from your backend: on signup, upgrade, top-up or cancel. Turn on Can manage budgets for the key first (project → SDK key).

guard.set_end_user_budget("ws_123", 20)                                 # $20 / month
guard.set_end_user_budget("ws_123", 5, window="total")                  # never resets
guard.set_end_user_budget("ws_123", 5, window="total", reset_spend=True)  # fresh pool
guard.set_end_user_budgets([{"end_user_id": "ws_1", "limit": 5}])       # up to 500
b = guard.get_end_user_budget("ws_123")   # EndUserBudget with used / remaining, or None
guard.delete_end_user_budget("ws_123")

Unlike check(), these raise (NoburnApiError with status and code), so a cap you meant to set never silently goes missing. Changes apply on the customer's next check(). Docs: https://noburn.dev/docs/per-customer-budgets

Agent runs

Bound a single agent invocation with its own cap:

run = guard.start_run(budget_cap_usd=0.5)

decision = guard.check(
    model="gpt-4o",
    estimated_tokens_in=1000,
    estimated_tokens_out=300,
    run_id=run.run_id,
)

guard.record(model="gpt-4o", tokens_in=980, tokens_out=290,
             cost_usd=0.005, was_blocked=False, run_id=run.run_id)

summary = guard.end_run(run.run_id)  # EndRunResult(status=..., spend_usd=..., ...)

Stopping a run

Stop a runaway run from outside the agent: a supervisor, a spend.spike webhook handler, or the dashboard's Stop run button. Every later check() for that run returns blocked=True, block_reason="run_stopped".

stopped = guard.stop_run(run_id, reason="spend spike")
# StopRunResult(run_id, status="stopped", stopped_now, spend_usd, stopped_at, stopped_reason, ...) | None

Stopping is idempotent (stopped_now=False when it was already stopped). It returns None on a network error or an unknown run, and raises only NoburnAuthError. If this guard has seen a run stopped, check() keeps blocking it even while the noburn API is unreachable.

Profit per customer

Send what each end user pays you, and the project's Customers page shows revenue, AI cost and margin per customer, money-losing customers first. Use the exact end_user_id you pass to check() / record() (it is not trimmed).

guard.record_revenue(
    end_user_id="user_abc123",
    amount_usd=49.0,                     # negative for refunds
    occurred_at="2026-10-01T09:30:00Z",  # optional, defaults to now
    external_id=invoice["id"],           # optional idempotency key (e.g. Stripe invoice id)
)

Like record(), it is fire-and-forget: queued and delivered by the background worker with retry. A repeated external_id is ignored server-side, so webhook retries never double-count. It never raises except NoburnAuthError, raised once a previous delivery was rejected for an invalid key. Invalid input, and revenue the server rejects with a 4xx, are dropped and counted in dropped_event_count.

Before a short-lived process exits, call guard.flush(), or guard.flush(timeout=5.0) to cap the wait. With a timeout it returns without raising, even if events are still queued. See the profit-per-customer guide for a Stripe invoice.paid webhook example.

Behavior

• check() never raises except NoburnAuthError (bad key). It asks the server for an authoritative decision; if the server is unreachable it falls back to a local in-memory check.
• Fast path (default). When the server allows a call with room to spare, it leases the SDK a slice of what's left (a tenth, at most $5 / 50 calls) for 10 seconds. Calls with the same model, end user, run and context are then decided locally with no network call (decided_locally=True), and the lease renews in a background thread. Leases count as spent in every check, and none is granted near a cap, so caps stay hard; a stopped run or lowered cap reaches a lease holder within 10 seconds. fast_path=False checks every call on the server.
• on_error controls that fallback when nothing local blocks the call: "allow" (default) fails open so a noburn outage never breaks your app; "block" fails closed (block_reason="noburn_unreachable").
• start_run() raises on network/auth error (NoburnTimeoutError on timeout) — a run must be registered before you proceed.

Note: check()/start_run() use blocking HTTP. In async apps, run them in a thread executor (e.g. await asyncio.to_thread(guard.check, ...)).

Loop detection

The LangChain handler stops an agent going in circles (on by default), judging from the AI tool calls and ToolMessages in the messages. When the model asks for a call its history refuses (the same call returning the same result a 3rd time, failing the same way a 4th time in a row, or a 4th A→B cycle), the answer is refused before the tool runs with NoburnBlockedError (block_reason="loop_detected", .loop has the pattern), and the run is stopped. on_loop gets a warn event one step earlier; loop_nudge(event) turns it into a note you can add to the prompt yourself. Steps are hashed locally; no prompt text is sent. Tune it with loop_detection={"exempt_tools": ["job_status"]}, turn it off with loop_detection=False, or use LoopDetector directly around your own tools.

Framework integrations

The integrations call check() before each model call and record() after it. Your framework still makes the provider call, and no prompt or completion text is sent to noburn. A blocked call raises NoburnBlockedError (with block_reason) before the request goes out. If noburn is unreachable, the call goes through (fail-open). import noburn never imports a framework.

LangChain

pip install "noburn[langchain]"

from langchain_openai import ChatOpenAI
from noburn.integrations.langchain import NoburnCallbackHandler

llm = ChatOpenAI(
    model="gpt-4o-mini",
    callbacks=[NoburnCallbackHandler(guard, end_user_id="user_123")],
)

CrewAI

Requires crewai>=1.15, which needs Python 3.10 or newer: pip install "noburn[crewai]".

from noburn.integrations.crewai import NoburnCrewGuard

crew_guard = NoburnCrewGuard(guard, budget_cap_usd=0.50)  # one run cap for the whole crew
result = crew_guard.kickoff(crew, inputs={"topic": "AI agents"})
print(crew_guard.summary)

The guard uses CrewAI's execution-scoped model-call hooks. Every LLM call by every agent is checked before it reaches the provider. A block stops the crew without task retries, and every allowed call, native tool-call turns included, is recorded once with its actual usage. Use one NoburnCrewGuard per concurrent crew.

Full options: https://noburn.dev/docs/integrations

License

MIT