Plugins and Hooks

Plugins are one of the core capabilities of the Bub app. Bub discovers plugins automatically through the standard Python Entry Points mechanism, and injects custom logic into every runtime stage of the app through Hook contracts.

Through plugins, you can extend three core capabilities for Bub:

• Custom tools: provide dedicated tool functions to the Agent via provide_tools;
• Long-running background services: mount resident daemons that run with the app (such as MCP clients, external channels, or event schedulers) via Lifecycle;
• Rewriting flow logic: intercept messages, rewrite prompts, approve tool calls, or persist session state through Hooks.

Basics: Create a plugin

Before diving into the more complex capabilities, let’s build an understanding of Bub plugins with an example plugin that requires no external model credentials.

Initialize the plugin project

Use uv to quickly scaffold a Python library project, then enter the project directory:

uv init --lib bub-echo-plugin
cd bub-echo-plugin

Create plugin.py inside the src/bub_echo_plugin/ source directory to hold the plugin and Hook logic. The initialized directory structure looks like this:

• Directorybub-echo-plugin/

  • pyproject.toml
  • Directorysrc/

    • Directorybub_echo_plugin/

      • __init__.py
      • plugin.py # New: holds the plugin and Hook logic

Declare the entry point

In pyproject.toml, expose the plugin object to Bub through [project.entry-points."bub"]:

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "bub-echo-plugin"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["bub[app]"]

[project.entry-points."bub"]
echo = "bub_echo_plugin.plugin:echo_plugin"

Note:

• entry-points name: must be "bub"; Bub automatically scans all plugins under this namespace at startup;
• Plugin identifier and target: the echo key is the plugin’s globally unique identifier; the value points to the module, instance, or factory class to be loaded.

Implement the Hook logic

When writing plugin logic in the module, mark hook implementations with the @hookimpl decorator:

from bub import hookimpl


class EchoPlugin:
    @hookimpl
    def run_model(self, prompt, session_id, state):
        return f"[echo:{session_id}] {prompt}"


echo_plugin = EchoPlugin()

This example implements run_model to replace the built-in model invocation stage, directly concatenating the input text with the session ID and returning it as the result, which makes it easy to verify the entire plugin pipeline offline.

Entry point declaration forms

Depending on whether a plugin needs to maintain internal state or access host context, an entry point supports three target forms:

Entry point target syntax	Registered object	When to use
bub_myplugin.plugin	Module	Hooks are pure functions with no shared state
bub_myplugin.plugin:plugin	Existing instance	Hooks need to share state through self
bub_myplugin.plugin:MyPlugin	Class or factory function	The plugin needs access to the global BubFramework instance

When the entry point targets a class or factory function, Bub automatically passes the current BubFramework instance as the first argument when instantiating the object:

from bub import BubFramework, hookimpl


class MyPlugin:
    def __init__(self, framework: BubFramework) -> None:
        self.framework = framework

    @hookimpl
    def build_prompt(self, message, session_id, state):
        # Access global runtime information such as the workspace path through the framework
        return f"Current workspace path: {self.framework.workspace}\n{message}"

Install and verify

After writing the plugin, you need to install it into the runtime environment where Bub runs, and verify how Hooks are loaded and in what order they execute.

Install for local development

From the plugin project root, run sync to install it into the current virtual environment in editable mode:

uv sync

Check hook registration

Use the bub hooks command to check whether the plugin has been recognized correctly:

uv run bub hooks

In the hook list printed in the terminal, the registered echo plugin should appear correctly under run_model.

Run a test

Send a test command to the app to verify that the plugin actually takes effect:

uv run bub run "hello from my plugin"

The console prints the echoed text intercepted by the plugin:

[echo:cli:local] hello from my plugin

Troubleshooting and dispatch

If the plugin is not loaded or executed as expected, troubleshoot using the following principles:

1. Check the runtime environment: confirm that the Python environment running the bub CLI is exactly the same as the one where the plugin is installed;
2. Re-sync metadata: if you changed the entry point configuration in pyproject.toml, you must run uv sync (or pip install -e .) again to refresh the package’s Entry Points metadata;
3. Check hook execution order and priority: Bub follows the Pluggy specification, registering built-in implementations first and then loading entry point plugins. At equal priority, plugins registered later run first (LIFO). To explicitly adjust the order, specify @hookimpl(tryfirst=True) (run as early as possible) or @hookimpl(trylast=True) (run as late as possible) in the decorator. See Hook priority and dispatch.

Advanced: Customize behavior with Hooks

Bub provides a full set of extension points covering inbound parsing, model invocation, tool interception, state, and lifecycle (see the Hooks reference for complete signatures and lifecycle contracts). In real-world scenarios, the most common customization needs fall into the following four categories:

1. Provide tools

Plugins can register Python functions declared with @tool to the Agent through the provide_tools hook:

from bub import hookimpl, tool
from bub.core.tools import Tool


@tool(name="orders.get")
def get_order(order_id: str) -> dict[str, str]:
    """Look up an order's shipping status by order ID."""
    return {"id": order_id, "status": "shipped"}


class OrdersPlugin:
    @hookimpl
    def provide_tools(self) -> list[Tool]:
        return [get_order]

Deferred loading

When a plugin provides a large number of tools, injecting the full schema of every tool into the system prompt at once consumes a significant amount of the context window. You can enable deferred loading in the decorator:

@tool(name="orders.export", deferred=True)
def export_orders(format: str) -> str:
    """Export a report of all historical orders."""
    ...

Note:

• Deferred mechanism: tools with deferred=True only expose their name and a brief description to the system prompt by default;
• Dynamic expansion: when the model decides it needs to call the tool, it automatically loads the full definition temporarily through the built-in tool.describe, and starts executing the call from the next step.

2. Rewrite and intercept flows

Rewrite the prompt

• Rewrite model input: build_prompt receives the inbound message, session ID, and the current turn’s state, and returns the final input provided to the model:

  from bub import hookimpl
  from bub.envelope import content_of
  
  
  @hookimpl
  def build_prompt(message, session_id, state):
      text = content_of(message)
      return [f"Please answer using the project's professional terminology:\n\n{text}"]

• Append system instructions: if you only want to append global constraints to the model without changing the inbound content, use system_prompt (content appended by multiple plugins is concatenated with line breaks according to priority):

  @hookimpl
  def system_prompt(prompt, state):
      return "Please include the paths of the referenced code files at the end of your reply."

Intercept calls

Interception hooks run before the actual call is triggered. Exceptions raised are recorded by the framework and skipped; to actively intervene, modify, or block execution, you must return the corresponding decision object (Decision):

• Limit LLM sampling parameters:

  from dataclasses import replace
  from bub import hookimpl
  from bub.hooks.interception import LLMCallRequest
  
  
  @hookimpl
  def before_llm_call(request: LLMCallRequest, state):
      max_tokens = min(request.max_tokens or 512, 512)
      return replace(request, max_tokens=max_tokens)

• Approve or deny high-risk tool calls:

  from bub import hookimpl
  from bub.hooks.interception import ToolCall, ToolCallDecision
  
  
  @hookimpl
  def before_tool_call(call: ToolCall, state):
      if call.tool == "orders.cancel" and not state.get("can_cancel_orders", False):
          return ToolCallDecision.deny("This session is not allowed to cancel orders.")
      return None

For the semantics of proceeding, rewriting, denying, and replacing, see Decisions come before side effects; for the complete return types, see the Hooks reference.

Call auditing

Completion hooks fire after a call finishes, making them suitable for measuring latency, reporting monitoring metrics, or recording audit logs:

from bub import hookimpl


@hookimpl
def after_llm_call(request, result, state):
    print(f"model={request.model} error={result.error!r} elapsed={result.duration_ms}ms")


@hookimpl
def after_tool_call(call, result, state):
    print(f"tool={call.tool} error={result.error!r} elapsed={result.duration_ms}ms")

Persist state across turns

The example below uses load_state and save_state to keep a count across Turns. For the lifetime of TurnState and where data belongs, see Turn and where data belongs.

import asyncio
from bub import hookimpl


class TurnCounter:
    def __init__(self) -> None:
        self.counts: dict[str, int] = {}
        self.lock = asyncio.Lock()

    @hookimpl
    async def load_state(self, message, session_id):
        async with self.lock:
            return {"completed_turns": self.counts.get(session_id, 0)}

    @hookimpl
    async def save_state(self, session_id, state, message, model_output):
        async with self.lock:
            self.counts[session_id] = state.get("completed_turns", 0) + 1


turn_counter = TurnCounter()

3. Mount services and resources

Long-running services

Components that must run alongside the app, such as MCP clients, channel listeners, or scheduled dispatchers, should inherit from the Lifecycle base class and be exposed through provide_channels:

from bub.channels.base import Lifecycle


class MyRuntime(Lifecycle):
    name = "myapp.lifecycle"

    async def start(self, stop_event) -> None:
        await self._connect(stop_event)

    async def stop(self) -> None:
        await self._close()

Note:

• Lifecycle contract: long-running services are initialized automatically when bub gateway starts or the channel runtime starts, and are cleaned up on exit; note that a single bub run execution does not activate long-running services.
• CLI filtering: at startup, you can enable one individually with --enable-channel, or explicitly exclude it with --enable-channel '!myapp.lifecycle'.

Lifecycle resources

When you need to hold resources such as a global database connection pool or HTTP client for the entire process lifetime, implement provide_lifespan with an async generator:

from bub import hookimpl


@hookimpl
async def provide_lifespan():
    client = await connect_database()
    try:
        yield
    finally:
        await client.close()

Session storage

provide_tape_store lets you replace the default local-file session store (for example, to use PostgreSQL or cloud storage; see SQLAlchemy integration):

from pathlib import Path
from bub import hookimpl
from bub.core.store import FileTapeStore


@hookimpl
def provide_tape_store():
    return FileTapeStore(Path("./application-tapes"))

4. Extend CLI commands

Plugins can register subcommands directly with the Bub root CLI (built on Typer) through register_cli_commands:

import typer
from bub import hookimpl


@hookimpl
def register_cli_commands(app: typer.Typer):
    @app.command()
    def my_cmd(name: str = "world") -> None:
        typer.echo(f"Executed successfully: {name}")

Interactive input

If a plugin subcommand needs to interactively confirm with the user or obtain input in the terminal, use the interactive input utilities provided by bub.inquirer:

from bub.inquirer import ask_confirm, ask_text

token = ask_text("Enter the verification token")
if ask_confirm("Continue execution?", kind="permission"):
    pass

Distribution and extensions

Package and distribute

A finished plugin can be packaged as a standard Python Wheel for distribution. A single distribution can contain plugin entry points, skill namespace packages, and custom CLI commands at the same time. For packaging configuration and pre-release verification, see Packaging and distribution. For cross-language ecosystem package specifications, see Load Agent Plugins.

WASM plugins (experimental)

With the bub-extism extension, binary modules compiled from other languages such as Go and Rust can be mounted directly as Bub Hook handlers via WebAssembly.

extism.json specifies the module files and hook mapping:

{
  "plugins": {
    "prompt": {
      "manifest": { "wasm": [{ "path": "go-build-prompt.wasm" }] },
      "wasi": true,
      "hooks": { "build_prompt": "build_prompt" }
    }
  }
}

export BUB_EXTISM_CONFIG_PATH=./extism.json
bub run "Say hello from a prompt built outside Python."