Hooks

This page lists the complete interface of all 27 hooks. A hook is an extension point through which plugins participate in framework flows: the same hook can be implemented by multiple plugins, the framework calls them in the order described below, and it decides what happens next based on their return values.

For concrete plugin examples, see Plugins and hooks; to learn which stages a message passes through, see Turn.

Understanding dispatch modes

Every hook’s overview and details state its “dispatch mode”, that is: how the framework calls the multiple implementations of the hook and how it handles their return values.

Dispatch mode	Meaning
First result	Calls implementations one by one by priority and uses the first non-None return value; later implementations are not run. False, "", and [] are also return values and do not equal None
Run all	Calls every implementation and collects the return values. Whether one failure affects the whole depends on the specific hook and call site
Sequential	Calls implementations one by one by priority; later implementations can see the results modified by earlier ones. Certain return values can end processing early. Used by Agent hooks

These types are also used in the tables:

Type	Meaning
Envelope	An alias for Any. A mapping, dataclass, Pydantic model, or object with attributes can be passed
TurnState	An alias for dict[str, Any], shared by hooks, tools, and events within the current Turn

All signatures below are taken directly from bub.hooks.specs.BubHookSpecs, so they uniformly use def and keep self. When implementing, declare only the parameters you actually use; the framework passes values by parameter name. Whether an asynchronous implementation is available depends on the specific call path; when it is unsupported, that implementation is skipped and hook.async_not_supported is logged.

With multiple plugins, who runs first

Bub uses pluggy to register hook implementations, and HookRuntime to invoke them uniformly.

Rule	Meaning
Registration order	Builtin implementations (builtin) are registered first, then plugins from bub entry points
Priority	@hookimpl(tryfirst=True) runs first; trylast=True runs last
Same priority	Implementations registered later run first
Runtime order	Calls the list returned by get_hookimpls() in reverse order, one item at a time
Argument passing	Passes arguments according to the argnames declared by an implementation; declare only the parameters you actually use
Unsupported arguments	wrapper=True, hookwrapper=True

If you call HookRuntime directly, the two result-collection methods differ as follows:

Method	Behavior
call_first / call_first_sync	Returns the first non-None result; empty containers and False also count as results
call_many / call_many_sync	Runs every implementation and collects results that were not skipped; whether it continues after a failure depends on the specific call site

The plugin list printed by bub hooks follows pluggy’s registration order, not the actual execution order. When diagnosing hook order, rely on the priority rules and the logs.

Turn Hooks

A single message-processing flow is called a Turn. The following nine hooks are triggered by BubFramework.process_inbound() and cover the whole process from determining the session to sending the reply.

Hook	What it does	Dispatch
resolve_session	Determines which session a message belongs to	First result
load_state	Restores the state needed for this Turn	Run all and merge results
build_prompt	Builds the prompt sent to the model	First result
run_model	Calls the model and returns the complete text	First result
run_model_stream	Calls the model and returns streaming events	First result
continue_prompt	Optionally appends a user message when the Agent continues running	First result
save_state	Saves application state after the model stage ends	Run all
render_outbound	Converts model output into messages to be sent	Run all and concatenate messages
dispatch_outbound	Hands messages to be sent to external channels	Run all, once per message

Exceptions from this group of hooks are ultimately caught by process_inbound(): it first calls on_error(stage="turn", ...) and then re-raises the original exception. When no implementation is found in the model stage, or when an error is reported in streaming output, stage="run_model" is used.

resolve_session

Determines the session_id for a new message so the framework knows which session it belongs to.

def resolve_session(self, message: Envelope) -> str:

• Dispatch: First result.
• Default fallback: When the result is empty, channel:chat_id is used as the session ID.

load_state

Restores the application state needed for this Turn from the session store.

def load_state(self, message: Envelope, session_id: str) -> TurnState:

• Dispatch: Run all; lower-priority results are merged first, higher-priority results after.
• Return value: Non-dictionary return values are ignored.

build_prompt

Builds the prompt sent to the model based on the original message and the restored state.

def build_prompt(self, message: Envelope, session_id: str, state: TurnState) -> list[UserContent] | LegacyPrompt:

• Dispatch: First result.
• Default fallback: Uses the first non-None return value; if that value is "" or [], the framework does not try later implementations and instead uses the original message content.

Prompt content

build_prompt returns list[UserContent]; list elements can be constructed as follows:

Content	Construction
Text	str
Image	republic.image(url, media_type=...)
Audio	republic.audio(url, media_type=...)
Video	republic.video(url, media_type=...)

Hooks that use the legacy prompt format (run_model, run_model_stream, continue_prompt, and system_prompt) receive:

Situation	Format received
No media	str
Contains media	A list of [{"type", "media_type", "url"}] content blocks

to_content() still accepts legacy strings and content-block lists, and handles the format conversion. To convert back to the legacy format, use bub.prompt.to_legacy_prompt().

run_model

Calls the model and returns the complete text output directly.

def run_model(self, prompt: str | list[dict], session_id: str, state: TurnState) -> str:

• Dispatch: First result.
• Prompt format: Receives the legacy prompt format; see Prompt content.

run_model_stream

Calls the model and returns streaming events that can be consumed one at a time.

def run_model_stream(self, prompt: str | list[dict], session_id: str, state: TurnState) -> AsyncStreamEvents:

• Dispatch: First result.
• Prompt format: Receives the legacy prompt format; see Prompt content.
• Relationship with run_model: You only need to implement one of the two. Plain text can be converted into streaming output, and streaming output can be aggregated into plain text.
• After replacing the builtin model stage: The builtin Agent’s interception hooks do not run automatically.

The framework selects the model hook implementation according to the priority rules above; tryfirst / trylast also apply to these two hooks.

continue_prompt

Optionally adds a user message when the Agent needs to continue running.

def continue_prompt(self, prompt: str | list[dict] | None, tape: Tape, state: StreamState) -> str | None:

• Dispatch: First result.
• When it is called: When StepFinished.status == "continue".
• Returning None: The conversation continues directly without adding a new user message.
• State type: Receives StreamState, not TurnState.
• Prompt format: Receives the legacy prompt format; see Prompt content.

save_state

Saves application state after the model stage ends.

def save_state(
    self,
    session_id: str,
    state: TurnState,
    message: Envelope,
    model_output: str,
) -> None:

• Dispatch: Run all.
• When it is called: In the finally block of the model stage. As long as execution has entered the model stage, it runs whether the model succeeds, fails, or is cancelled.
• Scope: It only saves application state; it is not responsible for cleaning up resources.

render_outbound

Converts model output into one or more messages to be sent.

def render_outbound(
    self,
    message: Envelope,
    session_id: str,
    state: TurnState,
    model_output: str,
) -> list[Envelope]:

• Dispatch: Run all; the message lists returned by each implementation are expanded and concatenated in order.
• Default fallback: If no implementation produces a message to send, the framework generates one reply from model_output, reusing the original message’s channel and chat_id.

dispatch_outbound

Hands a message to be sent to an external channel.

def dispatch_outbound(self, message: Envelope) -> bool:

• Dispatch: Run all; called once per message to be sent.
• Return value: Ignored; you cannot prevent other implementations from running through the return value.
• When you need to filter messages: Do it in the render_outbound stage.

App Hooks

The following 14 hooks manage configuration, resources, channels, and tools. They serve the whole application rather than handling a single message.

Hook	What it does	Dispatch
register_cli_commands	Registers commands with the root CLI app	Run all
onboard_config	Supplements or modifies onboarding configuration	Run all
provide_model_options	Provides models to choose from	Run all and merge results
provide_environment	Provides the environment for tool execution and file access	First result
on_error	Observes exceptions from framework stages	Run all
system_prompt	Provides system prompts	Run all and concatenate results
provide_tape_store	Provides storage for conversation Tapes	First result
provide_lifespan	Manages resources for the application runtime	Run all
provide_tape_sidecar	Attaches a Sidecar to each session Tape	Run all and deduplicate by name
provide_channels	Provides channels for receiving messages	Run all and deduplicate by name
build_tape_context	Builds the Tape context for the current session	First result
admit_message	Decides how to handle a new message	First result
provide_steering_inbox	Provides the inbox for messages that steer the current task	First result
provide_tools	Provides tools to the Agent	Run all and deduplicate by name

register_cli_commands

Registers custom CLI commands with Bub’s root Typer app.

def register_cli_commands(self, app: Any) -> None:

• Dispatch: Run all.

onboard_config

Supplements or modifies configuration during onboarding.

def onboard_config(self, current_config: dict[str, Any]) -> dict[str, Any] | None:

• Dispatch: Run all.
• Input: The currently loaded configuration, plus the modifications earlier implementations have made to it.
• Return value: The returned dictionary is merged into the configuration; the value used for merging must be a dictionary, otherwise a TypeError is raised.
• Plugin filtering: collect_onboard_config() is responsible for filtering by plugin name.

provide_model_options

Provides models for the session to choose from.

def provide_model_options(
    self,
    session_id: str,
    workspace: Path | None,
) -> ModelOptions | None:

• Dispatch: Run all; merges model lists and keeps the first current_model.
• Scope: It only provides candidate models. Whether to show them to the user, and how to persist the current model, is up to the caller.

provide_environment

Provides the runtime environment for bash, file tools, and code execution.

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

• Dispatch: First result.
• Lifecycle: One is cached per session and closed when running() exits.
• Default implementation: When there is no custom implementation, the builtin implementation returns LocalEnvironment.

on_error

Observes exceptions from framework stages; useful for logging and error reporting.

def on_error(self, stage: str, error: Exception, message: Envelope | None) -> None:

• Dispatch: Run all; when an implementation fails, it only logs hook.on_error_failed and is skipped.
• Scope: Only for observing errors; it cannot recover the task or prevent the original exception from propagating.

Common values of stage:

stage	When it appears
turn	An uncategorized exception occurs in a Turn, including exceptions in prompt, model-stage, and reply handling
run_model	No model implementation is found, or an error is reported in streaming output
resolve_session	Session resolution fails before a channel decides whether to accept a message
admit_message	An error occurs while a channel decides whether to accept a message

system_prompt

Provides a system prompt block that is prepended to the model prompt.

def system_prompt(self, prompt: str | list[dict], state: TurnState) -> str:

• Dispatch: Run all; joins non-empty results with blank lines from low to high priority.
• Multiple implementations: All results are kept and do not overwrite one another; even a higher-priority implementation can only append its own content.
• Prompt format: Receives the legacy prompt format; see Prompt content.

provide_tape_store

Provides the Tape store used to record conversations.

def provide_tape_store(self) -> TapeStore | AsyncTapeStore | None:

• Dispatch: First result; may return a generator.
• When it is called: Resolved only once during framework.running(). Outside that scope, get_tape_store() returns None.
• When you need to clean up resources: Return a generator and release the connection in finally.
• Unsupported form: Do not return an object decorated with @contextmanager / @asynccontextmanager; the runtime treats it as an ordinary value.

provide_lifespan

Holds resources that need to be started and cleaned up during framework.running().

def provide_lifespan(self) -> AsyncIterator[None] | Iterator[None] | None:

• Dispatch: Run all; each generator is started as a context manager.
• Return value: Must return a generator and yield exactly once.
• Startup failure: If any generator fails to start, the entire running() cannot start.
• Unsupported form: Do not return an object decorated with @contextmanager / @asynccontextmanager; the runtime treats it as an ordinary value.

provide_tape_sidecar

Provides a Sidecar for each session Tape, used to store data or capabilities attached to the main Tape.

def provide_tape_sidecar(self) -> TapeSidecar:

• Dispatch: Run all; deduplicates by name, keeping the highest-priority result when names collide.
• Storage name: The Tape is physically stored under the name <main tape>__sidecar__<sidecar>.
• Lifecycle: It is forked, merged, archived, and reset together with the main Tape. Removing the hook that provides a Sidecar does not delete existing data.

provide_channels

Provides channels for receiving messages.

def provide_channels(self, message_handler: MessageHandler) -> list[Channel]:

• Dispatch: Run all; deduplicates by name, keeping the highest-priority result when names collide.
• message_handler: An async function (message) -> None, called by the channel when a message arrives.
• Return value: Must return a list of channels, not a single channel.

build_tape_context

Builds the Tape context for the current session, used to generate context messages.

def build_tape_context(self) -> TapeContext:

• Dispatch: First result.
• Return value: Must be a TapeContext, otherwise a TypeError is raised.

admit_message

Decides, when a new message arrives, whether to process it immediately, drop it, queue it, or use it to steer the current task.

def admit_message(
    self,
    session_id: str,
    message: Envelope,
    turn: TurnSnapshot,
) -> AdmitDecision | None:

• Dispatch: First result.
• Returning None: Handles the message according to the channel’s default rules.
• Scope: It only decides how to handle the new message and does not create tasks; the actual effects are carried out by ChannelManager.

When it returns an AdmitDecision, ChannelManager handles the new message according to action:

action	Session is currently processing a message	Session is idle
process	Creates a new Turn immediately, in parallel with the current Turn	Creates a Turn immediately
drop	Drops the new message	Drops the new message
follow_up	Queues it; processes it in arrival order after the current Turn completes	Creates a Turn immediately
steer	Sends the new message to the current Turn to steer the running task; if it cannot be injected or is not handled in time, it is queued instead	Creates a Turn immediately

In short, steer means “adjust the current task to the new requirement”, while follow_up means “handle the next message after the current task completes”.

provide_steering_inbox

Provides an inbox for holding steer messages temporarily; MessagesPending then delivers those messages to the Agent.

def provide_steering_inbox(self) -> SteeringInbox | None:

• Dispatch: First result; may return a generator.
• When you need to clean up resources: Return a generator and release the connection in finally.
• Unsupported form: Do not return an object decorated with @contextmanager / @asynccontextmanager; the runtime treats it as an ordinary value.

provide_tools

Provides custom tools to the Agent.

def provide_tools(self) -> list[Tool]:

• Dispatch: Run all; deduplicates by name, keeping the highest-priority result when names collide.
• Duplicate tool names: A tool may be provided multiple times, but only the highest-priority implementation is used and the other tools with the same name are ignored.

When resources are created and released

framework.running() acquires resources in the following order:

1. Starts all generators returned by provide_lifespan.
2. Acquires the result of provide_tape_store.
3. Acquires the result of provide_steering_inbox.

On exit, the framework shuts down session execution environments and then releases the resources above in reverse order; cleanup runs even if an exception occurs partway through.

Scenario	How long resources last
bub run	While processing one message
bub chat / bub gateway	For the entire application runtime
Embedded in another program	Determined by the host; should usually cover the whole service lifecycle

• Do not let multiple running() scopes overlap on the same Framework.
• Background shell process-tree termination: on POSIX, SIGTERM is sent first and then SIGKILL; on Windows, only the direct shell is terminated. Even if the host forcibly kills the process, you still need to complete cleanup yourself.

Errors during the runtime

on_error can observe framework errors that occur during the runtime above, but it cannot recover the task and does not change how exceptions propagate. When embedding in another program, the host should catch exceptions raised by process_inbound() itself. Unhandled CLI exceptions usually exit with a non-zero status; however, for some model-stage errors, bub run only logs them and returns fallback text, so it may still exit with 0.

Agent Hooks

These four hooks run before and after model or tool calls. AgentPlugins triggers them by subscribing to Agent events, and they can also be connected directly through PluginAgentHooks.

Hook	What it does	Dispatch
before_llm_call	Observes, modifies, or terminates the request before calling the model	Sequential
after_llm_call	Observes the final result of one model call	Run all
before_tool_call	Modifies arguments, replaces the result, or denies the call before executing a tool	Sequential
after_tool_call	Observes or modifies the final result of one tool call	Run all

When an implementation in this group raises an exception, the framework logs hook.agent_hook_failed and skips it; when the return type is incorrect, it logs hook.agent_hook_bad_return and skips it. Raising an exception does not block tool execution; to intercept a tool you must return a decision.

before_llm_call

Observes or modifies the request, or supplies the final text directly, before the request is sent to the model provider.

def before_llm_call(self, request: LLMCallRequest, state: TurnState) -> LLMCallRequest | LLMCallDecision | None:

• Dispatch: Sequential. When it returns a modified LLMCallRequest, later implementations see the new value.
• Modifying the request: Because this is a frozen dataclass, use dataclasses.replace() to create a modified copy.
• Ending early: LLMCallDecision.finish(text) skips the model provider and uses text directly as the model output; it does not trigger after_llm_call either.
• Tool list: LLMCallRequest only exposes tool names and cannot add or remove tools through it. To change the tool set, do it in Agent.prepare_tools() or in the application’s tool policy.

after_llm_call

Observes the final result of one model call.

def after_llm_call(self, request: LLMCallRequest, result: LLMCallResult, state: TurnState) -> None:

• Dispatch: Run all; return values are ignored.
• When it is called: Runs whether the call succeeds or fails with an ordinary exception.
• When it does not run: When the task is cancelled, the consumer closes the output stream on its own, or before_llm_call ends early via finish(text).

before_tool_call

Modifies arguments, directly replaces the result, or denies the call before the tool executes.

def before_tool_call(self, call: ToolCall, state: TurnState) -> ToolCallDecision | None:

• Dispatch: Sequential.
• Continuing: Return None or ToolCallDecision.proceed(arguments); the latter can modify arguments, and later implementations see the new values.
• Replacing the result: ToolCallDecision.replace(result) skips the tool, uses the supplied result, and stops later interception implementations from running.
• Denying the call: ToolCallDecision.deny(message) skips the tool and returns a tool error, and also stops later interception implementations from running.

after_tool_call

Observes the final result of one tool call, and can also modify the result the model actually sees.

def after_tool_call(self, call: ToolCall, result: ToolCallResult, state: TurnState) -> None:

• Dispatch: Run all; all implementations share the same mutable result, and return values are ignored.
• When it is called: Runs whether the call succeeds, fails with an ordinary exception, is denied, or is replaced.
• When it does not run: When the task is cancelled or the consumer closes the output stream on its own.
• Modifying the result: Assigning to result.result changes the result the model sees; result.error still retains the original error.

Interception payload definitions

The following field definitions are taken directly from bub.core.events and match the Agent hook signatures. To focus on the interface, the code omits docstrings and convenience constructors.

LLMCallRequest

A request about to be sent to the model provider. The object is immutable; create a copy with dataclasses.replace().

@dataclass(frozen=True)
class LLMCallRequest:
    run_id: str
    model: str
    messages: list[republic.Message]
    tool_names: tuple[str, ...] = ()
    max_tokens: int | None = None

LLMCallResult

The final result of one model call. For streaming calls, the complete result is aggregated first, and then after_llm_call is triggered.

@dataclass(frozen=True)
class LLMCallResult:
    run_id: str
    text: str | None = None
    tool_calls: list[dict[str, Any]] = field(default_factory=list)
    usage: dict[str, Any] | None = None
    error: Exception | None = None
    duration_ms: int = 0

LLMCallDecision

The early-termination decision returned by before_llm_call, constructed with LLMCallDecision.finish(text).

@dataclass(frozen=True)
class LLMCallDecision:
    action: Literal["finish"] = "finish"
    text: str = ""

ToolCall

A tool call about to be executed. context is the tool runtime’s context; it is None when no context is available.

@dataclass(frozen=True)
class ToolCall:
    run_id: str
    tool: str
    arguments: dict[str, Any]
    context: ToolContext | None = field(default=None, repr=False, compare=False)

ToolCallDecision

The handling decision returned by before_tool_call, constructed with proceed(), replace(), and deny() respectively.

@dataclass(frozen=True)
class ToolCallDecision:
    action: Literal["proceed", "replace", "deny"] = "proceed"
    arguments: dict[str, Any] | None = None
    result: Any = None
    message: str | None = None

ToolCallResult

The final result of one tool call. This dataclass is not frozen=True, so after_tool_call can assign directly to result.result.

@dataclass
class ToolCallResult:
    run_id: str
    tool: str
    arguments: dict[str, Any]
    result: Any = None
    error: Exception | None = None
    duration_ms: int = 0

App hooks and Agent events

The concept page explains who intervenes in which direction; this section covers the adapter details. For the relationship between the three kinds of data and the decision semantics, see Agent events.

Event mapping

PluginAgentHooks connects app hooks to Agent events:

App hook	Agent event	Notes
before_llm_call	LLMCallStarted	Before every model call; may return a rewritten LLMCallRequest or an LLMCallDecision
after_llm_call	LLMCallFinished	Once after success or ordinary failure
before_tool_call	ToolCallStarted	Before every tool call; may proceed(arguments), replace(result), or deny(message)
after_tool_call	ToolCallFinished	Fires on success, failure, denial, and replacement
continue_prompt	StepFinished	Called only when status == "continue"; the result becomes next_prompt
provide_steering_inbox	MessagesPending	Drains steering messages into the next step of the current Run

Model-visible tool calls, including built-in tools such as .tape.handoff, also pass through ToolCallStarted / ToolCallFinished.

How the two differ

Aspect	Handling Agent events directly	App interception hooks
Registration	agent.events.on(Type, handler)	@hookimpl in a plugin
Order	Registration order	pluggy priority
Argument changes	Replace event.call / event.request in place	Return ToolCallDecision.proceed(arguments=...) or a new LLMCallRequest
Control	Set event.decision	Return a decision object
Exceptions	Propagate to the emit() caller and may fail the Run	Logged and skipped; the Run continues
Bad return type	Not applicable	Logged as a warning and skipped

A security policy written as a hook therefore cannot block a call by raising; it must return deny(...).

How to subscribe

PluginAgentHooks(runtime).subscribe(events) attaches the four interception hooks to an existing or new event bus; AgentPlugins.attach(agent) subscribes automatically and also wires continue_prompt, the steering inbox, and the system prompt. For embedding and lifecycle requirements, see Reusing Bub.

PluginAgentHooks

PluginAgentHooks converts Agent events into the hooks above:

Agent event	Corresponding hook
LLMCallStarted	before_llm_call
LLMCallFinished	after_llm_call
ToolCallStarted	before_tool_call
ToolCallFinished	after_tool_call

The difference between listening on the EventBus directly and using interception hooks is as follows:

Item	Interception hook	Direct EventBus handler
Exception handling	Logged and skipped	Propagated to the emit() caller
Control	Returns a decision	Modifies event fields
Execution order	Determined by pluggy priority	Handler registration order

PluginAgentHooks.subscribe(events) attaches the adapter to an existing event bus, or creates a new one. AgentPlugins.attach(agent) subscribes automatically and connects continue_prompt, the steering inbox (SteeringInbox), and system prompts to the Agent.

AgentPlugins

AgentComponents

class AgentComponents(TypedDict, total=False):
    store: TapeStore | AsyncTapeStore
    context: TapeContext
    sidecars: Sequence[TapeSidecar]

AgentPlugins

class AgentPlugins:
    def __init__(
        self,
        framework: BubFramework | None = None,
        *,
        plugins: Iterable[object] | None = None,
    ) -> None: ...

    def running(self) -> AbstractAsyncContextManager[AsyncExitStack]: ...

    def components(self) -> AgentComponents: ...

    def attach(self, agent: Agent, *, tools: bool = True) -> Agent: ...

    async def build_state(
        self,
        session_id: str,
        *,
        message: Envelope | None = None,
        agent: Agent | None = None,
    ) -> TurnState: ...