复用应用插件

你可以在现有 Python 服务中使用 Bub 应用插件。宿主继续管理 HTTP 路由、任务入口和输出处理；根据要复用的应用行为，选择 Agent 和运行入口。

BubAgent 已经将内核循环、设置、插件 hooks、skills 和工具策略组合在一起，它的直接流入口也可以嵌入宿主程序。需要完整消息处理流程时，使用 BubFramework.process_inbound()。

使用 core.Agent 时，由你选择组件并接入 hooks。下面分别取用搜索插件的工具、适配上下文插件的调用 hooks，每个示例都说明了该插件需要的依赖。后台服务还需要各自的启动和关闭流程。

给 Agent 加入搜索插件

bub-web-search 提供了可供 core.Agent 使用的搜索工具。将它安装到运行服务的 Python 环境中。本文示例使用 bub-contrib 的 dev 分支插件：

uv add "bub[app]" \
  "bub-web-search @ git+https://github.com/bubbuild/bub-contrib.git@dev#subdirectory=packages/bub-web-search"

插件使用应用配置代码，因此需要 app 依赖。你的程序仍可使用 core.Agent，沿用自己的模型、Tape 名称和运行生命周期。

使用 Jina 搜索时，在启动进程前设置 BUB_SEARCH_PROVIDER=jina 和 BUB_SEARCH_JINA_API_KEY。示例中的模型还需要 REPUBLIC_OPENAI_API_KEY。将以下代码保存为 search_agent.py：

import asyncio
from pathlib import Path

import republic
from bub_web_search.tools import register_tools

from bub.core import Agent
from bub.core.store import FileTapeStore


async def main() -> None:
    search = register_tools()
    if search is None:
        raise RuntimeError("Configure a search provider with BUB_SEARCH_* first.")

    agent = Agent(
        store=FileTapeStore(Path("sessions")),
        tools=[search],
        system_prompt="Search for sources before answering. Include their URLs.",
    )
    async with republic.get_model("openai:gpt-5-mini") as model:
        async with agent.run(
            "Find the official Python asyncio documentation.",
            model=model,
            tape="research-42",
            options={"max_tokens": 4096},
        ) as stream:
            async for event in stream:
                if event.kind == "text":
                    print(event.data.get("delta", ""), end="", flush=True)
                elif event.kind == "error":
                    raise RuntimeError(str(event.data.get("message", "Agent failed")))
        print()


if __name__ == "__main__":
    asyncio.run(main())

运行 uv run python search_agent.py。插件负责搜索请求及其 HTTP 连接；Agent 将会话记录到 sessions/，你的代码负责消费回复。

如果已有 Agent，在启动时执行 agent.tools[search.name] = search 即可加入返回的工具。它的名称是 web.search，模型看到的名称是 web_search。其余调用代码可以沿用，包括现有的 FastAPI 接口。

register_tools() 还会更新进程内的全局工具注册表，请在服务启动时完成这些准备。Agent 使用显式传入或加入的工具；本例取用返回的搜索工具，包注册的其他工具（如 Jina 的 web.read）可按需另行加入。

进阶：接入单个插件的调用 hooks

bub-context-budget 在模型调用前后工作。当估算的上下文超过阈值时，它会加入一条指令，请模型调用 tape.handoff。将这个工具交给 Agent，再把插件 hooks 接到 Agent 事件上即可。

uv add "bub-context-budget @ git+https://github.com/bubbuild/bub-contrib.git@dev#subdirectory=packages/bub-context-budget"

在程序中加入下面的函数。创建 Agent 后、首次运行前，调用 add_context_budget(agent)：

from pathlib import Path

from bub_context_budget import plugin as context_budget

from bub import BubFramework
from bub.builtin.tools import tape_handoff
from bub.core import Agent
from bub.hooks.agent import AgentPlugins


def add_context_budget(agent: Agent) -> None:
    framework = BubFramework(config_file=Path("bub.yml"))
    AgentPlugins(framework, plugins=[context_budget]).attach(agent)
    agent.tools[tape_handoff.name] = tape_handoff

适配器把这个插件的 before_llm_call 和 after_llm_call 接到现有 Agent 上，沿用原来的模型、提示词片段、存储和 agent.run()。本例接好了 context-budget 所需的 hooks 和 handoff 工具；适配其他插件时，需要分别核对它们的要求。

通过 BUB_CONTEXT_BUDGET_MAX_CONTEXT_TOKENS，或 bub.yml 中的 context-budget.max_context_tokens 设置上下文上限。插件从该上限中预留 16,384 tokens，再决定何时请求 handoff。这是一条给模型的预算提醒，实际执行取决于模型是否遵循指令。

Handoff 把摘要保存在锚点上。下一次模型请求需要这份摘要时，通过上下文选择或下一条提示词将它带入。

注册和接入在服务启动时各执行一次，这个插件通过调用 hooks 完成工作。插件注册属于 Framework，attach() 将其 agent 循环 hooks 接到选定 Agent 的事件上；使用该 Agent 的会话共用这些 hooks。需要不同插件集合时，分别创建 Framework，配置仍是进程级的。

适配器会捕获普通插件异常并跳过失败的 hook。需要阻止工具执行时，hook 应返回 ToolCallDecision.deny(...)。具体拦截语义见 Agent 事件。

接入插件的其他 agent 循环 hooks

AgentPlugins 接入所有影响 agent 循环的 hooks，不只是调用拦截。每个 hook 对应核心 Agent 的一个扩展点：

Hook	核心 Agent 扩展点
`before_llm_call`、`after_llm_call`、`before_tool_call`、`after_tool_call`	调用事件，通过 `attach(agent)` 接入。
`continue_prompt`	`status == "continue"` 的 `StepFinished`：hook 返回的文本成为下一步的用户消息。
`provide_steering_inbox`	每次模型请求前和步骤结束时，`MessagesPending` 取出 inbox 中的消息；取到消息时运行继续。
`system_prompt`	`attach(agent)` 在 Agent 已有片段之后添加的提示词片段。
`provide_tools`	`attach(agent)` 添加的工具；Agent 已有同名工具时保留原工具。
`provide_tape_store`、`build_tape_context`、`provide_tape_sidecar`	`plugins.components()` 中的 `store`、`context` 和 `sidecars`。
`load_state`、`provide_environment`	由 `await plugins.build_state(session_id)` 生成的运行状态。

from bub.hooks.agent import AgentPlugins

plugins = AgentPlugins(plugins=[SupportPlugin()])

async with plugins.running():
    agent = plugins.attach(Agent(system_prompt="Answer in plain text.", **plugins.components()))
    state = await plugins.build_state("customer-42", agent=agent)
    async with agent.run(prompt, model=model, tape="customer-42", state=state) as stream:
        ...

没有 hook 提供的组件不会出现在 components() 中，Agent 保留默认值；你自己传入的参数会替代它们。hooks 提供的存储和 steering inbox 只在 plugins.running()（即 framework.running()）内可用。先调用 attach()，再注册自己的处理器，这样它们能看到 hooks 的修改。

build_prompt、run_model、save_state、render_outbound、dispatch_outbound 等消息流程 hooks 不属于 agent 循环，只在 process_inbound() 中运行。plugins=[...] 只注册列出的插件。不带参数的 AgentPlugins() 和 bub 命令一样，加载 Bub 内置 hooks 和已安装的插件；内置系统提示词要求 agent 通过 channel 回复。传入 framework 则使用已有 framework 的 hooks，同时传入的 plugins 会注册到它上面。

按插件能力选择接法

宿主需要提供什么，取决于插件实现了哪些能力：

插件能力	接入方式
工具或存储	导入公开组件并传给 Agent。例如，`bub_tapestore_sqlalchemy.store.SQLAlchemyTapeStore` 接收数据库 URL，可作为 `store=` 传入。
模型或工具拦截、系统提示词、继续运行或 steering	注册指定插件并接入 `AgentPlugins`，同时核对插件需要的状态字段和工具。
消息准备、模型阶段替换、状态保存或消息发送	在宿主中嵌入应用流程，执行这些阶段并取得 `TurnResult`。
后台连接或动态发现的工具	启动服务、等待就绪、绑定工具，并随宿主关闭。具体顺序遵循插件自己的生命周期接口。

按所选存储的生命周期约定管理资源。framework.running() 会获取 hooks 提供的资源，并在退出时关闭它进入的上下文管理器。返回普通对象的存储仍需遵循自己的清理约定。

接入 MCP 时，按照插件的嵌入示例，用 MCPChannel.connect() 建立连接，通过 bind_agent() 把工具绑定到 BubAgent，最后调用 stop() 关闭连接。宿主在使用 framework.running() 之外，还需要管理这段生命周期。

复用 Bub 应用的默认行为

BubAgent 在内核循环上接入 Bub 设置、skills、工具策略和插件 hooks。宿主可以通过 framework.load_hooks() 加载已安装插件，进入 framework.running() 后构造 BubAgent。它的 run_stream() 返回由宿主消费和关闭的流。

如果也需要这些应用默认行为，可以选择这种接法。内置系统提示词要求 Agent 通过渠道发送回复；直接返回模型文本的服务需要明确自己的提示词策略。替换 agent.system_prompt 会替换所有片段，包括插件、skills、延迟工具和 CodeMode 的指令。

run_stream() 会使用系统提示词、继续提示词、调用拦截 hooks，以及配置好的 Tape 组件。省略 state 时，它通过一条只包含 Agent 引用的消息加载状态。插件需要请求元数据时，用 framework.build_state(message, session_id) 准备状态，再传给 run_stream(state=...)。在 message["_runtime_agent"] 中放入 Agent，让状态恢复使用它的存储。

直接调用流入口会跳过初始 build_prompt、run_model / run_model_stream、save_state、渲染和分发 hooks。宿主负责输出、自有状态的保存，以及同一会话的并发协调。需要完整消息流程时，使用 process_inbound()；即使设置 stream_output=True，它也会自行消费流，再返回完整的 TurnResult。

继续构建

工具与 CodeMode：Agent 如何使用你提供的工具。
Tape 与上下文：存储与 handoff。
打包插件：将自己的扩展分享给其他人。