Python API

Look up public interfaces by package. For usage, see the SDK, Tape, Events, and Turn.

• bub.core: Agent loop, model adaptation, tools, Tape, and events. Does not read YAML / BUB_*.
• Application layer: BubFramework, BubAgent, channels, and configuration. Requires installing the app optional dependencies.

republic.… names in signatures come from the external Republic dependency, not the Bub package; this page does not expand those types.

import bub.core as core
from bub import BubFramework

Importing only bub.core does not load the CLI, settings, or plugins.

Core

Agent

from bub.core import Agent

Agent(
    *,
    model=None,
    store=None,
    tools=(),
    system_prompt=(),
    context=None,
    sidecars=(),
    max_steps=sys.maxsize,
    max_tool_concurrency=None,
    name="agent",
    command_prefix=None,
)

Parameter	Default	Description
model	None	Default model: a Republic chat model (republic.ChatModel) or ModelClient; can also be supplied for each run
store	New InMemoryTapeStore	Synchronous TapeStore or AsyncTapeStore
tools	()	Initial Tool objects and provider-executed tools such as republic.tools.WebSearch(); later entries override earlier entries with the same name
system_prompt	()	Text, PromptSection, or SystemPrompt; rendered once per Run before the first model Step, then reused by later Steps
context	default_tape_context()	Strategy for selecting Tape entries as model messages
sidecars	()	Sidecars mounted with the Tape
max_steps	sys.maxsize	Maximum model Steps per Run
max_tool_concurrency	None	Tool concurrency limit within one model response; 1 executes tools sequentially
name	"agent"	Name of the Agent trace span
command_prefix	None	None disables commands; otherwise it must be non-empty and contain no whitespace, and once set it allows calling tools directly

agent.tools stores Tool objects by name, while agent.builtin_tools stores provider tools. max_tool_concurrency does not coordinate concurrency between different Runs; the host must do that. Construction attaches TapeRecorder to events. See Run an Agent in Python for a complete example.

Main entry points:

Method	Description
run(...) / start(...)	Executes one Run on a named Tape; run closes the stream on exit, while start requires the caller to call aclose()
tape(name, *, state=None)	Reads and writes a Tape in storage directly
session(name, *, state=None, merge_back=True)	Opens a branch and merges it on exit
run_on(tape, prompt, ...)	Runs on an already opened Tape
run_once(...)	Executes only one model Step and its tool calls
run_command(tape, line)	Executes a command directly without calling the model
prepare_tools(tape, prompt)	Subclasses can override this to choose tools once per Run; later Steps reuse them, and calls outside the prepared set use a fresh prepare_tools(tape, None) lookup

run / start share these parameters: prompt, model, options (external republic.ChatOptions), required tape name, state, merge_back, and metadata. See forks and merges for how a Run’s Tape fork merges on close.

Model

from bub.core import Model, ModelClient, ChatModelClient

Name	Description
Model	republic.ChatModel | ModelClient
ModelClient	Bub protocol: spec, max_tokens, open_stream(...)
ChatModelClient	Adapts republic.ChatModel to ModelClient; the caller owns the model lifecycle

Tools

from bub.core import Tool, ToolContext, tool

Tool fields: name, handler, description, parameters, context, exposure (auto / direct / code / command), renderer, output_schema, and deferred.

@tool / tool(...) constructs a Tool from a callable; @tool(model=...) uses a Pydantic parameter model. The decorator does not register globally; provide tools through Agent(tools=[...]) or provide_tools. ToolContext contains tape, run_id, state, code_mode, command, and events.

from bub.core.tools import content_result, result_content

content_result(content) constructs a tool result containing text and media, and result_content(result) converts it back into model content. Ordinary results are converted to text by Tool.render(). See Custom tools and code mode for usage, and media for how such results are recorded on the Tape.

System prompt

from bub.core import SystemPrompt, PromptSection, PromptContext

• PromptSection: Fixed text, or (PromptContext) -> str | None (may be async)
• SystemPrompt: Ordered sections; add / remove / render
• PromptContext: Current Run’s agent, tape, prompt, tools, and model

Prompt content

from bub.prompt import UserContent

UserContent is a Bub-defined alias for text or media. See the prompt contract for media forms and conversion to legacy hook inputs.

Tape and store

from bub.core import Tape, TapeContext, default_tape_context, session_tape_name
from bub.core.store import (
    AsyncTapeStore,
    AsyncTapeStoreAdapter,
    FileTapeStore,
    InMemoryQueryMixin,
    InMemoryTapeStore,
    TapeQuery,
    TapeStore,
    is_async_tape_store,
)
from bub.core.tape import AnchorSummary, TapeEntry, TapeInfo

A Tape is a read-write view of a named Tape, made of three parts: store (AsyncTapeStore, which persists entries), context (TapeContext, which decides the entries read_messages() selects), and sidecars (sibling tapes mounted on the Tape). When constructing one directly, bind a name with scoped(name) first:

tape = Tape(
    store=AsyncTapeStoreAdapter(InMemoryTapeStore()),
    context=default_tape_context(),
).scoped("demo")

Member	Description
name	The Tape’s name; accessing it before scoped raises ValueError
query()	Returns a TapeQuery bound to this Tape and store
read_messages()	Selects entries with TapeContext and converts them to model messages; entries with meta["context"] is False are skipped
handoff(name=..., state=...)	Writes an anchor and a handoff event, moving the default context start forward
reset(*, state=None, keep_sidecars=())	Clears the Tape and writes a new session/start anchor; mounted sidecars are reset too by default
info() / anchors(limit=20)	Return TapeInfo (entry count, anchor count, latest token usage, and more) and a list of AnchorSummary(name, state)
search(query)	Runs the given TapeQuery and returns matching entries
append_event(name, payload, **meta)	Appends an event entry
scoped(name) / with_context(context)	Returns a new Tape with a bound name or a replaced context
fork_tape(merge_back=True)	Async context manager: read and write on a fork, merged back into the main Tape on exit by default

TapeContext: anchor, select, and state. Use default_tape_context() to replay the complete conversation; a direct TapeContext() replays only ordinary message entries.

session_tape_name(session_id, workspace) returns a physical Tape name of the form <workspace hash>__<session hash>, which the framework uses to map one session to one Tape; SDK callers may also use their own names.

TapeSidecar

from bub.core.sidecars import TapeSidecar, sidecar_tape_name

class TapeSidecar(Protocol):
    @property
    def name(self) -> str: ...


def sidecar_tape_name(owner: str, sidecar: str) -> str: ...

A sidecar is backed by the sibling Tape <owner>__sidecar__<name>: it forks, merges, and resets with the main Tape, and its entries never enter the main context. Mounting is covered in Sidecars; the application-level provider is provide_tape_sidecar.

TapeEntry

The entry type accepted by append(tape, entry):

Field	Type	Description
id	int	Sequential ID assigned by the store; new entries carry 0 and are rewritten on append
kind	str	Common values: message, system, anchor, tool_call, tool_result, error, event
payload	dict[str, Any]	Entry content
meta	dict[str, Any]	Run metadata; context=False keeps the entry out of the model context
date	str	ISO time, defaulting to the current UTC time

Factories: TapeEntry.message(...), system(content), anchor(name, state=None), tool_call(calls, content=""), tool_result(results), error(error), and event(name, data=None). Writing through a Tape usually means you do not need to construct entries yourself.

Store protocols

class TapeStore(Protocol):
    def list_tapes(self) -> list[str]: ...
    def reset(self, tape: str) -> None: ...
    def fetch_all(self, query: TapeQuery) -> Iterable[TapeEntry]: ...
    def append(self, tape: str, entry: TapeEntry) -> None: ...

class AsyncTapeStore(Protocol):
    async def list_tapes(self) -> list[str]: ...
    async def reset(self, tape: str) -> None: ...
    async def fetch_all(self, query: TapeQuery) -> Iterable[TapeEntry]: ...
    async def append(self, tape: str, entry: TapeEntry) -> None: ...

Method	Contract
list_tapes()	Returns existing Tape names; internal physical tapes such as sidecars need not be listed
append(tape, entry)	Appends an entry and assigns an id starting at 1 and increasing monotonically within that Tape; do not rely on the caller’s entry.id
fetch_all(query)	Returns entries matching TapeQuery in recording order; the async version is still async def returning Iterable, not an async iterator
reset(tape)	Deletes the Tape’s records and ID progression, so the next append starts again at 1

Agent(store=...) accepts both synchronous and asynchronous implementations; synchronous ones are called through AsyncTapeStoreAdapter with asyncio.to_thread() to avoid blocking the event loop. is_async_tape_store(store) detects the type by checking whether append is a coroutine function. Database connections and transactions remain the implementation’s responsibility.

TapeQuery

TapeQuery(tape, store) is an immutable chain of filters; each method returns a new object and all() executes the query:

Method	Filter
query(value)	Case-insensitive containment in the JSON text of entry kind, date, payload, and meta
after_anchor(name) / last_anchor()	After the named or latest anchor
between_anchors(start, end)	Between two anchors, excluding the boundaries
between_dates(start, end)	ISO date or time range, including the boundaries; a date-only bound covers the whole day
kinds(*kinds)	Specified entry kinds
limit(value)	Maximum number of entries; applied after the other filters

Multiple conditions are combined with AND. A query for a missing anchor fails with BubError(kind=NOT_FOUND); a malformed date fails with INVALID_INPUT.

after_anchor and last_anchor return entries after the named or latest anchor, excluding the boundary anchor itself. A missing anchor raises BubError(kind=NOT_FOUND). The default query therefore does not pass the boundary anchor to the projection. When reading without anchor filtering, such as with anchor=None, the selector can receive anchor entries; the default selector converts any anchor it receives into an [Anchor created: …] assistant message. A raw TapeQuery does not apply TapeContext’s context=False filter.

For a synchronous store, all() returns an Iterable directly; for an asynchronous store, await it. A query from Tape.query() is always bound to an AsyncTapeStore:

query = tape.query().query("error").kinds("event", "error").limit(20)
entries = list(await query.all())

InMemoryQueryMixin can provide the filters above for simple backends that implement only read(tape) -> list[TapeEntry] | None; database backends should usually push filtering down into the query itself.

Built-in stores

Class	Constructor	Behavior
InMemoryTapeStore	InMemoryTapeStore()	In-process dictionary; read() returns copies of entries; lost when the process exits
FileTapeStore	FileTapeStore(directory: Path)	One <tape>.jsonl per Tape; creates the directory in the constructor; one thread lock per file; list_tapes() returns only files whose name contains exactly one __ (framework session names), excluding sidecars and unqualified names; no cross-process lock
AsyncTapeStoreAdapter	AsyncTapeStoreAdapter(store)	Adapts a synchronous TapeStore to AsyncTapeStore
InMemoryQueryMixin	Mixin	Provides in-memory filtering for a store that implements only read()

FileTapeStore uses the Tape name directly as a file name, so names must be safe as paths; multi-tenant or web services should use controlled identifiers. See Tape and memory.

EventBus

from bub.core import EventBus, ToolCallStarted, ToolCallFinished

EventBus: on / off / emit. Public event types include RunStarted / RunFinished, StepStarted / StepFinished, LLMCallStarted / LLMCallFinished, ToolCallStarted / ToolCallFinished, MessagesPending, ChatFinished, CommandFinished, and others.

Events at a glance

Event	When it fires	What a handler can read or change
RunStarted	A model Run begins	model, prompt, metadata
RunFinished	The Run’s stream closes	status (ok / error / cancelled), steps, elapsed time, error
StepStarted	A Step begins	step, optional prompt
StepFinished	A Step ends	Progress, usage; set next_prompt when continuing
MessagesPending	While a request is built, or when a Step ends	Append user content to messages
LLMCallStarted	Before calling the provider	Replace request; set LLMCallDecision.finish()
LLMCallFinished	After success or an ordinary provider failure	request and the aggregated result
ToolCallStarted	Before a tool runs	Replace call; set deny or replace
ToolCallFinished	Tool success, ordinary failure, denial, or replacement	Assign event.result.result to change the model-facing result
ChatFinished	When a Step’s exchange can be recorded	Messages, tool round-trips, usage, errors
CommandFinished	A direct command succeeds or fails	Command, status, output

RunFinished fires when the stream closes; a Run closed before iterating ends as cancelled with zero steps, and Finished events do not cover every cancellation path, so billing and cleanup must not rely on them alone. StepFinished.next_prompt supplies the next Step’s user message, and MessagesPending.messages receives messages that arrived during the Run. Exceeding max_steps (set by BUB_MAX_STEPS in the app) raises max_steps_reached. A direct command emits tool events and CommandFinished but no RunStarted or RunFinished. Every event carries state, and TapeEvent subclasses also carry tape.

Interception payloads (LLMCallRequest, LLMCallDecision, ToolCall, ToolCallDecision, and others) are also exported from bub.core. See Agent events for the decision semantics, and Hooks reference for the app-hook mapping.

Streams

from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState

StreamEvent.kind: text, reasoning, tool_call, tool_result, usage, error, and final. A tool_result event carries data["tool_results"], one entry per tool call in the Step; a failed call’s entry is its BubError, which the Tape and model receive as a dictionary. AsyncStreamEvents is an async iterator with aclose() / error / usage. See SDK output streams for consumption conventions.

Errors

from bub.errors import BubError, ErrorKind

ErrorKind: invalid_input, config, provider, tool, temporary, not_found, and unknown. BubError(kind, message, details=None); as_dict() is suitable for transport.

Code mode

from bub.core.codemode import CodeMode

CodeMode(tools, *, environment=None, models=None, ...) provides the model-facing run_code tool and a prompt section describing callable tools. See Custom tools and code mode for usage.

Environment

from bub.environment import (
    ENVIRONMENT_STATE_KEY,
    CallTool,
    CodeFailed,
    Environment,
    LocalEnvironment,
    Process,
)

Environment is the abstraction through which tools run processes and read or write files. The built-in bash, fs.read / fs.write / fs.edit, and code mode’s run_code all execute through it. The default implementation, LocalEnvironment, runs everything directly on the host without isolation.

Paths passed to an environment are paths inside that environment: relative paths resolve against workspace, which need not be a host path.

Members

Member	Signature	Description
workspace	str | None	Working directory inside the environment; relative paths are rejected when None
python	str	Python executable inside the environment, "python3" by default; LocalEnvironment uses the current sys.executable
spawn	async (command, *, cwd=None, env=None) -> Process	Abstract. A string runs through the shell, a sequence runs as an argument vector; cwd defaults to workspace, env is layered over the environment’s default variables
read_text	async (path) -> str	Abstract. Read a UTF-8 text file
read_bytes	async (path) -> bytes	Read bytes. The default reads the file with python inside the environment; implementations with direct filesystem access should override it
write_text	async (path, content) -> None	Abstract. Write a UTF-8 text file, creating missing parent directories
resolve_path	(path) -> str	Resolve a relative path against workspace; raises ValueError without a workspace. The base class uses POSIX rules; LocalEnvironment also expands ~ and resolves absolute paths
run_code	async (code, *, tools, call_tool, write) -> None	Abstract. Run model-written Python code; contract below
close	async () -> None	Release the environment’s resources; called when framework.running() exits

A custom environment must implement at least spawn, read_text, write_text, and run_code; the other members have defaults.

Process

spawn returns a Process, which exposes the process as streams:

Member	Description
stdout / stderr	asyncio.StreamReader
returncode	Exit code, or None while running
await wait()	Wait for exit and return the exit code
await write_stdin(data)	Write to standard input and wait until flushed
close_stdin()	Close standard input
signal(*, kill)	Ask the process and its descendants to exit, or force them to end when kill=True; must not raise once the process has exited
is_running()	Whether the process or its descendants are still alive; the default checks only the process itself

LocalProcess is the implementation used by LocalEnvironment: on POSIX the process runs in its own session, and signal targets the whole process group.

run_code contract

async def run_code(
    self,
    code: str,
    *,
    tools: Sequence[str],
    call_tool: CallTool,
    write: Callable[[str], None],
) -> None:

• code may use top-level await.
• Each name in tools appears in the code as tools.<name>(**kwargs); calls forward to await call_tool(name, kwargs), and exceptions from call_tool surface in the code as exceptions.
• The code may also call call_tool with names prefixed republic. for media loading and model calls.
• Everything the code writes to standard output must be passed to write as soon as it is produced, so output that has already been produced survives a timeout.
• Raise CodeFailed when the code itself raises; any other exception indicates a runtime failure. When cancelled (for example on timeout), stop the code before returning.

LocalEnvironment.run_code uses bub.core.codemode.runner.run_code_in_subprocess(self, ...): it runs the code in a Python process started with spawn, then terminates that process and everything it started. Custom environments with a Python interpreter can reuse it directly.

CodeFailed

CodeFailed(error, traceback=""): error is "<Type>: <message>", and traceback is a formatted traceback containing only the code’s own stack frames. Code mode turns it into a tool error for the model.

Where the environment comes from

The application layer provides one environment per session through the provide_environment hook (first result wins):

def provide_environment(self, session_id: str, workspace: Path) -> Environment | None:

The framework caches the returned environment per session and calls Environment.close() when framework.running() exits. The builtin implementation returns LocalEnvironment(workspace); a higher-priority plugin can replace it.

The environment is written to Turn state under ENVIRONMENT_STATE_KEY ("_runtime_environment"), and builtin tools read it through bub.builtin.environment.environment_from_state. When the state has no environment (for example when using an SDK Agent directly), the fallback is a LocalEnvironment based on _runtime_workspace. Without a workspace, relative paths raise ValueError.

Constructing CodeMode directly in the SDK does not read the environment from Turn state: CodeMode(tools, environment=...) accepts an Environment or (TurnState) -> Environment, and defaults to a LocalEnvironment() in the current working directory. The builtin code mode already wires it to environment_from_state.

Caution

LocalEnvironment executes directly on the host, and a separate process is not a security sandbox. When running untrusted model code, implement an environment with filesystem, network, and permission isolation.

Application

Framework and BubAgent

from bub import BubFramework
from bub.builtin.agent import BubAgent

Entry point	Description
BubFramework(config_file=...)	Loads application configuration; the initial workspace is the cwd
framework.load_hooks()	Registers builtin and installed plugin hooks
framework.load_builtin_hooks()	Registers only builtin implementations
framework.running()	Async scope managing the store, execution environments, and steering inbox (SteeringInbox)
framework.process_inbound(inbound, stream_output=False)	Executes a complete Turn and returns TurnResult
framework.get_tools() / get_channels(handler) / get_tape_store()	Plugin-host collection entry points
framework.get_model_options(...) / get_agent_hooks() / bind_channel_router(...)	Model options, event bridge, and channel routing
BubAgent(framework, *, tools=None, tape_store=None, skill_dirs=None, command_prefix=None)	Application Agent with settings, skills, and code mode
await agent.run_stream(session_id=..., prompt=..., state=None, ...)	Prepares the model stage and returns a stream

Configuration is cached process-wide. To use the framework’s active store, construct BubAgent inside async with framework.running():. An explicit tape_store takes precedence and remains the caller’s lifecycle responsibility. See Reuse the Bub app for embedding.

run_stream parameters: session_id (a temp/ prefix prevents the branch from merging), prompt, state (loaded through hooks when None), model, allowed_skills, allowed_tools, and reasoning_effort. Await it before iterating; it does not handle outbound rendering, save_state, or serialization of concurrent runs in the same session. See Turn for the differences from a complete Turn.

Messages and TurnResult

from bub.envelope import Envelope, content_of, field_of, normalize_envelope, unpack_batch
from bub.turn import TurnResult, TurnState

Envelope accepts mappings, dataclasses, Pydantic models, and objects with attributes. field_of / content_of read keys or attributes; normalize_envelope produces a mutable dictionary; unpack_batch expands a hook’s outbound batch.

TurnState is dict[str, Any]. TurnResult fields: session_id, prompt, model_output, outbounds, and state.

Model options

from bub.model_selection import ModelChoice, ModelOptions

ModelChoice: required id, optional name / description / meta. ModelOptions: models and current_model. BubFramework.get_model_options merges the results of each provide_model_options implementation.

Channels

from bub.channels import Channel, ChannelManager, ChannelMessage, Interface, Lifecycle
from bub.channels.contracts import ChannelRouter, MessageHandler
from bub.channels.admission import AdmitDecision, SteeringInbox, TurnSnapshot

ChannelMessage contains session_id, channel, chat_id, content, kind, context, media, is_active, output_channel, and optional message-level lifespan. MediaItem provides type, mime_type, and either url or an async data_fetcher; builtin prompt construction supports image, audio, and video. See Gateway for custom channels.

Channel implements at least start / stop; send, stream_events, and admit_message are optional. Interface is a user entry point, and Lifecycle is a background service.

ChannelRouter: dispatch_output, wrap_stream, quit, and admit_channel_message. ChannelManager implements this protocol and is bound through bind_channel_router.

Admission types: AdmitDecision(action, reason=None) (process / drop / follow_up / steer), TurnSnapshot, and SteeringInbox. follow_up processes the message after the current Turn, while steer injects it into the running Turn; see the Hooks reference and Turn for complete behavior.

Hooks

from bub.hooks import hookimpl, hookspec, BubHookSpecs

See the Hooks reference for signatures and priority.

Application events

from bub.application import CommandChanged, PromptChanged, WorkChanged

These events are published on framework.events, independently of agent.events:

Type	Fields	Trigger
CommandChanged	outcome	None when CLI invocation starts; succeeded / failed / cancelled when it ends
WorkChanged	busy, completed	Channel busy-state changes; may carry the result of a newly completed message
PromptChanged	kind	question / permission / auth when a modal prompt opens; None when it closes

They are suitable for status bars and external UIs and do not carry detailed model-call payloads.

bub.program_status.ProgramStatus is the builtin application event handler. The CLI automatically registers it on framework.events to emit OSC 7501 states on a TTY: idle, working, blocked, done, and error; prompt states also carry kind. Non-TTY output does not produce these control sequences.

Interactive input

from bub.inquirer import ask_checkbox, ask_confirm, ask_fuzzy, ask_secret, ask_select, ask_text

These synchronous functions collect multi-select, confirmation, fuzzy single-select, secret, ordinary single-select, and text input for plugin CLIs and onboard_config. When called through the Bub CLI, they publish PromptChanged; user cancellation raises typer.Abort. See Collect interactive input for usage.