Application types and lifecycle

Use these types when writing a plugin, adding a channel, or embedding the Bub application. Shared agent capabilities are covered in Tools, Tape and context, and Events.

Messages and turn results

Import message helpers from bub.envelope:

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 and content_of read both mapping keys and object attributes. normalize_envelope produces a mutable dictionary; unpack_batch normalizes a hook’s outbound batch.

TurnState is dict[str, Any]. The framework creates a fresh dictionary for each inbound turn, then merges load_state results into it. Plugins choose which values to persist.

process_inbound returns a TurnResult with these fields:

Field	Type	Contents
session_id	str	Resolved session identity
prompt	list[UserContent]	Normalized input sent to the model stage
model_output	str	Collected text result
outbounds	list[Envelope]	Rendered outbound messages
state	TurnState	State used during this turn

UserContent comes from bub.prompt. The prompt contract explains its media types and conversion to legacy hook inputs.

Model choices

from bub.model_selection import ModelChoice, ModelOptions

ModelChoice has a required id and optional name, description, and meta. ModelOptions contains models: list[ModelChoice] and an optional current_model.

BubFramework.get_model_options combines provide_model_options results. A channel or host presents the choices and stores the selection.

Streams

from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState

StreamEvent contains kind and a data dictionary. Kinds are text, reasoning, tool_call, tool_result, usage, error, and final. Text events carry data["delta"]. A builtin agent’s final marks a model step; continue consuming until the stream ends.

AsyncStreamEvents is an async iterator with aclose(), error, and usage. Exhaust it or close it when your consumer stops:

from contextlib import aclosing


async def consume(stream: AsyncStreamEvents) -> None:
    async with aclosing(stream):
        async for event in stream:
            if event.kind == "text":
                print(event.data.get("delta", ""), end="")

When a plugin creates an AsyncStreamEvents, it supplies the iterator and may supply a StreamState and async on_close callback. Use that callback for resources acquired before iteration begins. If you wrap another stream, the wrapper must also close its upstream and preserve the relevant error and usage information.

Stream events are output for consumers. Agent events have a separate delivery and control contract.

Errors and interception payloads

from bub.errors import BubError, ErrorKind
from bub.hooks.interception import (
    LLMCallDecision,
    LLMCallRequest,
    LLMCallResult,
    ToolCall,
    ToolCallDecision,
    ToolCallResult,
)

BubError.kind identifies the error category; as_dict() returns a payload suitable for transport. Interception payloads are shared with bub.core.events. Their use by plugins, including exception handling, is defined in Application hooks.

Channels and routing

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

MessageHandler is an async callable that accepts an envelope and returns None. A Channel implements start(stop_event) and stop(); it can also implement sending, stream handling, and message admission. Interface marks user-facing channels; Lifecycle marks background services.

ChannelRouter connects the framework to channel delivery:

Method	Role
dispatch_output(message)	Send an outbound envelope; return whether it was handled
wrap_stream(message, stream)	Return the channel’s async stream wrapper
quit(session_id)	End the session through its channel
admit_channel_message(session_id, message, turn)	Return a channel admission decision or None

ChannelManager implements this interface and binds it with BubFramework.bind_channel_router.

Admission and steering

TurnSnapshot exposes session_id, is_running, running_count, pending_count, and steering_count. An admission hook returns AdmitDecision(action, reason=None):

Action	Effect
process	Process the message
drop	Discard the message
follow_up	Queue it after active work
steer	Queue it for the running agent to consume

A SteeringInbox provides async enqueue_message(message, state) and drain_messages(state), plus synchronous message_count(state). The application obtains it through provide_steering_inbox.

Framework and agent entry points

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

Entry point	Responsibility
BubFramework()	Load application configuration and establish the workspace
framework.load_hooks()	Register builtin and installed plugin hooks
framework.running()	Acquire and release application resources
framework.process_inbound(inbound, stream_output=False)	Complete the application turn and return TurnResult
BubAgent(framework, tools=None, tape_store=None, skill_dirs=None)	Combine the core agent with application behavior
agent.run_stream(session_id=..., prompt=..., state=None, ...)	Return the application agent’s output stream

Application configuration is process-wide. Construct a BubAgent inside framework.running() when it should use the framework’s active store. An explicitly supplied store takes precedence and remains the caller’s lifecycle responsibility. Without either store, the agent uses memory.

run_stream loads initial state when the caller omits state and uses caller-supplied state directly otherwise. The host owns save-state hooks, outbound handling, and concurrent turn scheduling. See Embed the application for complete usage.

Imports

The top-level bub package exports BubFramework, Settings, config, ensure_config, home, hookimpl, and tool. Import other types from the defining modules above.