插件与 Hooks

插件（Plugin）是 Bub 应用核心的能力之一。Bub 基于标准 Python Entry Points（入口点）机制自动发现插件，并通过 Hook 契约将自定义逻辑注入应用的各个运行阶段。

通过插件，你可以为 Bub 扩展三类核心能力：

• 自定义工具：通过 provide_tools 为 Agent 提供专属工具函数；
• 常驻后台服务：通过 Lifecycle 挂载随应用运行的常驻守护进程（如 MCP 客户端、外部通道或事件调度器）；
• 改写流程逻辑：通过 Hook 实现拦截消息、改写提示词、审批工具调用或持久化会话状态。

基础：创建插件

在深入各类复杂能力之前，我们先通过一个无需外部模型凭据的插件示例来初步建立起 Bub 的插件认知。

初始化插件项目

使用 uv 快速脚手架初始化一个 Python 库项目，并进入项目目录：

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

在 src/bub_echo_plugin/ 源码目录下新建 plugin.py 用于编写插件与 Hook 逻辑。初始化后的整体目录结构如下：

• 文件夹bub-echo-plugin/

  • pyproject.toml
  • 文件夹src/

    • 文件夹bub_echo_plugin/

      • __init__.py
      • plugin.py # 新建：存放插件与 Hook 逻辑

声明入口点

在 pyproject.toml 中，通过 [project.entry-points."bub"] 将插件对象暴露给 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"

注意：

• entry-points 命名：必须为 "bub"，Bub 启动时会自动扫描该命名空间下的所有插件；
• 插件标识与路径：键名 echo 为插件的全局唯一标识；值指向待加载的模块、实例或工厂类。

实现 Hook 逻辑

在模块中编写插件逻辑时，需使用 @hookimpl 装饰器标记具体的 Hook 实现：

from bub import hookimpl


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


echo_plugin = EchoPlugin()

本例通过实现 run_model 以替换了内置的模型调用阶段，直接将输入文本拼接会话 ID 后作为结果返回，便于离线验证整个插件链路。

入口点声明形式

根据插件是否需要维护内部状态或获取宿主上下文，入口点支持三种目标形式：

入口点目标语法	注册对象	适用场景
bub_myplugin.plugin	模块（Module）	Hooks 为纯函数，彼此无状态共享
bub_myplugin.plugin:plugin	已有实例（Instance）	多个 Hooks 之间需要通过 self 维护共享状态
bub_myplugin.plugin:MyPlugin	类或工厂函数（Factory）	插件需要访问全局 BubFramework 实例

当入口点指向类或工厂函数时，Bub 在实例化该对象时会自动将当前的 BubFramework 实例作为第一个参数传入：

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):
        # 通过 framework 访问工作区路径等全局运行时信息
        return f"当前工作区路径：{self.framework.workspace}\n{message}"

安装与验证

编写完插件后，需要将其安装到 Bub 所在的运行环境中，并验证 Hook 的加载与执行顺序。

本地开发安装

在插件项目根目录下执行同步，以可编辑模式（Editable Mode）将其安装至当前虚拟环境：

uv sync

检查 Hook 注册状态

使用 bub hooks 命令检查插件是否已被正确识别：

uv run bub hooks

在终端输出的 Hook 清单中，run_model 下应正确列出已注册的 echo 插件。

运行测试

向应用发送一条测试指令，检验插件的实际生效情况：

uv run bub run "hello from my plugin"

控制台将输出被插件拦截后的回声文本：

[echo:cli:local] hello from my plugin

排错与调度机制

若插件未按预期加载或执行，可依据以下原则排查：

1. 检查运行环境：确认执行 bub 命令行时所在的 Python 环境与插件安装环境完全一致；
2. 重新同步元数据：若修改了 pyproject.toml 中的入口点配置，必须重新执行 uv sync（或 pip install -e .）刷新包的 Entry Points 元数据；
3. 排查 Hook 执行顺序与优先级：Bub 遵循 Pluggy 规范，先注册内置实现，再加载入口点插件。在相同优先级下，后注册的插件优先执行（LIFO 策略）。若需要显式调整执行先后，可在装饰器中指定 @hookimpl(tryfirst=True)（尽可能最先执行）或 @hookimpl(trylast=True)（尽可能最后执行）。

进阶：通过 Hooks 定制行为

Bub 提供了涵盖入站解析、模型调用、工具拦截、状态与生命周期的全套扩展点（签名见 Hooks 参考手册）。在实际业务场景中，最常用的定制需求可分为以下四类：

1. 提供工具

插件可以通过 provide_tools Hook，向 Agent 注册由 @tool 声明的 Python 函数：

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


@tool(name="orders.get")
def get_order(order_id: str) -> dict[str, str]:
    """根据订单 ID 查询订单发货状态。"""
    return {"id": order_id, "status": "shipped"}


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

延迟加载

当插件提供了大量工具时，将所有工具的完整 Schema 一次性注入系统提示词会严重消耗上下文窗口。可以在装饰器中启用延迟加载：

@tool(name="orders.export", deferred=True)
def export_orders(format: str) -> str:
    """导出所有历史订单报表。"""
    ...

注意：

• 延迟机制：带有 deferred=True 的工具默认只向系统提示词提供工具名称与简要描述；
• 动态展开：当模型判断需要调用该工具时，会自动通过内置的 tool.describe 临时加载其完整定义，并从下一步开始执行调用。

2. 流程改写与拦截

改写提示词

• 改写模型输入：build_prompt 接收入站消息、会话 ID 和本轮状态，返回最终提供给模型的输入内容：

  from bub import hookimpl
  from bub.envelope import content_of
  
  
  @hookimpl
  def build_prompt(message, session_id, state):
      text = content_of(message)
      return [f"请使用项目专业术语回答：\n\n{text}"]

• 追加系统指令：如果只想向模型追加全局约束而不改进入站内容，使用 system_prompt（多个插件追加的内容按优先级自动换行拼接）：

  @hookimpl
  def system_prompt(prompt, state):
      return "请在回复末尾附带所参考的代码文件路径。"

拦截调用

拦截 Hook 在调用实际触发前执行。抛出的异常会被框架记录并跳过；若要主动干预、修改或阻断执行，必须返回对应的决策对象（Decision）：

• 限制 LLM 采样参数：

  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)

• 审批或拒绝高危工具调用：

  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("当前会话无权执行撤单操作。")
      return None

放行、改写、拒绝与替代的语义见决策先于副作用；完整返回类型见 Hooks 参考。

调用审计

完成类 Hook 在调用结束后触发，适用于统计耗时、上报监控指标或记录审计日志：

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")

状态持久化

下面通过 load_state 与 save_state 在多个 Turn 之间保存计数。TurnState 的生命周期与数据归属见 Turn 和数据应该放在哪里。

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. 服务与资源挂载

常驻服务

MCP 客户端、通道监听器或定时调度器等需伴随应用常驻运行的组件，继承 Lifecycle 基类并通过 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()

注意：

• 生命周期契约：常驻服务会在 bub gateway 启动或渠道运行时自动初始化，并在退出时执行清理；注意单次执行的 bub run 不会激活常驻服务。
• 命令行过滤：启动时可通过 --enable-channel 单独启用，或使用 --enable-channel '!myapp.lifecycle' 显式排除。

生命周期资源

在整个进程生命周期中持有全局数据库连接池、HTTP 客户端等资源时，可通过异步生成器实现 provide_lifespan：

from bub import hookimpl


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

会话存储

通过 provide_tape_store 可以替换默认的本地文件会话存储（例如接入 PostgreSQL 或云端存储，详见 SQLAlchemy 集成）：

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. 扩展 CLI 命令

插件可以通过 register_cli_commands 直接向 Bub 根 CLI（基于 Typer）注册子命令：

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"执行成功：{name}")

交互式输入

若插件子命令需要在终端中与用户交互确认或获取输入，推荐使用 bub.inquirer 提供的交互式输入工具：

from bub.inquirer import ask_confirm, ask_text

token = ask_text("请输入验证 Token")
if ask_confirm("确认继续执行？", kind="permission"):
    pass

分发与扩展

打包分发

开发完成的插件可以打包为标准 Python Wheel 进行分发。一个发行包可以同时包含插件入口点、技能命名空间包以及自定义 CLI 命令，具体打包配置与发布前验证见打包与分发。跨语言生态包规范见加载 Agent Plugins。

WASM 插件（实验性）

借助 bub-extism 扩展，可以通过 WebAssembly 将 Go、Rust 等其他语言编译的二进制模块直接挂载为 Bub Hook 处理函数。

extism.json 指定模块文件和 hook 映射：

{
  "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."