From ac8d257e46d8f9542db54264f6616fa5ddb93c17 Mon Sep 17 00:00:00 2001 From: Frost Ming Date: Tue, 15 Sep 2026 19:10:45 +0800 Subject: [PATCH 1/3] feat: enhance Agent and framework with improved tool management and session recovery Signed-off-by: Frost Ming --- src/bub/builtin/__init__.py | 3 + src/bub/builtin/agent.py | 61 ++++++++++++------- src/bub/builtin/hook_impl.py | 30 ++++++---- src/bub/builtin/tools.py | 30 ++++++---- src/bub/framework.py | 32 ++++------ src/bub/skills.py | 13 ++-- tests/test_builtin_agent.py | 3 + tests/test_builtin_cli.py | 4 +- tests/test_builtin_hook_impl.py | 4 +- tests/test_framework.py | 84 +++++++------------------- tests/test_sdk_isolation.py | 103 ++++++++++++++++++++++++++++++++ tests/test_skills.py | 2 +- tests/test_subagent_tool.py | 1 + 13 files changed, 235 insertions(+), 135 deletions(-) create mode 100644 tests/test_sdk_isolation.py diff --git a/src/bub/builtin/__init__.py b/src/bub/builtin/__init__.py index e69de29b..a6df24ad 100644 --- a/src/bub/builtin/__init__.py +++ b/src/bub/builtin/__init__.py @@ -0,0 +1,3 @@ +from .agent import Agent + +__all__ = ["Agent"] diff --git a/src/bub/builtin/agent.py b/src/bub/builtin/agent.py index 3924c4d9..c2640bd3 100644 --- a/src/bub/builtin/agent.py +++ b/src/bub/builtin/agent.py @@ -25,15 +25,10 @@ from bub.envelope import field_of from bub.framework import BubFramework from bub.skills import discover_skills, render_skills_prompt -from bub.store import AsyncTapeStoreAdapter, InMemoryTapeStore, is_async_tape_store +from bub.store import AsyncTapeStore, AsyncTapeStoreAdapter, InMemoryTapeStore, TapeStore, is_async_tape_store from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState from bub.tape import Tape -from bub.tools import ( - REGISTRY, - Tool, - ToolContext, - model_tools, -) +from bub.tools import REGISTRY, Tool, ToolContext, model_tools from bub.turn import TurnState from bub.utils import workspace_from_state @@ -44,18 +39,32 @@ class Agent: """Agent that processes prompts using hooks, tools, tape, and any-llm-sdk.""" - def __init__(self, framework: BubFramework) -> None: + def __init__( + self, + framework: BubFramework, + *, + tools: Collection[Tool] | None = None, + tape_store: TapeStore | AsyncTapeStore | None = None, + skill_dirs: Collection[Path] | None = None, + ) -> None: self.settings = load_settings() self.framework = framework + self.tools = {tool.name: tool for tool in tools} if tools is not None else REGISTRY.copy() + self.tape_store = tape_store + self.skill_dirs = skill_dirs self.model_runner = ModelRunner(self.settings, hooks=framework.get_agent_hooks()) @cached_property def tape(self) -> Tape: import bub - tape_store = self.framework.get_tape_store() - if tape_store is None: - tape_store = InMemoryTapeStore() + tape_store: TapeStore | AsyncTapeStore | None + if self.tape_store is not None: + tape_store = self.tape_store + else: + tape_store = self.framework.get_tape_store() + if tape_store is None: + tape_store = InMemoryTapeStore() if not is_async_tape_store(tape_store): tape_store = AsyncTapeStoreAdapter(tape_store) return Tape( @@ -91,10 +100,11 @@ async def run_stream( *, session_id: str, prompt: str | list[dict], - state: TurnState, + state: TurnState | None = None, model: str | None = None, allowed_skills: Collection[str] | None = None, allowed_tools: Collection[str] | None = None, + reasoning_effort: str | None = None, ) -> AsyncStreamEvents: if not prompt: return self._events_from_iterable([ @@ -102,6 +112,13 @@ async def run_stream( StreamEvent("final", {"text": "error: empty prompt", "ok": False}), ]) + if state is None: + state = await self.framework.build_state({"_runtime_agent": self}, session_id) + state["_runtime_agent"] = self # Override the agent to the current instance. + if model is None: + model = state.get("model") + if reasoning_effort is not None: + state["reasoning_effort"] = reasoning_effort state.setdefault("session_id", session_id) tape = self.tape.session_tape( session_id, workspace_from_state(state), context=replace(self.tape.context, state=state) @@ -139,13 +156,15 @@ async def _run_command(self, tape: Tape, *, line: str) -> str: output = "" status = "ok" try: - if name not in REGISTRY: - output = await REGISTRY["bash"].run(context=context, cmd=line) + if name not in self.tools: + if "bash" not in self.tools: + raise ValueError("bash tool is not available") # noqa: TRY301 + output = await self.tools["bash"].run(context=context, cmd=line) else: args = _parse_args(arg_tokens) - if REGISTRY[name].context: + if self.tools[name].context: args.kwargs["context"] = context - output = REGISTRY[name].run(*args.positional, **args.kwargs) + output = self.tools[name].run(*args.positional, **args.kwargs) if inspect.isawaitable(output): output = await output except Exception as exc: @@ -311,7 +330,7 @@ async def _stream_events_with_auto_handoff( def _load_skills_prompt(self, prompt: str, workspace: Path, allowed_skills: set[str] | None = None) -> str: skill_index = { skill.name.casefold(): skill - for skill in discover_skills(workspace) + for skill in discover_skills(workspace, skill_dirs=self.skill_dirs) if allowed_skills is None or skill.name.casefold() in allowed_skills } expanded_skills = set(HINT_RE.findall(prompt)) & set(skill_index.keys()) @@ -330,14 +349,14 @@ async def _run_once( if allowed_tools is not None: from bub.builtin.tools import resolve_tool_names - allowed_tools = resolve_tool_names(allowed_tools) + allowed_tools = resolve_tool_names(allowed_tools, all_names=self.tools) if allowed_skills is not None: allowed_skills = {name.casefold() for name in allowed_skills} tape.context.state["allowed_skills"] = list(allowed_skills) if allowed_tools is not None: - tools = [tool for tool in REGISTRY.values() if tool.name in allowed_tools] + tools = [tool for tool in self.tools.values() if tool.name in allowed_tools] else: - tools = list(REGISTRY.values()) + tools = list(self.tools.values()) return await self._run_once_stream( tape=tape, prompt=prompt, @@ -394,7 +413,7 @@ def _system_prompt( blocks: list[str] = [] if result := self.framework.get_system_prompt(prompt=prompt, state=state): blocks.append(result) - tools_prompt = render_tools_prompt(tools if tools is not None else REGISTRY.values()) + tools_prompt = render_tools_prompt(tools if tools is not None else self.tools.values()) if tools_prompt: blocks.append(tools_prompt) workspace = workspace_from_state(state) diff --git a/src/bub/builtin/hook_impl.py b/src/bub/builtin/hook_impl.py index 75c48843..df268e8d 100644 --- a/src/bub/builtin/hook_impl.py +++ b/src/bub/builtin/hook_impl.py @@ -83,12 +83,14 @@ def __init__(self, framework: BubFramework) -> None: self.framework = framework self._agent: Agent | None = None - def _get_agent(self) -> Agent: + def _get_agent(self, state: TurnState | None = None) -> Agent: + if state and "_runtime_agent" in state: + return cast("Agent", state["_runtime_agent"]) if self._agent is None: self._agent = Agent(self.framework) return self._agent - async def _recover_session_model(self, session_id: str) -> str | None: + async def _recover_session_model(self, session_id: str, *, agent: Agent) -> str | None: """Recover the latest per-session model override recorded on the session tape. The ``model`` tool records each switch as a ``model_switch`` event on the @@ -97,7 +99,7 @@ async def _recover_session_model(self, session_id: str) -> str | None: restored. Returns ``None`` when nothing was recorded, so a fresh session never inherits another session's model. """ - session = self._get_agent().tape.session_tape(session_id, self.framework.workspace) + session = agent.tape.session_tape(session_id, self.framework.workspace) entries = list(await session.store.fetch_all(session.query().kinds("event"))) for entry in reversed(entries): if entry.kind == "event" and entry.payload.get("name") == "model_switch": @@ -105,9 +107,9 @@ async def _recover_session_model(self, session_id: str) -> str | None: return str(model) if model else None return None - async def _recover_session_reasoning_effort(self, session_id: str) -> str | None: + async def _recover_session_reasoning_effort(self, session_id: str, *, agent: Agent) -> str | None: """Recover the latest per-session reasoning effort override.""" - session = self._get_agent().tape.session_tape(session_id, self.framework.workspace) + session = agent.tape.session_tape(session_id, self.framework.workspace) entries = list(await session.store.fetch_all(session.query().kinds("event"))) for entry in reversed(entries): if entry.kind == "event" and entry.payload.get("name") == "reasoning_effort_switch": @@ -166,15 +168,19 @@ async def load_state(self, message: ChannelMessage, session_id: str) -> TurnStat lifespan = field_of(message, "lifespan") if lifespan is not None: await lifespan.__aenter__() - state = {"session_id": session_id, "_runtime_agent": self._get_agent()} + # SDK calls supply their agent before recovery so state comes from its store. + agent = field_of(message, "_runtime_agent") + if agent is None: + agent = self._get_agent() + state = {"session_id": session_id, "_runtime_agent": agent} if context := field_of(message, "context_str"): state["context"] = context # Carry over a previously recorded per-session model override from the # session tape. Only set when a prior turn actually recorded one, so a # fresh/unknown session never inherits another session's model. - if model := await self._recover_session_model(session_id): + if model := await self._recover_session_model(session_id, agent=agent): state["model"] = model - if reasoning_effort := await self._recover_session_reasoning_effort(session_id): + if reasoning_effort := await self._recover_session_reasoning_effort(session_id, agent=agent): state["reasoning_effort"] = reasoning_effort if model := field_of(message, "context", {}).get("model"): state["model"] = model @@ -227,7 +233,7 @@ async def build_prompt(self, message: ChannelMessage, session_id: str, state: Tu @hookimpl async def run_model_stream(self, prompt: str | list[dict], session_id: str, state: TurnState) -> AsyncStreamEvents: - return await self._get_agent().run_stream( + return await self._get_agent(state).run_stream( session_id=session_id, prompt=prompt, state=state, @@ -425,9 +431,11 @@ async def before_tool_call( replace it with a guidance ``tool_result`` so the model can re-issue a valid call on the next step. """ - from bub.tools import REGISTRY, model_tools + from bub.tools import model_tools - available_tools = tuple(tool_item.name for tool_item in model_tools(REGISTRY.values())) + agent = self._get_agent(state) + + available_tools = tuple(tool_item.name for tool_item in model_tools(agent.tools.values())) if call.tool in available_tools: return None diff --git a/src/bub/builtin/tools.py b/src/bub/builtin/tools.py index 862a88f0..ad63df0f 100644 --- a/src/bub/builtin/tools.py +++ b/src/bub/builtin/tools.py @@ -25,9 +25,10 @@ def _to_model_name(name: str) -> str: return name.replace(".", "_") -def _tool_name_index() -> dict[str, str]: - real_names = {tool_name.casefold(): tool_name for tool_name in REGISTRY} - alias_names = {_to_model_name(tool_name).casefold(): tool_name for tool_name in REGISTRY} +def _tool_name_index(all_names: Iterable[str]) -> dict[str, str]: + names = tuple(all_names) + real_names = {tool_name.casefold(): tool_name for tool_name in names} + alias_names = {_to_model_name(tool_name).casefold(): tool_name for tool_name in names} return {**alias_names, **real_names} @@ -36,15 +37,15 @@ def resolve_tool_name(name: str) -> str | None: key = name.strip().casefold() if not key: return None - return _tool_name_index().get(key) + return _tool_name_index(REGISTRY).get(key) -def _resolve_explicit_tool_names(names: Iterable[str]) -> tuple[set[str], set[str]]: +def _resolve_explicit_tool_names(names: Iterable[str], index: dict[str, str]) -> tuple[set[str], set[str]]: resolved: set[str] = set() unknown: set[str] = set() for name in names: normalized_name = name.strip() - if resolved_name := resolve_tool_name(normalized_name): + if resolved_name := index.get(normalized_name.casefold()): resolved.add(resolved_name) else: unknown.add(normalized_name) @@ -56,15 +57,19 @@ def _raise_unknown_tool_names(names: set[str]) -> None: raise ValueError(f"unknown tool name(s): {formatted}") -def resolve_tool_names(names: Iterable[str] | None = None, *, exclude: Iterable[str] = ()) -> set[str]: +def resolve_tool_names( + names: Iterable[str] | None = None, *, exclude: Iterable[str] = (), all_names: Iterable[str] | None = None +) -> set[str]: """Resolve tool names from either runtime names or model-facing aliases.""" - excluded, unknown_excluded = _resolve_explicit_tool_names(exclude) + available = tuple(REGISTRY if all_names is None else all_names) + index = _tool_name_index(available) + excluded, unknown_excluded = _resolve_explicit_tool_names(exclude, index) if unknown_excluded: _raise_unknown_tool_names(unknown_excluded) if names is None: - return set(REGISTRY) - excluded + return set(available) - excluded - resolved, unknown = _resolve_explicit_tool_names(names) + resolved, unknown = _resolve_explicit_tool_names(names, index) if unknown: _raise_unknown_tool_names(unknown) return resolved - excluded @@ -239,12 +244,13 @@ def skill_describe(name: str | None = None, *, context: ToolContext) -> str: """ from bub.utils import workspace_from_state + agent = _get_agent(context) allowed_skills = context.state.get("allowed_skills") if allowed_skills is not None and name and name.casefold() not in allowed_skills: return f"(skill '{name}' is not allowed in this context)" workspace = workspace_from_state(context.state) - skill_index = {skill.name: skill for skill in discover_skills(workspace)} + skill_index = {skill.name: skill for skill in discover_skills(workspace, skill_dirs=agent.skill_dirs)} if name is None: return "Available skills:\n" + "\n".join(f"- {skill.name}" for skill in skill_index.values()) if name.casefold() not in skill_index: @@ -338,7 +344,7 @@ async def run_subagent(param: SubAgentInput, *, context: ToolContext) -> str: else: subagent_session = param.session state = {**context.state, "session_id": subagent_session} - allowed_tools = resolve_tool_names(param.allowed_tools or None, exclude={"subagent"}) + allowed_tools = resolve_tool_names(param.allowed_tools or None, exclude={"subagent"}, all_names=agent.tools) output = "" async for event in await agent.run_stream( session_id=subagent_session, diff --git a/src/bub/framework.py b/src/bub/framework.py index 72aca545..7911c486 100644 --- a/src/bub/framework.py +++ b/src/bub/framework.py @@ -49,9 +49,9 @@ class BubFramework: def __init__(self, config_file: Path = DEFAULT_CONFIG_FILE) -> None: self.workspace = Path.cwd().resolve() self.config_file = config_file.resolve() - self._plugin_manager = pluggy.PluginManager(BUB_HOOK_NAMESPACE) - self._plugin_manager.add_hookspecs(BubHookSpecs) - self._hook_runtime = HookRuntime(self._plugin_manager) + self.plugin_manager = pluggy.PluginManager(BUB_HOOK_NAMESPACE) + self.plugin_manager.add_hookspecs(BubHookSpecs) + self._hook_runtime = HookRuntime(self.plugin_manager) self._agent_hooks = AgentHooks(self._hook_runtime) self._plugin_status: dict[str, PluginStatus] = {} self._channel_router: ChannelRouter | None = None @@ -60,13 +60,13 @@ def __init__(self, config_file: Path = DEFAULT_CONFIG_FILE) -> None: configure.load(self.config_file) def load_builtin_hooks(self) -> None: - """Register Bub's builtin hook implementations.""" + """Load Bub's builtin hook implementations.""" from bub.builtin.hook_impl import BuiltinImpl impl = BuiltinImpl(self) try: - self._plugin_manager.register(impl, name="builtin") + self.plugin_manager.register(impl, name="builtin") except Exception as exc: self._plugin_status["builtin"] = PluginStatus(is_success=False, detail=str(exc)) else: @@ -89,22 +89,14 @@ def load_hooks(self) -> None: for plugin_name, plugin in pending_plugins: try: - self.register_plugin(plugin, name=plugin_name) + if callable(plugin): # Support entry points that are classes + plugin = plugin(self) + self.plugin_manager.register(plugin, name=plugin_name) except Exception as exc: - logger.warning(f"Failed to register plugin '{plugin_name}': {exc}") - - def register_plugin(self, plugin: Any, name: str | None = None) -> str | None: - """Register a plugin instance or framework-aware factory and return its registered name.""" - try: - if callable(plugin): # Support entry points that are classes - plugin = plugin(self) - name = self._plugin_manager.register(plugin, name=name) - except Exception as exc: - self._plugin_status[name or plugin.__class__.__name__] = PluginStatus(is_success=False, detail=str(exc)) - raise - else: - self._plugin_status[name or plugin.__class__.__name__] = PluginStatus(is_success=True) - return name + logger.warning(f"Failed to initialize plugin '{plugin_name}': {exc}") + self._plugin_status[plugin_name] = PluginStatus(is_success=False, detail=str(exc)) + else: + self._plugin_status[plugin_name] = PluginStatus(is_success=True) def create_cli_app(self) -> typer.Typer: """Create CLI app by collecting commands from hooks. Can be used for custom CLI entry point.""" diff --git a/src/bub/skills.py b/src/bub/skills.py index 2bd53377..a286769c 100644 --- a/src/bub/skills.py +++ b/src/bub/skills.py @@ -6,7 +6,7 @@ import string import sys import warnings -from collections.abc import Collection +from collections.abc import Collection, Iterable from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -47,11 +47,16 @@ def body(self) -> str: }) -def discover_skills(workspace_path: Path) -> list[SkillMetadata]: +def discover_skills(workspace_path: Path, *, skill_dirs: Collection[Path] | None = None) -> list[SkillMetadata]: """Discover skills from project, global, and builtin roots with override precedence.""" skills_by_name: dict[str, SkillMetadata] = {} - for root, source in _iter_skill_roots(workspace_path): + skill_roots_iter: Iterable[tuple[Path, str]] + if skill_dirs is not None: + skill_roots_iter = ((root, "custom") for root in skill_dirs) + else: + skill_roots_iter = iter_skill_roots(workspace_path) + for root, source in skill_roots_iter: if not root.is_dir(): continue for skill_dir in sorted(root.iterdir()): @@ -168,7 +173,7 @@ def _builtin_skills_root() -> list[Path]: return [Path(p) for p in importlib.import_module("skills").__path__] -def _iter_skill_roots(workspace_path: Path) -> list[tuple[Path, str]]: +def iter_skill_roots(workspace_path: Path) -> list[tuple[Path, str]]: roots: list[tuple[Path, str]] = [] for source in SKILL_SOURCES: if source == "project": diff --git a/tests/test_builtin_agent.py b/tests/test_builtin_agent.py index f28a88cc..bc4c69fb 100644 --- a/tests/test_builtin_agent.py +++ b/tests/test_builtin_agent.py @@ -51,6 +51,9 @@ async def build_prompt(message: dict[str, Any], session_id: str, state: dict[str agent.settings = AgentSettings.model_construct(model="test:model", api_key="k", api_base="b", client_args={}) agent.framework = framework + agent.tools = REGISTRY.copy() + agent.tape_store = None + agent.skill_dirs = None agent.model_runner = _FakeModelRunner(agent.settings) return agent diff --git a/tests/test_builtin_cli.py b/tests/test_builtin_cli.py index f517046a..6e7d09d3 100644 --- a/tests/test_builtin_cli.py +++ b/tests/test_builtin_cli.py @@ -65,7 +65,7 @@ def onboard_config(self, current_config): "telegram": {"token": cli.typer.prompt("Telegram token", hide_input=True)}, } - framework.register_plugin(OnboardPlugin(), name="onboard-plugin") + framework.plugin_manager.register(OnboardPlugin(), name="onboard-plugin") app = framework.create_cli_app() answers = iter(["openai:gpt-5", "123:abc"]) @@ -438,7 +438,7 @@ def render_outbound(self, message, session_id, state, model_output): async def dispatch_outbound(self, message) -> bool: return True - framework.register_plugin(RunPlugin(), name="run-plugin") + framework.plugin_manager.register(RunPlugin(), name="run-plugin") app = framework.create_cli_app() result = CliRunner().invoke( diff --git a/tests/test_builtin_hook_impl.py b/tests/test_builtin_hook_impl.py index 12c639ad..8c1731d3 100644 --- a/tests/test_builtin_hook_impl.py +++ b/tests/test_builtin_hook_impl.py @@ -13,6 +13,7 @@ from bub.store import AsyncTapeStoreAdapter, FileTapeStore, InMemoryTapeStore from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState from bub.tape import Tape, TapeContext +from bub.tools import REGISTRY class RecordingLifespan: @@ -38,6 +39,7 @@ def _fake_tape(home: Path) -> Tape: class FakeAgent: def __init__(self, home: Path, *, tape: Tape | None = None) -> None: self.settings = SimpleNamespace(home=home) + self.tools = REGISTRY.copy() # A real in-memory async tape so load_state's recovery path runs against # the same store the tests write `model_switch` events to. self.tape = tape if tape is not None else _fake_tape(home) @@ -183,7 +185,7 @@ async def test_recover_session_model_returns_latest_recorded(tmp_path: Path) -> await session.append_event("model_switch", {"model": "openai:gpt-4o"}) await session.append_event("model_switch", {"model": "anthropic:claude-3"}) - assert await impl._recover_session_model("resolved-session") == "anthropic:claude-3" + assert await impl._recover_session_model("resolved-session", agent=agent) == "anthropic:claude-3" @pytest.mark.asyncio diff --git a/tests/test_framework.py b/tests/test_framework.py index a7f8fb2c..4833ce54 100644 --- a/tests/test_framework.py +++ b/tests/test_framework.py @@ -18,7 +18,7 @@ from bub.channels.message import ChannelMessage from bub.channels.telegram import TelegramSettings from bub.configure import ensure_config -from bub.framework import BubFramework, PluginStatus +from bub.framework import BubFramework from bub.hooks import hookimpl from bub.model_selection import ModelChoice, ModelOptions from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState @@ -54,7 +54,7 @@ def workspace_command(ctx: typer.Context) -> None: current = ctx.ensure_object(BubFramework) typer.echo(str(current.workspace)) - framework.register_plugin(CliPlugin(), name="cli-plugin") + framework.plugin_manager.register(CliPlugin(), name="cli-plugin") app = framework.create_cli_app() result = CliRunner().invoke(app, ["--workspace", str(tmp_path), "workspace"]) @@ -80,8 +80,8 @@ class HighPriorityPlugin: def provide_channels(self, message_handler): return [make_named_channel("shared", "high"), make_named_channel("high-only", "high")] - framework.register_plugin(LowPriorityPlugin(), name="low") - framework.register_plugin(HighPriorityPlugin(), name="high") + framework.plugin_manager.register(LowPriorityPlugin(), name="low") + framework.plugin_manager.register(HighPriorityPlugin(), name="high") channels = framework.get_channels(message_handler) @@ -109,9 +109,9 @@ class EmptyPlugin: def system_prompt(self, prompt: str, state: dict[str, str]) -> str | None: return None - framework.register_plugin(LowPriorityPlugin(), name="low") - framework.register_plugin(HighPriorityPlugin(), name="high") - framework.register_plugin(EmptyPlugin(), name="empty") + framework.plugin_manager.register(LowPriorityPlugin(), name="low") + framework.plugin_manager.register(HighPriorityPlugin(), name="high") + framework.plugin_manager.register(EmptyPlugin(), name="empty") prompt = framework.get_system_prompt(prompt="hello", state={}) @@ -134,10 +134,10 @@ def __init__(self, sidecar: Sidecar) -> None: def provide_tape_sidecar(self) -> Sidecar: return self.sidecar - framework.register_plugin(SidecarPlugin(Sidecar("shared", "low")), name="low-shared") - framework.register_plugin(SidecarPlugin(Sidecar("low-only", "low")), name="low-only") - framework.register_plugin(SidecarPlugin(Sidecar("shared", "high")), name="high-shared") - framework.register_plugin(SidecarPlugin(Sidecar("high-only", "high")), name="high-only") + framework.plugin_manager.register(SidecarPlugin(Sidecar("shared", "low")), name="low-shared") + framework.plugin_manager.register(SidecarPlugin(Sidecar("low-only", "low")), name="low-only") + framework.plugin_manager.register(SidecarPlugin(Sidecar("shared", "high")), name="high-shared") + framework.plugin_manager.register(SidecarPlugin(Sidecar("high-only", "high")), name="high-only") sidecars = {sidecar.name: sidecar for sidecar in framework.get_tape_sidecars()} @@ -166,8 +166,8 @@ async def continue_prompt(self, prompt: str, tape: Any, state: StreamState) -> s assert state.usage == {"total_tokens": 42} return "async prompt" - framework.register_plugin(SyncPlugin(), name="sync") - framework.register_plugin(AsyncPlugin(), name="async") + framework.plugin_manager.register(SyncPlugin(), name="sync") + framework.plugin_manager.register(AsyncPlugin(), name="async") prompt = await framework.continue_prompt(prompt="current prompt", tape=tape, state=state) @@ -195,7 +195,7 @@ def provide_tape_store(self): finally: tape_store.exit_count += 1 - framework.register_plugin(TapePlugin(), name="tape") + framework.plugin_manager.register(TapePlugin(), name="tape") async with framework.running(): assert framework.get_tape_store() is tape_store @@ -271,48 +271,6 @@ def register_cli_commands(self, app: typer.Typer) -> None: assert framework._plugin_status["config-plugin"].is_success is True -def test_register_plugin_initializes_callable_with_framework() -> None: - framework = BubFramework() - - class FrameworkAwarePlugin: - def __init__(self, received_framework: BubFramework) -> None: - self.framework = received_framework - - registered_name = framework.register_plugin(FrameworkAwarePlugin, name="framework-aware") - - assert registered_name == "framework-aware" - plugin = framework._plugin_manager.get_plugin("framework-aware") - assert isinstance(plugin, FrameworkAwarePlugin) - assert plugin.framework is framework - assert framework._plugin_status["framework-aware"] == PluginStatus(is_success=True) - - -def test_register_plugin_records_initialization_failure() -> None: - framework = BubFramework() - - class BrokenPlugin: - def __init__(self, _framework: BubFramework) -> None: - raise RuntimeError("initialization failed") - - with pytest.raises(RuntimeError, match="initialization failed"): - framework.register_plugin(BrokenPlugin, name="broken") - - assert framework._plugin_manager.get_plugin("broken") is None - assert framework._plugin_status["broken"] == PluginStatus( - is_success=False, - detail="initialization failed", - ) - - -def test_load_builtin_hooks_can_be_called_directly() -> None: - framework = BubFramework() - - framework.load_builtin_hooks() - - assert framework._plugin_manager.get_plugin("builtin") is not None - assert framework._plugin_status["builtin"] == PluginStatus(is_success=True) - - def test_collect_onboard_config_passes_accumulated_updates_to_later_hooks(write_config) -> None: with patch.dict(os.environ, {}, clear=True): framework = BubFramework(config_file=write_config("model: openai:gpt-5")) @@ -330,8 +288,8 @@ def onboard_config(self, current_config): observed_configs.append(("second", configure.merge({}, current_config))) return {"second": {"enabled": True}} - framework.register_plugin(FirstPlugin(), name="first") - framework.register_plugin(SecondPlugin(), name="second") + framework.plugin_manager.register(FirstPlugin(), name="first") + framework.plugin_manager.register(SecondPlugin(), name="second") result = framework.collect_onboard_config() @@ -377,7 +335,7 @@ def render_outbound(self, message, session_id, state, model_output): async def dispatch_outbound(self, message) -> bool: return True - framework.register_plugin(NonStreamingPlugin(), name="non-streaming") + framework.plugin_manager.register(NonStreamingPlugin(), name="non-streaming") result = await framework.process_inbound( ChannelMessage(session_id="s", channel="cli", chat_id="room", content="hi") @@ -399,7 +357,7 @@ def admit_message(self, session_id, message, turn): assert turn.pending_count == 1 return AdmitDecision("follow_up", reason="busy") - framework.register_plugin(AdmissionPlugin(), name="admission") + framework.plugin_manager.register(AdmissionPlugin(), name="admission") decision = await framework.admit_message( session_id="session", message={"content": "hello"}, @@ -438,8 +396,8 @@ def provide_model_options(self, session_id, workspace): current_model="high", ) - framework.register_plugin(LowPriorityPlugin(), name="low") - framework.register_plugin(HighPriorityPlugin(), name="high") + framework.plugin_manager.register(LowPriorityPlugin(), name="low") + framework.plugin_manager.register(HighPriorityPlugin(), name="high") options = await framework.get_model_options(session_id="session", workspace=tmp_path) @@ -508,7 +466,7 @@ async def dispatch_output(self, message) -> bool: async def quit(self, session_id: str) -> None: return None - framework.register_plugin(StreamingPlugin(), name="streaming") + framework.plugin_manager.register(StreamingPlugin(), name="streaming") framework.bind_channel_router(RecordingRouter()) result = await framework.process_inbound( diff --git a/tests/test_sdk_isolation.py b/tests/test_sdk_isolation.py new file mode 100644 index 00000000..4f1567dd --- /dev/null +++ b/tests/test_sdk_isolation.py @@ -0,0 +1,103 @@ +from pathlib import Path +from unittest.mock import Mock + +import pytest + +from bub.builtin import Agent +from bub.builtin.tools import resolve_tool_names, run_subagent +from bub.framework import BubFramework +from bub.store import InMemoryTapeStore +from bub.streaming import AsyncStreamEvents, StreamEvent +from bub.tape import Tape +from bub.tools import Tool, ToolContext + + +def _reply() -> AsyncStreamEvents: + async def events(): + yield StreamEvent("text", {"delta": "done"}) + yield StreamEvent("final", {"text": "done"}) + + return AsyncStreamEvents(events()) + + +@pytest.fixture +def framework(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> BubFramework: + monkeypatch.setenv("BUB_HOME", str(tmp_path)) + framework = BubFramework(config_file=tmp_path / "config.yml") + framework.workspace = tmp_path + framework.load_builtin_hooks() + return framework + + +@pytest.mark.asyncio +@pytest.mark.parametrize("has_saved_state", [False, True]) +@pytest.mark.parametrize("override", [False, True]) +async def test_sdk_recovers_only_its_store_and_honors_explicit_overrides( + framework: BubFramework, has_saved_state: bool, override: bool +) -> None: + builtin = framework.plugin_manager.get_plugin("builtin") + builtin_tape = builtin._get_agent().tape.session_tape("shared", framework.workspace) + await builtin_tape.append_event("model_switch", {"model": "test:other"}) + await builtin_tape.append_event("reasoning_effort_switch", {"reasoning_effort": "low"}) + + agent = Agent(framework, tools=[], tape_store=InMemoryTapeStore(), skill_dirs=[]) + tape = agent.tape.session_tape("shared", framework.workspace) + if has_saved_state: + await tape.append_event("model_switch", {"model": "test:saved"}) + await tape.append_event("reasoning_effort_switch", {"reasoning_effort": "high"}) + + runner = Mock(side_effect=lambda **kwargs: _reply()) + agent.model_runner.run = runner + stream = await agent.run_stream( + session_id="shared", + prompt="hello", + model="test:explicit" if override else None, + reasoning_effort="medium" if override else None, + ) + assert [event.kind async for event in stream] == ["text", "final"] + call = runner.call_args.kwargs + expected_model = "test:saved" if has_saved_state else agent.settings.model + assert call["model"] == ("test:explicit" if override else expected_model) + state = call["tape"].context.state + assert state.get("reasoning_effort") == ("medium" if override else "high" if has_saved_state else None) + assert state["_runtime_agent"] is agent + + +def test_instance_tool_names_resolve_aliases_and_exclusions_from_one_index() -> None: + names = ["sdk.lookup", "sdk.other"] + assert resolve_tool_names([" SDK_LOOKUP "], all_names=iter(names)) == {"sdk.lookup"} + assert resolve_tool_names(exclude=["SDK_OTHER"], all_names=iter(names)) == {"sdk.lookup"} + assert resolve_tool_names(["sdk_lookup"], exclude=["sdk.lookup"], all_names=names) == set() + assert resolve_tool_names(all_names=[]) == set() + with pytest.raises(ValueError, match="bash"): + resolve_tool_names(["bash"], all_names=names) + with pytest.raises(ValueError, match="bash"): + resolve_tool_names(exclude=["bash"], all_names=names) + + +@pytest.mark.asyncio +async def test_agent_allowlist_accepts_unregistered_instance_tool(framework: BubFramework) -> None: + tool = Tool.from_callable(lambda: "found", name="sdk.lookup") + agent = Agent(framework, tools=[tool], tape_store=InMemoryTapeStore(), skill_dirs=[]) + runner = Mock(side_effect=lambda **kwargs: _reply()) + agent.model_runner.run = runner + stream = await agent.run_stream(session_id="sdk", prompt="lookup", allowed_tools=[" SDK_LOOKUP "]) + assert [event.kind async for event in stream] == ["text", "final"] + assert [tool.name for tool in runner.call_args.kwargs["tools"]] == ["sdk_lookup"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("allowed_tools", [None, ["SDK_LOOKUP"]]) +async def test_subagent_uses_parent_instance_tools(framework: BubFramework, allowed_tools: list[str] | None) -> None: + tool = Tool.from_callable(lambda: "found", name="sdk.lookup") + agent = Agent(framework, tools=[tool, run_subagent], tape_store=InMemoryTapeStore(), skill_dirs=[]) + runner = Mock(side_effect=lambda **kwargs: _reply()) + agent.model_runner.run = runner + tape: Tape = agent.tape.session_tape("parent", framework.workspace) + context = ToolContext( + tape=tape, + state={"_runtime_agent": agent, "session_id": "parent", "_runtime_workspace": str(framework.workspace)}, + ) + result = await run_subagent.run(prompt="lookup", allowed_tools=allowed_tools, context=context) + assert result == "done" + assert [tool.name for tool in runner.call_args.kwargs["tools"]] == ["sdk_lookup"] diff --git a/tests/test_skills.py b/tests/test_skills.py index 617445f3..5961518a 100644 --- a/tests/test_skills.py +++ b/tests/test_skills.py @@ -126,7 +126,7 @@ def test_discover_skills_prefers_project_over_global_and_builtin(tmp_path: Path, _write_skill(global_root, "global-only", description="global only") monkeypatch.setattr( - "bub.skills._iter_skill_roots", + "bub.skills.iter_skill_roots", lambda _workspace: [ (project_root, "project"), (global_root, "global"), diff --git a/tests/test_subagent_tool.py b/tests/test_subagent_tool.py index c8d907c9..c5bb64d7 100644 --- a/tests/test_subagent_tool.py +++ b/tests/test_subagent_tool.py @@ -20,6 +20,7 @@ def __init__(self, state: dict[str, Any]) -> None: class FakeAgent: def __init__(self) -> None: + self.tools = REGISTRY.copy() self.run_stream = AsyncMock(side_effect=self._run_stream) async def _run_stream(self, **kwargs: Any) -> AsyncStreamEvents: From 137b6489b16dabd77fbb747e3c85797632884fa7 Mon Sep 17 00:00:00 2001 From: Frost Ming Date: Tue, 15 Sep 2026 19:22:59 +0800 Subject: [PATCH 2/3] feat: enhance documentation for Python SDK with installation and usage examples Signed-off-by: Frost Ming --- src/bub/builtin/agent.py | 45 ++++ src/bub/framework.py | 58 +++- website/src/content/docs/docs/build/sdk.md | 246 +++++++++++++++++ .../src/content/docs/zh-cn/docs/build/sdk.md | 248 ++++++++++++++++++ 4 files changed, 596 insertions(+), 1 deletion(-) create mode 100644 website/src/content/docs/docs/build/sdk.md create mode 100644 website/src/content/docs/zh-cn/docs/build/sdk.md diff --git a/src/bub/builtin/agent.py b/src/bub/builtin/agent.py index c2640bd3..a413db2a 100644 --- a/src/bub/builtin/agent.py +++ b/src/bub/builtin/agent.py @@ -47,6 +47,21 @@ def __init__( tape_store: TapeStore | AsyncTapeStore | None = None, skill_dirs: Collection[Path] | None = None, ) -> None: + """Create a builtin agent with instance-specific tools, skills, and storage. + + Args: + framework: Configured hook runtime supplying prompts, tape context, + interception hooks, and optional shared resources. + tools: Tools available to this instance. None snapshots the global + registry; an empty collection disables tools. + tape_store: Explicit store, preferred over the framework's active + store. Without either, the agent uses an in-memory store. + skill_dirs: Skill roots in precedence order. None uses project, user, + and builtin discovery; an empty collection disables discovery. + + Settings come from Bub's process-wide configuration. The caller owns the + lifecycle of an explicitly supplied store. + """ self.settings = load_settings() self.framework = framework self.tools = {tool.name: tool for tool in tools} if tools is not None else REGISTRY.copy() @@ -56,6 +71,12 @@ def __init__( @cached_property def tape(self) -> Tape: + """Return the lazily constructed, cached tape factory for this agent. + + Select the explicit store, active framework store, or an in-memory fallback, + in that order. Adapt synchronous stores and use hook-provided context and + sidecars. Archive files use ``bub.home / 'tapes'`` independently of the store. + """ import bub tape_store: TapeStore | AsyncTapeStore | None @@ -106,6 +127,30 @@ async def run_stream( allowed_tools: Collection[str] | None = None, reasoning_effort: str | None = None, ) -> AsyncStreamEvents: + """Prepare a turn and return its stream; await this method before iterating. + + Args: + session_id: Session identity within the workspace. A ``temp/`` prefix + prevents the turn's fork from merging back into its parent tape. + prompt: Text or multimodal content parts. Text beginning with a comma + after stripping whitespace invokes a builtin command. + state: Mutable turn state. None loads state through framework hooks + using this agent's store; supplied state skips that loading. + The current agent is always bound into the state. + model: Per-turn override, ahead of the state and configured model. + allowed_skills: Case-insensitive skill names available to this turn; + None leaves discovery unrestricted. + allowed_tools: Instance tool names or model aliases for the agent loop; + None allows all instance tools and an empty collection allows none. + Command execution uses the instance's tools directly. + reasoning_effort: Per-turn override of the value in state. + + Consume the stream to completion to finish execution and tape merging. + A ``final`` event ends a model step, not necessarily the whole turn. + The returned object exposes ``error`` and ``usage``; execution can also + raise exceptions. This method does not render or dispatch outbound messages, + call save-state hooks, or serialize concurrent turns in the same session. + """ if not prompt: return self._events_from_iterable([ StreamEvent("text", {"delta": "error: empty prompt"}), diff --git a/src/bub/framework.py b/src/bub/framework.py index 7911c486..825c128d 100644 --- a/src/bub/framework.py +++ b/src/bub/framework.py @@ -47,6 +47,11 @@ class BubFramework: """Minimal framework core. Everything grows from hook skills.""" def __init__(self, config_file: Path = DEFAULT_CONFIG_FILE) -> None: + """Create a hook runtime and load the process-wide configuration file. + + The workspace initially points to the current directory. Register plugins + or load builtin hooks before executing turns; construction does not load them. + """ self.workspace = Path.cwd().resolve() self.config_file = config_file.resolve() self.plugin_manager = pluggy.PluginManager(BUB_HOOK_NAMESPACE) @@ -73,6 +78,11 @@ def load_builtin_hooks(self) -> None: self._plugin_status["builtin"] = PluginStatus(is_success=True) def load_hooks(self) -> None: + """Load builtin hooks, then plugins from the ``bub`` entry-point group. + + Callable entry points receive this framework. Failed plugins are recorded + for diagnostics without preventing the remaining plugins from loading. + """ import importlib.metadata pending_plugins: list[tuple[str, Any]] = [] @@ -133,6 +143,12 @@ async def continue_prompt(self, prompt: str | list[dict], tape: Tape, state: Str raise TypeError("hook.continue_prompt must return str") async def build_state(self, message: Envelope, session_id: str) -> TurnState: + """Merge runtime defaults and load-state hooks into a fresh turn state. + + Higher-priority hooks override lower-priority values. SDK callers can + supply their Agent in the message's ``_runtime_agent`` field so builtin + session recovery reads that agent's store. + """ state = {"_runtime_workspace": str(self.workspace), "_runtime_steering_inbox": self.get_steering_inbox()} for hook_state in reversed( await self._hook_runtime.call_many("load_state", message=message, session_id=session_id) @@ -142,7 +158,12 @@ async def build_state(self, message: Envelope, session_id: str) -> TurnState: return state async def process_inbound(self, inbound: Envelope, stream_output: bool = False) -> TurnResult: - """Run one inbound message through hooks and return turn result.""" + """Resolve, execute, save, render, and dispatch one complete message turn. + + With ``stream_output=True``, consume model events through the bound channel + router. This method still returns a completed TurnResult, not an iterator. + Use inside ``running()`` when hooks provide stores or other resources. + """ try: session_id = await self.resolve_session(inbound) @@ -230,18 +251,25 @@ def hook_report(self) -> dict[str, list[str]]: return self._hook_runtime.hook_report() def bind_channel_router(self, router: ChannelRouter | None) -> None: + """Attach the outbound/stream router, or detach it with ``None``.""" self._channel_router = router async def dispatch_via_channel_router(self, message: Envelope) -> bool: + """Dispatch through the bound router; return False when no router is bound.""" if self._channel_router is None: return False return await self._channel_router.dispatch_output(message) async def quit_via_channel_router(self, session_id: str) -> None: + """Ask the bound router to quit a session; do nothing without a router.""" if self._channel_router is not None: await self._channel_router.quit(session_id) async def admit_message(self, *, session_id: str, message: Envelope, turn: TurnSnapshot) -> AdmitDecision | None: + """Ask admission hooks how to handle a message arriving during a turn. + + Return None when no hook decides; reject unsupported return types. + """ decision = await self._hook_runtime.call_first( "admit_message", session_id=session_id, @@ -292,6 +320,11 @@ async def steer_message( state: TurnState, reason: str | None = None, ) -> bool: + """Enqueue a message for an active turn, returning False without an inbox. + + Set the state's session id if absent and attach the optional reason to + message context when the envelope supports attribute assignment. + """ inbox = self.get_steering_inbox() if inbox is None: return False @@ -344,6 +377,7 @@ async def _collect_outbounds( return [fallback] def get_channels(self, message_handler: MessageHandler) -> dict[str, Channel]: + """Collect channels by name, preferring higher-priority providers on duplicates.""" channels: dict[str, Channel] = {} for result in self._hook_runtime.call_many_sync("provide_channels", message_handler=message_handler): for channel in result: @@ -353,6 +387,13 @@ def get_channels(self, message_handler: MessageHandler) -> dict[str, Channel]: @contextlib.asynccontextmanager async def running(self) -> AsyncGenerator[contextlib.AsyncExitStack, None]: + """Acquire hook-provided stores and steering resources for an application lifespan. + + Yield an AsyncExitStack for additional application resources. Exit closes + acquired context managers and clears the framework's resource references. + Enter before an Agent first accesses its cached tape; avoid overlapping + lifespans on the same framework instance. + """ async with contextlib.AsyncExitStack() as stack: tape_store = self._hook_runtime.call_first_sync("provide_tape_store") # Allow plugins to return either TapeStore/AsyncTapeStore instances or context managers for them @@ -368,21 +409,30 @@ async def running(self) -> AsyncGenerator[contextlib.AsyncExitStack, None]: self._steering_inbox = None def get_tape_store(self) -> TapeStore | AsyncTapeStore | None: + """Return the store acquired by ``running()``, or None when unavailable.""" return self._tape_store def get_tape_sidecars(self) -> tuple[TapeSidecar, ...]: + """Collect tape sidecars, keeping the highest-priority provider for each name.""" sidecars: dict[str, TapeSidecar] = {} for sidecar in self._hook_runtime.call_many_sync("provide_tape_sidecar"): sidecars.setdefault(sidecar.name, sidecar) return tuple(sidecars.values()) def get_steering_inbox(self) -> SteeringInbox | None: + """Return the inbox acquired by ``running()``, or None when unavailable.""" return self._steering_inbox def get_agent_hooks(self) -> AgentHooks: + """Return the model and tool interception adapter for this framework's hooks.""" return self._agent_hooks def get_system_prompt(self, prompt: str | list[dict], state: dict[str, Any]) -> str: + """Join nonempty system-prompt hook results from low to high priority. + + Hooks contribute additional blocks; a higher-priority hook does not replace + a lower-priority prompt. Blocks are separated by blank lines. + """ return "\n\n".join( result for result in reversed(self._hook_runtime.call_many_sync("system_prompt", prompt=prompt, state=state)) @@ -390,12 +440,18 @@ def get_system_prompt(self, prompt: str | list[dict], state: dict[str, Any]) -> ) def build_tape_context(self) -> TapeContext: + """Get the highest-priority tape context, raising TypeError if none is valid.""" context = self._hook_runtime.call_first_sync("build_tape_context") if isinstance(context, TapeContext): return context raise TypeError("hook.build_tape_context must return TapeContext") def collect_onboard_config(self) -> dict[str, Any]: + """Merge onboarding hook contributions and validate the resulting configuration. + + Each hook receives the accumulated config; higher-priority hooks run last. + This method collects settings but does not write the configuration file. + """ current_config: dict[str, Any] = {} for impl in reversed(list(self._hook_runtime._iter_hookimpls("onboard_config"))): diff --git a/website/src/content/docs/docs/build/sdk.md b/website/src/content/docs/docs/build/sdk.md new file mode 100644 index 00000000..83f131cc --- /dev/null +++ b/website/src/content/docs/docs/build/sdk.md @@ -0,0 +1,246 @@ +--- +title: Python SDK +description: Embed Bub in a Python application with custom tools, prompts, skills, and session storage. +sidebar: + order: 0 +--- + +Use `bub.builtin.Agent` to run the builtin agent loop inside your application. +`BubFramework` supplies configuration and hooks; `Agent` owns execution with your tools and store. + +## Install and configure + +Install Bub in your Python 3.12+ project: + +```bash +uv add bub +``` + +Configure a model and its credentials through `BUB_MODEL`, `BUB_API_KEY`, and optionally `BUB_API_BASE`, +or through a configuration file. See [Configuration](/docs/reference/settings/). +The API below describes this source checkout; use a release containing these interfaces or install the checkout. + +## Create an agent + +Save the following as `sdk_example.py`: + +```python +import asyncio +from pathlib import Path + +from bub import BubFramework, hookimpl +from bub.builtin import Agent +from bub.builtin.hook_impl import BuiltinImpl +from bub.builtin.tools import skill_describe +from bub.store import FileTapeStore +from bub.tools import Tool + + +class ApplicationHooks(BuiltinImpl): + def __init__(self, framework: BubFramework, prompts: list[str]) -> None: + super().__init__(framework) + self.prompts = tuple(prompts) + + @hookimpl + def system_prompt(self, prompt, state) -> str: + return "\n\n".join(self.prompts) + + +async def lookup_order(order_id: str) -> dict[str, str]: + """Look up the status of an order.""" + # Replace this demonstration response with your application's order service. + return {"order_id": order_id, "status": "shipped"} + + +def create_agent() -> tuple[BubFramework, Agent]: + root = Path(__file__).resolve().parent + framework = BubFramework(config_file=root / "config.yml") + framework.workspace = root + framework.plugin_manager.register( + ApplicationHooks(framework, [ + "You are an order assistant. Reply directly to the user.", + "Use lookup_order to check order status before answering.", + ]), + name="application", + ) + agent = Agent( + framework, + tools=[Tool.from_callable(lookup_order), skill_describe], + skill_dirs=[root / "skills"], + tape_store=FileTapeStore(root / "sessions"), + ) + return framework, agent + + +async def main() -> None: + framework, agent = create_agent() + async with framework.running(): + stream = await agent.run_stream( + session_id="customer-42", + prompt="Where is order A123?", + ) + 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()) +``` + +Run it with `uv run python sdk_example.py` after configuring your model. +An absent `config.yml` is allowed; environment settings still apply. + +`ApplicationHooks` replaces the builtin system-prompt method while keeping the other builtin hooks. +Register this instance once: do not also call `load_builtin_hooks()` or `load_hooks()` in this example. +System-prompt hooks are additive, so registering an extra prompt hook alongside the standard builtin implementation +would retain its channel instructions and workspace `AGENTS.md` content. + +For standard Bub defaults instead, create a framework and call `framework.load_builtin_hooks()`. +Use `load_hooks()` when you also want installed plugins from the `bub` entry-point group. + +## Tools, skills, and sessions + +| Parameter | Behavior | +| --- | --- | +| `tools=[...]` | Accepts `Tool` objects. `Tool.from_callable()` derives a schema from annotations and a description from the docstring. | +| `tools=None` | Copies the global tool registry at construction time. `tools=[]` disables tools. | +| `skill_dirs=[Path(...)]` | Searches only these roots, in order; the first skill of a given name wins. | +| `skill_dirs=None` | Searches project, user, and builtin roots. `skill_dirs=[]` disables discovery. | +| `tape_store=...` | Uses the supplied `TapeStore` or `AsyncTapeStore`. `FileTapeStore(path)` persists session tapes in that directory. | +| `tape_store=None` | Uses the framework's active store, or an instance-local memory store if no store is active. | + +Unlike `@tool`, `Tool.from_callable()` does not register the tool globally. +Include `skill_describe` in the tool set when the model needs to load skill bodies on demand. +Skills supply instructions; they do not automatically grant shell or filesystem tools. + +Each skill lives in its own directory: + +```text +skills/ +└── order-policy/ + └── SKILL.md +``` + +```markdown +--- +name: order-policy +description: Rules for answering order-status questions. +--- +Check the order service before giving a delivery status. +``` + +Reuse a `session_id` with the same workspace and store to continue a conversation. +Choose a new ID for a new conversation. IDs starting with `temp/` run on a fork that is not merged back. +Archive and builtin sidecar paths still follow Bub configuration, independently of `FileTapeStore`'s directory. + +Per-turn options narrow the instance's capabilities and override persisted model settings: + +```python +stream = await agent.run_stream( + session_id="customer-42", + prompt="Use $order-policy to check order A123.", + allowed_tools=["lookup_order", "skill"], + allowed_skills=["order-policy"], + model="provider:model-id", # Replace with your configured provider/model. + reasoning_effort="high", # Use a value supported by that model. +) +``` + +Always consume the returned stream. `allowed_tools` accepts runtime names and model aliases +(for example, `fs.read` and `fs_read`). Explicit `model` and `reasoning_effort` take precedence over saved settings. +Normally omit `state` to load session state automatically; passing a state dictionary skips that loading and mutates it. +Comma-prefixed text is still treated as a command. Command execution uses the instance tool set directly, +so loop-level `allowed_tools` filtering is not a command permission boundary. + +## Streams and lifecycle + +`run_stream()` is awaited first, then its result is iterated. Events include `text`, `reasoning`, `tool_call`, +`tool_result`, `usage`, `error`, and `final`. A `final` event finishes a model step; exhaust the iterator to finish +the whole turn. Check error events and handle exceptions from iteration. The stream also exposes `error` and `usage`. + +For early exit from an iteration that has started, close its iterator explicitly: + +```python +from contextlib import aclosing + +stream = await agent.run_stream(session_id="customer-42", prompt="Check order A123") +async with aclosing(stream.__aiter__()) as events: + async for event in events: + print(event.kind, event.data) +``` + +Keep `framework.running()` open until all turns finish. Accessing the agent's cached `tape` before entering that +lifespan can bind it to an in-memory fallback. An explicitly injected store's lifecycle belongs to your application. +`Agent` itself is not an async context manager and has no `run()` convenience method. + +## Embed in FastAPI + +Install the server dependencies with `uv add fastapi uvicorn`, then save this as `app.py` next to `sdk_example.py`: + +```python +import asyncio +from contextlib import asynccontextmanager + +from fastapi import FastAPI, HTTPException, Request +from pydantic import BaseModel + +from sdk_example import create_agent + + +@asynccontextmanager +async def lifespan(app: FastAPI): + framework, agent = create_agent() + async with framework.running(): + app.state.agent = agent + app.state.turn_lock = asyncio.Lock() + yield + + +app = FastAPI(lifespan=lifespan) + + +class Task(BaseModel): + session_id: str + prompt: str + + +@app.post("/tasks") +async def run_task(task: Task, request: Request): + async with request.app.state.turn_lock: + stream = await request.app.state.agent.run_stream( + session_id=task.session_id, + prompt=task.prompt, + ) + parts: list[str] = [] + error: str | None = None + async for event in stream: + if event.kind == "text": + parts.append(str(event.data.get("delta", ""))) + elif event.kind == "error": + error = str(event.data.get("message", "Agent failed")) + if error is not None: + raise HTTPException(status_code=502, detail=error) + return {"session_id": task.session_id, "output": "".join(parts)} +``` + +Start with `uv run uvicorn app:app`. This example serializes all turns in one process. +For concurrent independent sessions, use application-owned per-session locking with a cleanup policy. +Multiple worker processes sharing a store need coordination across workers; a file write lock does not serialize a turn. + +## Agent versus the full message pipeline + +Direct `Agent.run_stream()` uses the builtin loop and its model/tool interception hooks. +It does not run `build_prompt`, `save_state`, outbound rendering, or channel dispatch, and does not select a plugin's +replacement `run_model` implementation. Your application consumes the result. + +Use `BubFramework.process_inbound()` for the complete message pipeline and model-runner plugins. +It returns a `TurnResult` with `model_output`, `state`, and `outbounds`; `stream_output=True` routes streaming output +through a bound channel router but still returns the completed result. +See [Hooks](/docs/build/hooks/) for those extension points. + +Configuration loading remains process-wide. Configure once at application startup; +separate tool/skill/store instances do not imply isolated configuration files or environment variables. diff --git a/website/src/content/docs/zh-cn/docs/build/sdk.md b/website/src/content/docs/zh-cn/docs/build/sdk.md new file mode 100644 index 00000000..381195b7 --- /dev/null +++ b/website/src/content/docs/zh-cn/docs/build/sdk.md @@ -0,0 +1,248 @@ +--- +title: Python SDK +description: 在 Python 应用中嵌入 Bub，自定义工具、提示词、skills 和会话存储。 +sidebar: + order: 0 +--- + +使用 `bub.builtin.Agent` 在应用内运行 builtin agent loop。 +`BubFramework` 提供配置和 hooks，`Agent` 使用指定的工具与存储执行任务。 + +## 安装与配置 + +在 Python 3.12+ 项目中安装： + +```bash +uv add bub +``` + +通过 `BUB_MODEL`、`BUB_API_KEY` 和可选的 `BUB_API_BASE`，或配置文件设置模型与凭据。 +详见[配置参考](/zh-cn/docs/reference/settings/)。本文描述当前源码接口； +请使用包含这些接口的版本，或直接安装当前源码。 + +## 创建 Agent + +将下面的示例保存为 `sdk_example.py`： + +```python +import asyncio +from pathlib import Path + +from bub import BubFramework, hookimpl +from bub.builtin import Agent +from bub.builtin.hook_impl import BuiltinImpl +from bub.builtin.tools import skill_describe +from bub.store import FileTapeStore +from bub.tools import Tool + + +class ApplicationHooks(BuiltinImpl): + def __init__(self, framework: BubFramework, prompts: list[str]) -> None: + super().__init__(framework) + self.prompts = tuple(prompts) + + @hookimpl + def system_prompt(self, prompt, state) -> str: + return "\n\n".join(self.prompts) + + +async def lookup_order(order_id: str) -> dict[str, str]: + """Look up the status of an order.""" + # Replace this demonstration response with your application's order service. + return {"order_id": order_id, "status": "shipped"} + + +def create_agent() -> tuple[BubFramework, Agent]: + root = Path(__file__).resolve().parent + framework = BubFramework(config_file=root / "config.yml") + framework.workspace = root + framework.plugin_manager.register( + ApplicationHooks(framework, [ + "You are an order assistant. Reply directly to the user.", + "Use lookup_order to check order status before answering.", + ]), + name="application", + ) + agent = Agent( + framework, + tools=[Tool.from_callable(lookup_order), skill_describe], + skill_dirs=[root / "skills"], + tape_store=FileTapeStore(root / "sessions"), + ) + return framework, agent + + +async def main() -> None: + framework, agent = create_agent() + async with framework.running(): + stream = await agent.run_stream( + session_id="customer-42", + prompt="Where is order A123?", + ) + 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 sdk_example.py`。 +`config.yml` 不存在时也可以运行，环境变量配置仍然生效。 +示例中的订单查询返回演示数据，接入应用时请替换为实际服务。 + +`ApplicationHooks` 覆盖 builtin 的系统提示方法，并继承其他 builtin hooks。 +只注册这一个实例即可；这个示例中不要再调用 `load_builtin_hooks()` 或 `load_hooks()`。 +系统提示 hooks 会累加，单独新增一个 prompt hook 会保留默认 builtin 的 channel 指令和工作区 `AGENTS.md` 内容。 + +如需 Bub 标准默认行为，创建 framework 后调用 `framework.load_builtin_hooks()`。 +需要自动加载 `bub` entry-point group 下已安装的插件时，使用 `load_hooks()`。 + +## 工具、Skills 与会话 + +| 参数 | 行为 | +| --- | --- | +| `tools=[...]` | 接受 `Tool` 对象。`Tool.from_callable()` 从类型注解生成 schema，从 docstring 获取描述。 | +| `tools=None` | 构造时复制全局工具注册表。`tools=[]` 禁用工具。 | +| `skill_dirs=[Path(...)]` | 仅搜索这些目录，按顺序处理，同名 skill 使用第一个。 | +| `skill_dirs=None` | 搜索项目、用户和 builtin 目录。`skill_dirs=[]` 禁用发现。 | +| `tape_store=...` | 使用传入的 `TapeStore` 或 `AsyncTapeStore`。`FileTapeStore(path)` 将会话 tape 持久化到指定目录。 | +| `tape_store=None` | 使用 framework 当前的 store；没有活动 store 时使用实例独立的内存存储。 | + +与 `@tool` 不同，`Tool.from_callable()` 不会把工具注册到全局表。 +模型需要按需读取 skill 正文时，将 `skill_describe` 加入工具集合。 +Skills 提供指令，不会自动授予 shell 或文件系统工具。 + +每个 skill 单独放在一个目录中： + +```text +skills/ +└── order-policy/ + └── SKILL.md +``` + +```markdown +--- +name: order-policy +description: Rules for answering order-status questions. +--- +Check the order service before giving a delivery status. +``` + +在相同 workspace 和 store 中复用 `session_id` 可以继续对话，新对话使用新 ID。 +以 `temp/` 开头的会话运行在 fork 中，不会合并回父 tape。 +归档和 builtin sidecar 的路径仍由 Bub 配置决定，独立于 `FileTapeStore` 的目录。 + +每轮调用可以缩小实例能力范围，并覆盖已保存的模型设置： + +```python +stream = await agent.run_stream( + session_id="customer-42", + prompt="Use $order-policy to check order A123.", + allowed_tools=["lookup_order", "skill"], + allowed_skills=["order-policy"], + model="provider:model-id", # Replace with your configured provider/model. + reasoning_effort="high", # Use a value supported by that model. +) +``` + +必须消费返回的 stream。`allowed_tools` 支持工具原名和模型别名，例如 `fs.read` 与 `fs_read`。 +显式 `model` 和 `reasoning_effort` 优先于已保存设置；示例中的模型标识需要替换成实际模型， +reasoning effort 也需使用该模型支持的值。 +通常省略 `state`，由 framework 自动加载；传入字典会跳过加载，并在执行中修改该字典。 +逗号开头的文本仍被解释为命令。命令直接使用实例工具集合，agent loop 的 `allowed_tools` 过滤不适合作为命令权限边界。 + +## 流式事件与生命周期 + +先 `await run_stream()`，再迭代其返回值。事件包括 `text`、`reasoning`、`tool_call`、 +`tool_result`、`usage`、`error` 和 `final`。 +`final` 表示一次模型步骤结束，整个 turn 需要等到迭代完成。 +调用方应检查 error 事件并处理迭代抛出的异常。stream 对象还提供 `error` 和 `usage` 属性。 + +如果要提前退出已经开始的迭代，显式关闭迭代器： + +```python +from contextlib import aclosing + +stream = await agent.run_stream(session_id="customer-42", prompt="Check order A123") +async with aclosing(stream.__aiter__()) as events: + async for event in events: + print(event.kind, event.data) +``` + +所有 turn 完成之前，保持 `framework.running()` 开启。 +在进入生命周期前访问 Agent 缓存的 `tape`，可能使其绑定到内存 fallback。 +显式注入的 store 由应用管理生命周期。 +`Agent` 本身不是异步上下文管理器，当前也没有 `run()` 便捷方法。 + +## 接入 FastAPI + +运行 `uv add fastapi uvicorn` 安装服务依赖，然后在 `sdk_example.py` 旁保存 `app.py`： + +```python +import asyncio +from contextlib import asynccontextmanager + +from fastapi import FastAPI, HTTPException, Request +from pydantic import BaseModel + +from sdk_example import create_agent + + +@asynccontextmanager +async def lifespan(app: FastAPI): + framework, agent = create_agent() + async with framework.running(): + app.state.agent = agent + app.state.turn_lock = asyncio.Lock() + yield + + +app = FastAPI(lifespan=lifespan) + + +class Task(BaseModel): + session_id: str + prompt: str + + +@app.post("/tasks") +async def run_task(task: Task, request: Request): + async with request.app.state.turn_lock: + stream = await request.app.state.agent.run_stream( + session_id=task.session_id, + prompt=task.prompt, + ) + parts: list[str] = [] + error: str | None = None + async for event in stream: + if event.kind == "text": + parts.append(str(event.data.get("delta", ""))) + elif event.kind == "error": + error = str(event.data.get("message", "Agent failed")) + if error is not None: + raise HTTPException(status_code=502, detail=error) + return {"session_id": task.session_id, "output": "".join(parts)} +``` + +使用 `uv run uvicorn app:app` 启动。这个示例在单进程内串行执行所有 turn。 +需要不同会话并行时，可以由应用维护带清理机制的按会话锁。 +多个 worker 共享 store 时需要跨进程协调，文件写锁并不能保证整个 turn 串行执行。 + +## Agent 与完整消息管线 + +直接调用 `Agent.run_stream()` 会运行 builtin loop，以及模型和工具拦截 hooks。 +它不会调用 `build_prompt`、`save_state`、outbound 渲染或 channel 分发， +也不会选择插件替换的 `run_model` 实现。输出由应用消费。 + +需要完整消息管线或模型执行插件时，使用 `BubFramework.process_inbound()`。 +它返回包含 `model_output`、`state` 和 `outbounds` 的 `TurnResult`； +`stream_output=True` 会通过绑定的 channel router 路由流式输出，但最终仍返回完整结果。 +扩展点详见 [Hooks](/zh-cn/docs/build/hooks/)。 + +配置加载仍是进程级的，应在应用启动时统一配置。 +工具、skills 和 store 按实例隔离，不代表配置文件和环境变量也按实例隔离。 From d0b47248c6363a6371e8aefb5b20a54995783488 Mon Sep 17 00:00:00 2001 From: Frost Ming Date: Wed, 16 Sep 2026 07:41:41 +0800 Subject: [PATCH 3/3] feat: refactor ApplicationHooks to SystemPrompts for improved prompt management Signed-off-by: Frost Ming --- website/src/content/docs/docs/build/sdk.md | 27 +++++++++---------- .../src/content/docs/zh-cn/docs/build/sdk.md | 27 +++++++++---------- 2 files changed, 26 insertions(+), 28 deletions(-) diff --git a/website/src/content/docs/docs/build/sdk.md b/website/src/content/docs/docs/build/sdk.md index 83f131cc..abd92ca5 100644 --- a/website/src/content/docs/docs/build/sdk.md +++ b/website/src/content/docs/docs/build/sdk.md @@ -36,9 +36,8 @@ from bub.store import FileTapeStore from bub.tools import Tool -class ApplicationHooks(BuiltinImpl): - def __init__(self, framework: BubFramework, prompts: list[str]) -> None: - super().__init__(framework) +class SystemPrompts: + def __init__(self, prompts: list[str]) -> None: self.prompts = tuple(prompts) @hookimpl @@ -57,11 +56,11 @@ def create_agent() -> tuple[BubFramework, Agent]: framework = BubFramework(config_file=root / "config.yml") framework.workspace = root framework.plugin_manager.register( - ApplicationHooks(framework, [ + SystemPrompts([ "You are an order assistant. Reply directly to the user.", "Use lookup_order to check order status before answering.", ]), - name="application", + name="system_prompts", ) agent = Agent( framework, @@ -94,7 +93,7 @@ if __name__ == "__main__": Run it with `uv run python sdk_example.py` after configuring your model. An absent `config.yml` is allowed; environment settings still apply. -`ApplicationHooks` replaces the builtin system-prompt method while keeping the other builtin hooks. +`SystemPrompts` replaces the builtin system-prompt method while keeping the other builtin hooks. Register this instance once: do not also call `load_builtin_hooks()` or `load_hooks()` in this example. System-prompt hooks are additive, so registering an extra prompt hook alongside the standard builtin implementation would retain its channel instructions and workspace `AGENTS.md` content. @@ -104,14 +103,14 @@ Use `load_hooks()` when you also want installed plugins from the `bub` entry-poi ## Tools, skills, and sessions -| Parameter | Behavior | -| --- | --- | -| `tools=[...]` | Accepts `Tool` objects. `Tool.from_callable()` derives a schema from annotations and a description from the docstring. | -| `tools=None` | Copies the global tool registry at construction time. `tools=[]` disables tools. | -| `skill_dirs=[Path(...)]` | Searches only these roots, in order; the first skill of a given name wins. | -| `skill_dirs=None` | Searches project, user, and builtin roots. `skill_dirs=[]` disables discovery. | -| `tape_store=...` | Uses the supplied `TapeStore` or `AsyncTapeStore`. `FileTapeStore(path)` persists session tapes in that directory. | -| `tape_store=None` | Uses the framework's active store, or an instance-local memory store if no store is active. | +| Parameter | Behavior | +| ------------------------ | ---------------------------------------------------------------------------------------------------------------------- | +| `tools=[...]` | Accepts `Tool` objects. `Tool.from_callable()` derives a schema from annotations and a description from the docstring. | +| `tools=None` | Copies the global tool registry at construction time. `tools=[]` disables tools. | +| `skill_dirs=[Path(...)]` | Searches only these roots, in order; the first skill of a given name wins. | +| `skill_dirs=None` | Searches project, user, and builtin roots. `skill_dirs=[]` disables discovery. | +| `tape_store=...` | Uses the supplied `TapeStore` or `AsyncTapeStore`. `FileTapeStore(path)` persists session tapes in that directory. | +| `tape_store=None` | Uses the framework's active store, or an instance-local memory store if no store is active. | Unlike `@tool`, `Tool.from_callable()` does not register the tool globally. Include `skill_describe` in the tool set when the model needs to load skill bodies on demand. diff --git a/website/src/content/docs/zh-cn/docs/build/sdk.md b/website/src/content/docs/zh-cn/docs/build/sdk.md index 381195b7..1a097a29 100644 --- a/website/src/content/docs/zh-cn/docs/build/sdk.md +++ b/website/src/content/docs/zh-cn/docs/build/sdk.md @@ -36,9 +36,8 @@ from bub.store import FileTapeStore from bub.tools import Tool -class ApplicationHooks(BuiltinImpl): - def __init__(self, framework: BubFramework, prompts: list[str]) -> None: - super().__init__(framework) +class SystemPrompts: + def __init__(self, prompts: list[str]) -> None: self.prompts = tuple(prompts) @hookimpl @@ -57,11 +56,11 @@ def create_agent() -> tuple[BubFramework, Agent]: framework = BubFramework(config_file=root / "config.yml") framework.workspace = root framework.plugin_manager.register( - ApplicationHooks(framework, [ + SystemPrompts([ "You are an order assistant. Reply directly to the user.", "Use lookup_order to check order status before answering.", ]), - name="application", + name="system_prompts", ) agent = Agent( framework, @@ -95,7 +94,7 @@ if __name__ == "__main__": `config.yml` 不存在时也可以运行，环境变量配置仍然生效。 示例中的订单查询返回演示数据，接入应用时请替换为实际服务。 -`ApplicationHooks` 覆盖 builtin 的系统提示方法，并继承其他 builtin hooks。 +`SystemPrompts` 覆盖 builtin 的系统提示方法，并继承其他 builtin hooks。 只注册这一个实例即可；这个示例中不要再调用 `load_builtin_hooks()` 或 `load_hooks()`。 系统提示 hooks 会累加，单独新增一个 prompt hook 会保留默认 builtin 的 channel 指令和工作区 `AGENTS.md` 内容。 @@ -104,14 +103,14 @@ if __name__ == "__main__": ## 工具、Skills 与会话 -| 参数 | 行为 | -| --- | --- | -| `tools=[...]` | 接受 `Tool` 对象。`Tool.from_callable()` 从类型注解生成 schema，从 docstring 获取描述。 | -| `tools=None` | 构造时复制全局工具注册表。`tools=[]` 禁用工具。 | -| `skill_dirs=[Path(...)]` | 仅搜索这些目录，按顺序处理，同名 skill 使用第一个。 | -| `skill_dirs=None` | 搜索项目、用户和 builtin 目录。`skill_dirs=[]` 禁用发现。 | -| `tape_store=...` | 使用传入的 `TapeStore` 或 `AsyncTapeStore`。`FileTapeStore(path)` 将会话 tape 持久化到指定目录。 | -| `tape_store=None` | 使用 framework 当前的 store；没有活动 store 时使用实例独立的内存存储。 | +| 参数 | 行为 | +| ------------------------ | ------------------------------------------------------------------------------------------------ | +| `tools=[...]` | 接受 `Tool` 对象。`Tool.from_callable()` 从类型注解生成 schema，从 docstring 获取描述。 | +| `tools=None` | 构造时复制全局工具注册表。`tools=[]` 禁用工具。 | +| `skill_dirs=[Path(...)]` | 仅搜索这些目录，按顺序处理，同名 skill 使用第一个。 | +| `skill_dirs=None` | 搜索项目、用户和 builtin 目录。`skill_dirs=[]` 禁用发现。 | +| `tape_store=...` | 使用传入的 `TapeStore` 或 `AsyncTapeStore`。`FileTapeStore(path)` 将会话 tape 持久化到指定目录。 | +| `tape_store=None` | 使用 framework 当前的 store；没有活动 store 时使用实例独立的内存存储。 | 与 `@tool` 不同，`Tool.from_callable()` 不会把工具注册到全局表。 模型需要按需读取 skill 正文时，将 `skill_describe` 加入工具集合。