复用 Bub 应用

把 Bub 嵌入现有服务时，最重要的不是“哪个入口功能最多”，而是哪些组装和生命周期仍由宿主负责。你可以只拿一个工具，也可以让 Bub 处理完整消息 Turn。

本页的应用层入口需要 app 可选依赖，并配置可用模型凭据（例如 REPUBLIC_OPENAI_API_KEY）：

uv add "bub[app]"

如果只需要模型循环、工具和 Tape，从核心 SDK开始即可。

先选择复用层级

需要复用的能力	入口	宿主仍负责
一个工具实现	导入公开工具，传给 core.Agent	模型、Tape、Run 与资源生命周期
插件的 Agent 循环 hooks	AgentPlugins.attach()	驱动 Run、入站与出站消息、同会话并发
应用配置、技能、工具策略和命令	BubAgent.run_stream()	消费与关闭流、回复投递、完整 Turn 状态、同会话并发
从入站消息到出站结果的完整 Turn	BubFramework.process_inbound()	服务生命周期、模型配置、异常、并发，以及未委托渠道的投递
现成的命令行或网关	bub run / bub chat / bub gateway，或自定义入口点	配置、凭据、部署与扩展，不再编写应用循环

这些是并列入口，不是必须逐级升级的流程。从能满足需求的最小层级开始。

注意

安装一个插件包不等于每个入口都会执行它的全部能力。工具和 Agent hooks 可以被较低层级复用；渠道、CLI 和完整消息管道只有在相应运行时中才会执行。插件还可能依赖其他 hooks、状态字段或后台服务，应以其文档为准。

推荐起点：执行完整 Turn

process_inbound() 适合希望 Bub 负责会话解析、初始提示词、模型阶段、状态保存和出站消息组装的宿主。

创建 embedded.py：

import asyncio

from bub import BubFramework
from bub.envelope import content_of


async def main() -> None:
    framework = BubFramework()
    framework.load_hooks()

    async with framework.running():
        result = await framework.process_inbound({
            "session_id": "embedded:customer-42",
            "channel": "embedded",
            "chat_id": "customer-42",
            "content": "请说明你能帮我处理哪些订单问题。",
        })

    print(result.session_id)
    for outbound in result.outbounds:
        print(content_of(outbound))


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

运行：

uv run python embedded.py

load_hooks() 在启动时调用一次，加载内置实现和已安装插件。framework.running() 持有插件生命周期、Tape 存储和 SteeringInbox 等资源，应覆盖整个嵌入服务的运行期，而不是每个请求进入一次。

process_inbound() 返回完整的 TurnResult，不是事件迭代器。本例不绑定发送渠道，因此宿主读取 result.outbounds 并自行投递。如果安装了会发送消息的 dispatch_outbound 插件，要避免插件与宿主重复投递。

设置 stream_output=True 会选择并消费流式模型路径，完成后仍返回 TurnResult。绑定 Router 时可以用 wrap_stream() 包装事件流；不绑定 Router 也能使用 stream_output=True。

只复用模型阶段：BubAgent.run_stream

需要 Bub 的配置、技能、工具策略和命令，但希望自己处理事件流时，使用 BubAgent.run_stream()。

run_stream() 返回 AsyncStreamEvents。它没有异步上下文管理器协议，应使用 contextlib.aclosing() 保证提前退出和异常时也关闭：

import asyncio
import contextlib

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


async def main() -> None:
    framework = BubFramework()
    framework.load_hooks()

    async with framework.running():
        agent = BubAgent(framework)
        stream = await agent.run_stream(
            session_id="customer-42",
            prompt="请直接用纯文本回答：你能处理哪些订单问题？",
            allowed_tools=[],
            allowed_skills=[],
        )

        async with contextlib.aclosing(stream):
            async for event in stream:
                if event.kind == "text":
                    print(event.data["delta"], end="", flush=True)
        print()


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

在 framework.running() 内创建 BubAgent，它才能取得该作用域提供的存储。一个 Agent 可以跨会话复用；按 Turn 变化的工具、技能和状态应放在调用参数或 TurnState 中。

与完整 Turn 相比，这个直接入口：

• 仍会解析会话 Tape、绑定运行时 Agent，并在 state=None 时通过 hooks 加载状态；
• 仍可执行直接命令；
• 不执行初始入站 build_prompt、save_state、出站渲染和分发；
• 不协调同会话并发；
• 不负责替你关闭流，第一个 final 也只表示一个 Step 结束。

内置系统提示不会要求模型通过 channel skill 回复；bub gateway 和 bub chat 背后的渠道管理器会注册一个 system_prompt Hook，只在它运行的 Turn 中加入这条指令。自己调用 process_inbound() 的服务可以注册自己的 Hook，说明回复如何送达。整体替换 agent.system_prompt 也会移除插件、技能、延迟工具和 代码模式 的提示片段。

按调用收窄工具和技能

import asyncio
import contextlib

from bub import BubFramework
from bub.builtin.agent import BubAgent
from bub.core.tools import Tool


def lookup_order(order_id: str) -> str:
    return f"订单 {order_id}：处理中。"


async def main() -> None:
    framework = BubFramework()
    framework.load_hooks()

    async with framework.running():
        order_tool = Tool.from_callable(
            lookup_order,
            name="orders.get",
            description="按订单号查询示例订单。",
        )
        agent = BubAgent(framework, tools=[order_tool])
        stream = await agent.run_stream(
            session_id="customer-42",
            prompt="查询订单 A123。",
            allowed_tools={"orders.get"},
            allowed_skills=[],
        )
        async with contextlib.aclosing(stream):
            async for event in stream:
                if event.kind == "text":
                    print(event.data["delta"], end="", flush=True)


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

参数	None	空集合
allowed_tools	允许实例上的全部工具	不允许工具
allowed_skills	使用默认技能发现结果	禁用技能发现

未知名称会抛出 ValueError。筛选覆盖模型直接调用和 代码模式，但逗号命令直接使用 Agent 实例的工具，不受 allowed_tools 限制；需要约束所有执行路径时，在 before_tool_call 中执行权限策略。

这些限制只有在宿主代码决定参数时才可靠。让模型自己生成子 Agent 的允许列表不构成安全边界。

显式传入 state 会跳过 load_state hooks。需要正常恢复会话又要增加请求元数据时，使用 framework.build_state() 构造状态，而不是用一个新字典替换全部状态。完整的 _runtime_agent 和状态形式见 Python API。

只复用工具或 Agent hooks

工具实现

已有 core.Agent 时，可以直接导入内置工具或插件提供的公开构造函数。下面用 bub-web-search 演示只复用搜索工具。先安装依赖并配置搜索与模型凭据：

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

export BUB_SEARCH_PROVIDER=jina
export BUB_SEARCH_JINA_API_KEY=jina_...
export REPUBLIC_OPENAI_API_KEY=sk-...

将下面脚本保存为 search_agent.py：

import asyncio
from pathlib import Path

import republic
from bub_web_search.tools import build_tools

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


async def main() -> None:
    search_tools = build_tools()
    search = search_tools.get("web.search")
    if search is None:
        raise RuntimeError("Configure BUB_SEARCH_PROVIDER and BUB_SEARCH_JINA_API_KEY first.")

    agent = Agent(
        store=FileTapeStore(Path("sessions")),
        tools=[search],
        system_prompt="先搜索可靠来源，再用中文回答，并附上来源链接。",
    )
    async with republic.get_model("openai:gpt-5-mini") as model:
        async with agent.run(
            "查找 Python asyncio 官方文档。",
            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 会把会话记录到 sessions/。这里只复用了选中的搜索工具；模型、Tape、事件流和资源生命周期仍由宿主管理。更多插件说明见 bub-web-search。

Agent 循环 hooks

AgentPlugins 把插件 hooks 接到核心 Agent 的事件、提示词、工具和运行状态上。这些机制分别建立在哪种数据之上，见 Agent 事件：事件与其他扩展机制。

下面的 PlainTextPlugin 是一个完整的最小插件示例。将代码保存为 agent_hooks.py，并确保已安装 bub[app]、配置 REPUBLIC_OPENAI_API_KEY：

import asyncio

import republic

from bub.core import Agent
from bub.core.store import InMemoryTapeStore
from bub.hooks import hookimpl
from bub.hooks.agent import AgentPlugins
from bub.turn import TurnState


class PlainTextPlugin:
    """为 Agent 添加一条系统提示词。"""

    @hookimpl
    def system_prompt(self, prompt: str, state: TurnState) -> str:
        return "用简短、清楚的中文回答。"


async def main() -> None:
    plugins = AgentPlugins(plugins=[PlainTextPlugin()])
    async with plugins.running():
        components = {**plugins.components(), "store": InMemoryTapeStore()}
        agent = plugins.attach(Agent(system_prompt="回答用户的问题。", **components))
        state = await plugins.build_state("customer-42", agent=agent)

        async with republic.get_model("openai:gpt-5-mini") as model:
            async with agent.run("你能做什么？", model=model, tape="customer-42", state=state) as stream:
                async for event in stream:
                    if event.kind == "text":
                        print(event.data.get("delta", ""), end="", flush=True)
        print()


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

示例中的 PlainTextPlugin 是最小的自定义插件：它通过 system_prompt Hook 为 Agent 增加提示词。AgentPlugins(plugins=[...]) 只注册列出的插件，不会自动加载 Bub 内置 Hook 或已安装插件；需要它们时，创建 BubFramework、调用 load_hooks() 并将它传给 AgentPlugins。

运行示例：uv run python agent_hooks.py。

components() 返回普通 kwargs 字典，不会自动解决重复参数。示例用字典合并显式覆盖 store；若直接同时传入同名关键字会抛出异常。

存储和 SteeringInbox 等资源只在 plugins.running() 内可用。先 attach()，再注册自己的事件处理器，后者才能观察 hooks 对事件的修改。

attach() 连接的是 Agent 循环，不会自动提供完整 Turn 的初始 build_prompt、run_model / run_model_stream、save_state、出站渲染与分发。steering 等特定路径仍可能调用 build_prompt，内置错误报告也可能调用 dispatch_outbound，所以插件宿主应阅读具体 hook 的执行时机。

例如，bub-context-budget 通过模型调用 hooks 提醒模型在上下文接近上限时调用 tape.handoff。接入时除了 attach()，还要把 tape.handoff 工具提供给 Agent；这个提醒本身不强制执行压缩。此例说明插件常有 attach() 以外的依赖，具体参数见 tape.handoff 参考。

MCP 等需要常驻连接的插件还有独立生命周期；attach() 不能代替连接、启动和关闭。嵌入方式见 bub-mcp。

需要渠道投递时绑定 Router

宿主已有发送逻辑时，直接使用 TurnResult.outbounds 即可。需要 Bub 渠道负责出站和流包装时，实现并绑定 ChannelRouter：

• dispatch_output() 执行实际投递；返回 False 不会触发自动重试或备用投递，process_inbound() 仍返回渲染后的 outbounds。
• wrap_stream() 必须返回异步可迭代对象，并保留希望 Bub 继续处理的模型事件；model_output 由包装后流中的文本增量聚合。
• admit_channel_message() 属于 ChannelManager 的准入路径。直接调用 process_inbound() 不会自动执行准入或会话排队。

接口签名见 Python API 与渠道概念。

生命周期、异常与并发清单

无论选择哪个应用层入口，都应检查：

1. 启动一次：load_hooks() 在服务组装阶段调用，不在每个请求中重复发现插件。
2. 一个运行作用域：同一 Framework 不要重叠进入多个 running()；ChannelManager.listen_and_run() 已会进入作用域，不要再套一层。
3. 配置不是隔离沙箱：不同 Framework 可以隔离插件集合，应用配置仍是进程级缓存。
4. 异常由宿主接住：run_stream() 的迭代和 process_inbound() 都可能抛错；完整 Turn 的错误钩子不会代替宿主异常处理。
5. 同会话显式协调：两个直接入口都不会按 session_id 自动串行化。使用宿主锁，或通过 ChannelManager 的 follow_up 准入策略排队。
6. 关闭流：正常消费到结束，提前退出时用 aclosing()；temp/ 前缀会话不合并 Tape，普通会话关闭时默认合并，包括部分记录。
7. 副作用不是事务：后续模型失败、取消或丢弃 Tape 都不会回滚工具已经产生的外部副作用。

直接运行或发布命令行

不需要嵌入代码时，直接使用 bub run、bub chat 或 bub gateway。需要自己的命令名时，在发行包中声明 [project.scripts] 调用 Bub 入口，并随包分发插件和技能；完整流程见打包与分发。