`agentkit.core.runtime`¶

RunContext, Invoker, Budget, Quota, EventBus, Services, and the null-context helpers used in tests.

See the Runtime concept for the mental model.

L1 runtime — the one runner and the run-scoped state it threads.

RunContext = identity + meters + services. Invoker runs a unit of work through its middleware chain. Meter governs spend at two scopes: Budget per run, Quota per tenant. EventBus is the in-process pub/sub for stream-scoped events; NullCtx is the minimum-viable Ctx for single-shot agents that don't yet need a full RunContext.

RunContext `dataclass` ¶

RunContext(
    correlation_id: str,
    scope: Scope,
    budget: Budget = Budget(),
    services: Services = Services(),
    meters: list[Any] = list(),
    depth: int = 0,
    cancel: CancellationToken | None = None,
    autonomy: AutonomyLiteral = "auto",
    actor_budget: ActorBudget | None = None,
    signal_channel: SignalChannel[Any, Any] | None = None,
)

asker `property` ¶

asker: Any

The run's human transport, or None. See Services.asker.

child ¶

child(
    *,
    actor_budget: Any = _UNSET,
    signal_channel: Any = _UNSET,
) -> RunContext

Return a child context — depth+1, sharing budget/services/cancel by reference.

Kwargs are opt-in overrides for the multi-agent coordination seams. Both default to the sentinel _UNSET (propagate the parent's value); pass None to explicitly UNSET at this level of the tree, or a concrete value to override. Existing callers (ctx.child() with no args) keep the unchanged propagation-by-reference behavior.

semaphore ¶

semaphore() -> Any

This context's concurrency pool — scoped to depth.

Passing self.depth is what makes nested fan-out deadlock-free; see Budget.semaphore for why a single tree-wide pool cannot work.

check_cancelled ¶

check_cancelled() -> None

Cooperative cancellation check for patterns (no-op if no token is attached).

emit `async` ¶

emit(
    kind: ObservationKind,
    render: str = "",
    *,
    payload: Any = None,
    agent: str = "",
    parent_id: str | None = None,
) -> None

Emit a structured observation on the run's observer (no-op default; never raises).

If a tracer is attached and a span is currently open, the resulting Observation also carries a trace_context so UI consumers can deep-link from the observation to its span tree. For CRITICAL kinds (result/error) we additionally drop an observation.emitted event on the currently-open span so the trace timeline is complete without cross-referencing streams. Both trace-side effects are best-effort — they must never break the run.

Services `dataclass` ¶

Services(
    invoker: Any = None,
    store: Any = None,
    checkpointer: Any = None,
    vector: Any = None,
    trace: TracePort = NoopTrace(),
    observer: ObserverPort = NoopObserver(),
    replay: ReplayStore = NoopReplayStore(),
    metrics: MetricsPort = NoopMetrics(),
    sampler: SamplerPort = AlwaysOnSampler(),
    asker: Any = None,
)

App/process-shared collaborators, injected once and shared across runs.

Invoker ¶

Invoker(
    *,
    llm: Any,
    chat_middleware: Sequence[Any] = (),
    tool_middleware: Sequence[Any] = (),
)

Runs a unit of work through its composed (stream-shaped) middleware chain to a terminal seam stream.

stream() is the one primitive; chat() ≡ collect(stream()); invoke_tool() collects the one-item tool stream. The app builds the two chains once; patterns pick the consumption mode.

Budget `dataclass` ¶

Budget(
    max_cost_usd: Decimal | float | int | str | None = None,
    max_calls: int | None = None,
    max_depth: int = 4,
    max_concurrency: int = 8,
    spent_usd: float | None = 0.0,
    calls: int = 0,
    on_exceeded: Literal["raise", "stop"] = "raise",
    usage: Usage = Usage(),
    _spent: Decimal = Decimal(0),
    _ceiling: Decimal | None = None,
    _ceiling_src: Any = _UNSET_CEILING,
    _lock: Any = None,
    _sems: dict[int, Any] = dict(),
)

The per-run meter + the run's depth/concurrency authority (one instance per agent-tree run).

Money. max_cost_usd accepts a Decimal, float, int, or str and is normalised to an exact Decimal. A ceiling with more than :data:MONEY_SCALE decimal places raises :class:MoneyPrecisionError at construction rather than being silently rounded.

Tokens. usage accumulates the full :class:Usage — input, output, cache-read and cache-write tokens survive to the end of a multi-agent run, because Budget is shared by reference across ctx.child().

Exhaustion. on_exceeded picks the behaviour when a ceiling is crossed:

"raise" (default) — MeterExceeded from inside charge, exactly as before.
"stop" — charge returns Charge(ok=False, ...) and raises nothing. The caller acts on it; the ReAct cognition writes a suspended checkpoint and ends the run with stop_reason="budget_exhausted", so the spend is recoverable.

"raise" remains the default deliberately. Flipping it would silently change the control flow of every existing wiring — a run that used to abort would now continue past its ceiling in any caller that ignores the return value, which is a worse failure than the one being fixed. Callers opt into recoverability; a future major version can reconsider.

Known property, not a bug. _check compares spent > ceiling AFTER the work has run, so a budget is always overrun by at most one call's cost — there is no pre-flight estimate. Setting a ceiling slightly below the true limit is the mitigation; on_exceeded="stop" makes the overshoot recoverable rather than fatal.

cost_complete `property` ¶

cost_complete: bool

Whether every charge is priced; spent() otherwise returns a known lower bound.

spent ¶

spent() -> Decimal

The exact known spend. If usage is unpriced this is a lower bound — spent_usd is a float rendering of it and cannot be summed losslessly.

spent_cents ¶

spent_cents() -> Decimal

The exact spend rounded to whole cents, for invoicing.

Quantization happens HERE, at read time, never per-charge: rounding each charge to cents would round every sub-cent call to zero and undercount the entire run.

remaining_usd ¶

remaining_usd() -> float | None

Headroom as a float — the shape the tracing middleware and existing callers expect. None when no ceiling is set.

remaining ¶

remaining() -> Decimal | None

Headroom, exactly. None when no ceiling is set.

ceiling ¶

ceiling() -> Decimal | None

The normalised ceiling, exactly.

Re-derives when max_cost_usd has been ASSIGNED since the last normalisation. Budget is a mutable dataclass and budget.max_cost_usd = 10.0 is the obvious way to raise a ceiling and resume an exhausted run — it is what the spend recipe tells an operator to do. Caching the normalised value at construction and never looking again would make that assignment silently do nothing, which is a worse failure than the float ledger this class was rewritten to remove.

The re-derivation is NON-strict, unlike the one in __post_init__: this runs on a read path reached from inside charge(), and raising MoneyPrecisionError there would abort a run mid-flight — exactly the unrecoverable abort being designed out. Precision is refused at construction, where it is free to fix; a post-construction assignment is quantized.

exhausted ¶

exhausted() -> bool

Has a ceiling been crossed? A cheap synchronous read for a caller that wants to check between units of work — the ReAct loop uses it after each chat call to decide whether to checkpoint and stop.

verdict ¶

verdict() -> Charge

The current verdict without charging anything.

guard `async` ¶

guard(call: Any = None) -> Charge

Check the ceilings before the work runs.

charge `async` ¶

charge(call: Any, usage: Usage) -> Charge

Accrue usage and return the resulting verdict.

The books are updated BEFORE the ceiling is evaluated, so an over-ceiling call is still recorded — the spend really happened and dropping it would make the ledger lie.

A charge is QUANTIZED to :data:MONEY_SCALE, not refused, even though an over-precise ceiling is refused at construction. Refusing here would mean a custom pricing= callable returning full float precision aborts a run mid-flight — re-creating the unrecoverable abort this class was rewritten to remove. The ceiling is intent (refuse); a charge is a measurement (record).

Usage is frozen, so self.usage is REPLACED via its __add__, never mutated in place.

semaphore ¶

semaphore(depth: int = 0) -> Semaphore

The concurrency permit pool for one DEPTH of the agent tree.

One semaphore per depth, not one for the whole tree. A single shared semaphore deadlocks, and not hypothetically: a parent's fan-out holds its permits for the ENTIRE duration of each child run, so a nested fan-out draws from a pool its own ancestors have already drained. With max_concurrency=2, an agent dispatching two as_tool sub-agents that each dispatch their own tools hangs forever — the two outer permits are held until the inner runs finish, and the inner runs cannot start without a permit. Deeper trees hit the same wall at any cap once the ancestors' outstanding acquisitions reach it.

Keying on depth breaks the cycle structurally: an ancestor at depth d can only ever hold permits from pool d, and its children draw from pool d+1, so no acquisition can ever wait on a permit held by its own ancestor. Every nesting boundary in the framework goes through ctx.child() (as_tool, run_agents, the coordinator policies), so depth genuinely increments at each level.

The trade is an honest one: the bound is now max_concurrency PER LEVEL rather than across the whole tree, so worst-case in-flight work is max_concurrency * (max_depth + 1). Set max_concurrency with that in mind. A single tree-wide cap cannot be both deadlock-free and respected by nested acquisition — a re-entrant permit would let one level's fan-out multiply without limit, which is a weaker bound than per-level, not a stronger one.

Lazy per depth, so the semaphore binds to the running loop at first use rather than to whatever loop existed at Budget construction.

Charge `dataclass` ¶

Charge(
    ok: bool,
    reason: str = "",
    spent: Decimal = Decimal(0),
    remaining: Decimal | None = None,
    calls: int = 0,
    cost_complete: bool = True,
    usage: Usage = Usage(),
)

The verdict a meter returns from guard/charge.

A value, not an exception, so the caller decides what happens next. The tool-loop cognition uses ok=False to write a checkpoint and stop cleanly; a batch driver might use it to switch to a cheaper model; a fire-and-forget caller ignores it and lets on_exceeded="raise" do the old thing.

usage is the meter's cumulative token totals, so a caller reading the verdict never has to re-aggregate what the framework already summed.

raise_if_exceeded ¶

raise_if_exceeded() -> Charge

Convert the verdict back into the exception, for a caller who wants the old control flow at one specific site. Returns self when ok, so it chains: (await budget.charge(...)).raise_if_exceeded().

Meter ¶

Bases: Protocol

Guard a ceiling, then charge usage.

Both methods return a :class:Charge. Implementations written against the older -> None signature still satisfy this at runtime and still work through the meter middleware, which treats None as "no verdict offered" — see middlewares/meter.py. New implementations should return a verdict so a caller can degrade instead of only ever catching.

MeterExceeded ¶

Bases: RuntimeError

A ceiling (run cost/calls or tenant RPM/TPM/$) was crossed → the caller degrades gracefully.

MoneyPrecisionError ¶

Bases: ValueError

A monetary ceiling was given more precision than the ledger keeps.

Raised at CONSTRUCTION only. Silently rounding a ceiling changes what the operator asked for; refusing tells them at wiring time, when it is free to fix. Charges are handled the other way round (quantized, not refused) — see :meth:Budget.charge, where raising mid-run would re-create the exact unrecoverable abort this module exists to remove.

Quota `dataclass` ¶

Quota(
    max_rpm: int | None = None,
    max_tpm: int | None = None,
    max_usd: Decimal | float | int | str | None = None,
    window: float = 60.0,
    clock: Callable[[], float] = monotonic,
    on_exceeded: Literal["raise", "stop"] = "raise",
    _ceiling: Decimal | None = None,
    _last_sweep: float = float("-inf"),
    _reqs: dict[str, list[float]] = (
        lambda: defaultdict(list)
    )(),
    _charges: dict[
        str, list[tuple[float, int, Decimal | None]]
    ] = (lambda: defaultdict(list))(),
    _lock: Any = None,
    _ceiling_src: Any = _UNSET_CEILING,
)

The per-tenant meter — rolling RPM/TPM/$ windows keyed by scope.key(), independent of Budget. The noisy-neighbor guard and chargeback source. In-memory ref; prod swaps a Redis-backed Meter.

Unlike :class:Budget, Quota enforces on guard (before the work) rather than on charge (after it), so its charge has nothing to refuse — it returns an ok verdict for protocol symmetry. guard honours the same on_exceeded switch so a caller can drive both meters with one control flow.

ceiling ¶

ceiling() -> Decimal | None

The normalised per-window ceiling. Re-derives on a post-construction assignment to max_usd — see Budget.ceiling for the rationale.

spent_in_window ¶

spent_in_window(key: str) -> Decimal

Exact known spend for one tenant inside the current window. Unpriced charges make this a lower bound; Charge.cost_complete exposes that fact.

NullCtx ¶

NullCtx(scope: Scope | None = None)

Minimum-viable Ctx. Structurally satisfies the Ctx Protocol in agentkit.core.contracts.protocols; every operation is a no-op or raises if it would require real behavior (only the invoker raises).

Use this when the calling code touches a subset of the Ctx surface — e.g., RequestBuilder.build() only reads .trace.span(...). When the caller later threads a real RunContext, behavior at the call site is unchanged thanks to structural typing.

check_cancelled ¶

check_cancelled() -> None

No cancellation token attached — always a no-op.

semaphore ¶

semaphore() -> _NullSemaphore

Return an always-available semaphore-like. gather_bounded uses async with, which this satisfies without serialising.

child ¶

child() -> NullCtx

No per-task isolation in a null context — the same instance is reusable as its own child. Real RunContext.child() bumps depth and shares services; here there is nothing to bump or share.

emit `async` ¶

emit(
    kind: ObservationKind,
    render: str = "",
    *,
    payload: Any = None,
    agent: str = "",
    parent_id: str | None = None,
) -> None

No observer attached — drop the observation on the floor. Matches RunContext.emit's contract of never raising into the run.

CallRuntime ¶

CallRuntime(
    *,
    adapters: Mapping[str, ModeAdapter] | None = None,
    pricing: PriceTable | None = None,
    capabilities: CapabilityTable = DEFAULT_CAPABILITIES,
    retry: RetryPolicy = _DEFAULT_RETRY,
    capture_text: bool = False,
    native_tool_policy: NativeToolPolicy = WARN,
    unpriced_policy: UnpricedPolicy = REFUSE,
    max_concurrent_cli: int = 8,
    acquire: Callable[
        [Mode], AbstractAsyncContextManager[Any]
    ]
    | None = None,
    argument_masker: Callable[[str, Any], Any]
    | None = None,
    result_masker: Callable[[str, str], str] | None = None,
)

Owns adapter resources and the common contract.

Use as an async context manager when CLI sessions must survive multiple calls. Inject adapters for offline tests and prices/capabilities for hosts.

ask `async` ¶

ask(
    binding: Binding,
    prompt: ComposedPrompt | str,
    output: OutputSpec | type | None = None,
    **kwargs: Any,
) -> Answer[Any]

A one-shot call, with one shared bounded correction turn.

to_money ¶

to_money(
    value: Decimal | float | int | str,
    *,
    strict: bool = False,
) -> Decimal

Coerce a monetary amount to an exact Decimal at :data:MONEY_SCALE.

Floats go through str() first: Decimal(0.01) is 0.01000000000000000020816681711721685... (the binary value), whereas Decimal(str(0.01)) is exactly 0.01 — the number the caller meant.

strict=True refuses input carrying more precision than the ledger keeps instead of quantizing it.

ask `async` ¶

ask(
    binding: Binding,
    prompt: ComposedPrompt | str,
    output: Any = None,
    **kwargs: Any,
) -> Answer[Any]

Convenience one-shot call; use CallRuntime for long-lived sessions.

run `async` ¶

run(
    binding: Binding,
    prompt: ComposedPrompt | str,
    tools: Any = None,
    **kwargs: Any,
) -> Answer[Any]

Convenience agent loop with shared tool handling.

agentkit.core.runtime¶

RunContext dataclass ¶

asker property ¶

child ¶

semaphore ¶

check_cancelled ¶

emit async ¶

Services dataclass ¶

Invoker ¶

Budget dataclass ¶

cost_complete property ¶

spent ¶

spent_cents ¶

remaining_usd ¶

remaining ¶

ceiling ¶

exhausted ¶

verdict ¶

guard async ¶

charge async ¶

semaphore ¶

Charge dataclass ¶

raise_if_exceeded ¶

Meter ¶

MeterExceeded ¶

MoneyPrecisionError ¶

Quota dataclass ¶

ceiling ¶

spent_in_window ¶

NullCtx ¶

check_cancelled ¶

semaphore ¶

child ¶

emit async ¶

CallRuntime ¶

ask async ¶

to_money ¶

ask async ¶

run async ¶

`agentkit.core.runtime`¶

RunContext `dataclass` ¶

asker `property` ¶

emit `async` ¶

Services `dataclass` ¶

Budget `dataclass` ¶

cost_complete `property` ¶

guard `async` ¶

charge `async` ¶

Charge `dataclass` ¶

Quota `dataclass` ¶

emit `async` ¶

ask `async` ¶

ask `async` ¶

run `async` ¶