Budgets

Cap what a run or a conversation may spend, and how many calls it may make. The ledger prices every model and paid-API call; BudgetHook reads it before each call.

flowchart LR
  C["tool or model call"] --> B{"BudgetHook<br/>spent ≥ max_usd?"}
  B -- "no" --> RUN["run"] --> L["Ledger.record<br/>priced from the table"]
  B -- "yes, stop" --> E["BudgetExceeded"]
  B -- "yes, warn" --> W["budget.warning event"] --> RUN
  L -.->|"total_usd(run / thread)"| B

1. Prices

# Illustrative prices (USD per 1M tokens, USD per unit); load your own table in real code.
PRICES = Prices.load(
    {
        "version": "demo",
        "models": {"house-llm": {"provider": "host", "input": 2.0, "output": 8.0}},
        "units": {"acme:geocode": {"provider": "acme", "per_unit": 0.05}},
    }
)

Prices.default() ships OpenAI and Gemini list prices; add your own with merged() or load() (Cost). An unknown model costs None (counted as 0) and logs one WARNING, so a missing price never blocks a call, but it also never counts toward a budget.

2. Wire the ledger and the hook

ledger, events = Ledger(prices=PRICES), EventLog()
executor = Executor(
    registry,
    ledger=ledger,
    sink=events,
    hooks=[BudgetHook(0.50)],  # scope="run": each run may spend $0.50
    limits=RunLimits(max_calls=10),  # and make at most 10 tool calls
)
house = CallableModel(house_llm, provider="host", model="house-llm")
llm = executor.model(house)  # hooks, ledger and events around every model call

executor.model(port) wraps any model port with the executor's hooks, ledger and events, so each model call is one priced LedgerEntry(kind="llm").

3. Per run

ask = LLMRequest(
    messages=(Message(role="user", content="Summarise the returns policy"),), stage="answer"
)
async with executor.run_scope() as ctx:
    answered = 0
    try:
        while True:
            await llm.bind(ctx).generate(ask)  # $0.20 each
            answered += 1
    except BudgetExceeded as exc:  # raised before the call that would start over budget
        print(answered, exc.scope, exc.limit_usd, round(exc.spent_usd, 2))
    llm_calls = ledger.summary(ctx.run_id).by_kind["llm"]
    print(round(ledger.total_usd(ctx.run_id), 2), llm_calls.calls, llm_calls.errors)

The hook checks before each call, so a run can end slightly over the cap: here three calls of $0.20 run, the fourth is stopped at $0.60. The stopped call is recorded as a failed entry at $0. Set max_usd below the hard limit by one call's cost if you must never cross it.

4. Per conversation

chat = executor.replace(hooks=[BudgetHook(0.25, scope="thread", on_exceed="warn")])
for run in ("run_a", "run_b"):  # two turns of one conversation
    ctx = chat.context(run_id=run, thread_id="chat-7")
    await chat.run("geocode", {"address": "Paris"}, ctx)  # $0.05 each
    await (
        chat.model(house)
        .bind(ctx)
        .generate(  # $0.20 each
            LLMRequest(messages=(Message(role="user", content="hi"),))
        )
    )
print(round(ledger.total_usd(thread_id="chat-7"), 2), len(events.of_type("budget.warning")))

scope	Reads	Use for
"run" (default)	total_usd(ctx.run_id)	one agent turn or job
"thread"	total_usd(thread_id=ctx.thread_id)	a chat across turns; an empty thread_id falls back to run
"global"	total_usd()	everything this ledger holds (one process)

on_exceed	At or over the cap
"stop" (default)	raises BudgetExceeded(limit_usd, spent_usd, scope=, run_id=)
"warn"	logs once per run and emits budget.warning; the call runs

Several hooks may stack: [BudgetHook(0.50), BudgetHook(5.00, scope="thread")].

5. Paid APIs inside a tool

@registry.tool
def geocode(address: str, ctx: ToolContext) -> str:
    """Geocode an address with a paid API."""
    if ctx.ledger is not None:  # record what the vendor bills: one unit per lookup
        bill = LedgerEntry(
            run_id=ctx.run_id,
            call_id=ctx.call_id,
            entry_id=f"{ctx.call_id}:geocode",  # not the call id: that is the tool's own entry
            thread_id=ctx.thread_id,  # so thread budgets see it
            kind="api",
            provider="acme",
            model="geocode",
            units=1,
        )
        ctx.ledger.record(bill)
    return f"{address}: 18.52N 73.86E"


@registry.tool(once_per_run=True)
def send_summary(text: str) -> str:
    """Email the run's summary (at most once per run)."""
    return "sent"

Field	Rule
kind="api", provider, model, units	priced as units["provider:model"].per_unit × units
entry_id	give it its own id: the executor already records the tool's entry under the call id, and the same (run_id, entry_id) replaces an entry
thread_id=ctx.thread_id	without it a thread budget does not see the cost

6. Call-count caps

async with executor.run_scope() as ctx:
    print(await executor.run("send_summary", {"text": "done"}, ctx))
    again = await executor.run_call(ToolCall(name="send_summary", arguments={"text": "x"}), ctx)
    print(again.error.kind if again.error else "")  # once_per_run
    geocoded = 0
    try:
        while True:
            await executor.run("geocode", {"address": "Berlin"}, ctx)
            geocoded += 1
    except RunLimitExceeded:
        print("max_calls reached after", geocoded, "geocodes")  # plus the one summary

once_per_run and RunLimits(max_calls=) count a call only once it will execute (valid arguments, approval granted). A replay of the same call id, tool and arguments is not a new call. Both raise RunLimitExceeded (kind run-limit-exceeded).

Output

3 run 0.5 0.6
0.6 4 1
0.5 1
sent
run-limit-exceeded
max_calls reached after 9 geocodes

Notes

Note	Detail
No ledger, no budget	BudgetHook without a ledger does nothing and warns once
Cache hits are free	a response- or tool-cache hit is recorded at $0 (Caching)
Several processes, users, teams, days	the in-memory Ledger counts one process; for caps per user or team, per day or month, shared by every worker, see Team budgets
Reports	ledger.summary(run_id): totals by stage and kind, cache hit rates, parallelism

Files

File	Does
examples/howto_budgets.py	prices, a paid tool, per-run and per-thread caps, call caps
tests/docs/test_howto_examples.py	runs it offline and checks the output