`agentkit.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,
)

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.

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(),
)

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

EventBus ¶

EventBus(*, ring_size: int = 1024, queue_size: int = 256)

Bases: Generic[E]

In-process pub/sub bus, generic over the event type E.

One instance per app/host. Producers call publish; consumers call subscribe and iterate. close_stream ends every in-flight subscriber's loop on terminal transitions / shutdown.

Create with EventBus[MyEventType]() to type the publish / subscribe surface. The mechanism itself is fully content-agnostic.

publish `async` ¶

publish(stream_id: str, event: E) -> VersionedEvent[E]

Stamp a version, append to the stream's ring buffer, and deliver to every live subscriber's queue. Returns the versioned event so callers can log or expose the version.

Drops on a per-subscriber basis: a full subscriber queue causes that subscriber's event to be dropped (logged), not the publish itself. The publisher never blocks waiting for a slow downstream.

subscribe ¶

subscribe(
    stream_id: str,
    *,
    name: str = "anon",
    from_version: int = 0,
    replay: bool = True,
) -> AsyncIterator[VersionedEvent[E]]

Return an async iterator over events for stream_id.

Replay semantics:

replay=True (default) — yields the replay slice first (events in the ring with version >= from_version), then switches to live events. from_version=0 replays the entire available ring; from_version=K resumes from version K.
replay=False — skips the replay slice entirely and yields only events published after the subscriber attaches. The from_version argument is ignored in this mode. Use this for "live-only" subscribers (e.g. a fresh WebSocket fan-out that has its own baseline snapshot and doesn't want historical deltas behind it).

The returned iterator's finally block unsubscribes; consumers just async for ve in bus.subscribe(...) and break out when done.

Pass name something identifying ("checkpointer", "ws:") — drop logs key off it.

close_stream `async` ¶

close_stream(stream_id: str) -> None

Tear down a stream's channel: push the close sentinel into every subscriber queue (their async for loops exit) and drop the channel. Called from the application's terminal phase and from shutdown.

Idempotent — closing an already-closed (or never-opened) stream is a no-op. Subsequent publish calls create a fresh channel, which is harmless if the application is genuinely past the terminal phase (it shouldn't be).

shutdown `async` ¶

shutdown() -> None

Close every channel. Called from the application's lifespan exit so any consumer tasks wrapped around subscribe() unwind cleanly before the host process terminates.

VersionedEvent `dataclass` ¶

VersionedEvent(version: int, event: E)

Bases: Generic[E]

An event with a monotonic version stamped by the bus at publish time. Subscribers use the version to dedupe + resume from a known point — the version is internal bus bookkeeping and does not cross any wire by itself (callers unwrap .event before sending).

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: float | None = None,
    max_calls: int | None = None,
    max_depth: int = 4,
    max_concurrency: int = 8,
    spent_usd: float = 0.0,
    calls: int = 0,
    _lock: Any = None,
    _sem: Any = None,
)

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

MeterExceeded ¶

Bases: RuntimeError

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

Quota `dataclass` ¶

Quota(
    max_rpm: int | None = None,
    max_tpm: int | None = None,
    max_usd: float | None = None,
    window: float = 60.0,
    clock: Callable[[], float] = monotonic,
    _reqs: dict[str, list[float]] = (
        lambda: defaultdict(list)
    )(),
    _charges: dict[str, list[tuple[float, int, float]]] = (
        lambda: defaultdict(list)
    )(),
    _lock: Any = None,
)

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.

NullCtx ¶

NullCtx(scope: Scope | None = None)

Minimum-viable Ctx. Structurally satisfies the Ctx Protocol in agentkit.kernel.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: str,
    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.

agentkit.runtime¶

RunContext dataclass ¶

child ¶

check_cancelled ¶

emit async ¶

Services dataclass ¶

EventBus ¶

publish async ¶

subscribe ¶

close_stream async ¶

shutdown async ¶

VersionedEvent dataclass ¶

Invoker ¶

Budget dataclass ¶

MeterExceeded ¶

Quota dataclass ¶

NullCtx ¶

check_cancelled ¶

semaphore ¶

child ¶

emit async ¶

`agentkit.runtime`¶

RunContext `dataclass` ¶

emit `async` ¶

Services `dataclass` ¶

publish `async` ¶

close_stream `async` ¶

shutdown `async` ¶

VersionedEvent `dataclass` ¶

Budget `dataclass` ¶

Quota `dataclass` ¶

emit `async` ¶