Hooks

Turn Hooks

resolve_session

def resolve_session(self, message: Envelope) -> str:

load_state

def load_state(self, message: Envelope, session_id: str) -> TurnState:

build_prompt

def build_prompt(self, message: Envelope, session_id: str, state: TurnState) -> list[UserContent] | LegacyPrompt:

run_model

def run_model(self, prompt: str | list[dict], session_id: str, state: TurnState) -> str:

run_model_stream

def run_model_stream(self, prompt: str | list[dict], session_id: str, state: TurnState) -> AsyncStreamEvents:

continue_prompt

def continue_prompt(self, prompt: str | list[dict] | None, tape: Tape, state: StreamState) -> str | None:

save_state

def save_state(
    self,
    session_id: str,
    state: TurnState,
    message: Envelope,
    model_output: str,
) -> None:

render_outbound

def render_outbound(
    self,
    message: Envelope,
    session_id: str,
    state: TurnState,
    model_output: str,
) -> list[Envelope]:

dispatch_outbound

def dispatch_outbound(self, message: Envelope) -> bool:

App Hooks

register_cli_commands

def register_cli_commands(self, app: Any) -> None:

onboard_config

def onboard_config(self, current_config: dict[str, Any]) -> dict[str, Any] | None:

provide_model_options

def provide_model_options(
    self,
    session_id: str,
    workspace: Path | None,
) -> ModelOptions | None:

provide_environment

def provide_environment(self, session_id: str, workspace: Path) -> Environment | None:

on_error

def on_error(self, stage: str, error: Exception, message: Envelope | None) -> None:

system_prompt

def system_prompt(self, prompt: str | list[dict], state: TurnState) -> str:

provide_tape_store

def provide_tape_store(self) -> TapeStore | AsyncTapeStore | None:

provide_lifespan

def provide_lifespan(self) -> AsyncIterator[None] | Iterator[None] | None:

provide_tape_sidecar

def provide_tape_sidecar(self) -> TapeSidecar:

provide_channels

def provide_channels(self, message_handler: MessageHandler) -> list[Channel]:

build_tape_context

def build_tape_context(self) -> TapeContext:

admit_message

def admit_message(
    self,
    session_id: str,
    message: Envelope,
    turn: TurnSnapshot,
) -> AdmitDecision | None:

provide_steering_inbox

def provide_steering_inbox(self) -> SteeringInbox | None:

provide_tools

def provide_tools(self) -> list[Tool]:

Agent Hooks

before_llm_call

def before_llm_call(self, request: LLMCallRequest, state: TurnState) -> LLMCallRequest | LLMCallDecision | None:

LLMCallStarted.request

@dataclass(frozen=True)
class LLMCallRequest:
    run_id: str
    model: str
    messages: list[republic.Message]
    tool_names: tuple[str, ...] = ()
    max_tokens: int | None = None

LLMCallStarted.decision

@dataclass(frozen=True)
class LLMCallDecision:
    action: Literal["finish"] = "finish"
    text: str = ""

    @classmethod
    def finish(cls, text: str) -> LLMCallDecision: ...

after_llm_call

def after_llm_call(self, request: LLMCallRequest, result: LLMCallResult, state: TurnState) -> None:

LLMCallFinished.result

@dataclass(frozen=True)
class LLMCallResult:
    run_id: str
    text: str | None = None
    tool_calls: list[dict[str, Any]] = field(default_factory=list)
    usage: dict[str, Any] | None = None
    error: Exception | None = None
    duration_ms: int = 0

before_tool_call

def before_tool_call(self, call: ToolCall, state: TurnState) -> ToolCallDecision | None:

ToolCallStarted.call

@dataclass(frozen=True)
class ToolCall:
    run_id: str
    tool: str
    arguments: dict[str, Any]
    context: ToolContext | None = field(default=None, repr=False, compare=False)

ToolCallStarted.decision

@dataclass(frozen=True)
class ToolCallDecision:
    action: Literal["proceed", "replace", "deny"] = "proceed"
    arguments: dict[str, Any] | None = None
    result: Any = None
    message: str | None = None

    @classmethod
    def proceed(cls, arguments: dict[str, Any] | None = None) -> ToolCallDecision: ...

    @classmethod
    def replace(cls, result: Any) -> ToolCallDecision: ...

    @classmethod
    def deny(cls, message: str) -> ToolCallDecision: ...

after_tool_call

def after_tool_call(self, call: ToolCall, result: ToolCallResult, state: TurnState) -> None:

ToolCallFinished.result

@dataclass
class ToolCallResult:
    run_id: str
    tool: str
    arguments: dict[str, Any]
    result: Any = None
    error: Exception | None = None
    duration_ms: int = 0

App Hooks 与 Agent 事件

概念页只说明「谁在什么方向介入」，这里给出适配细节。三种数据的关系与决策语义见 Agent 事件。

事件映射

PluginAgentHooks 把应用 hooks 接到 Agent 事件上：

应用 hook	Agent 事件	说明
before_llm_call	LLMCallStarted	每次模型调用前；可返回改写后的 LLMCallRequest 或 LLMCallDecision
after_llm_call	LLMCallFinished	调用成功或普通异常失败后各一次
before_tool_call	ToolCallStarted	每次工具调用前；可 proceed(arguments)、replace(result) 或 deny(message)
after_tool_call	ToolCallFinished	成功、失败、拒绝与替代都会触发
continue_prompt	StepFinished	仅在 status == "continue" 时调用，结果写入 next_prompt
provide_steering_inbox	MessagesPending	排出待注入的 steering 消息，加入当前 Run 的下一步

模型可见的工具调用（包括 .tape.handoff 这类内置工具）同样经过 ToolCallStarted / ToolCallFinished。

两种接入方式的差异

对比项	直接处理 Agent 事件	应用拦截 hook
注册	agent.events.on(Type, handler)	插件中的 @hookimpl
执行顺序	注册顺序	pluggy 优先级
参数修改	原地替换 event.call / event.request	返回 ToolCallDecision.proceed(arguments=...) 或新的 LLMCallRequest
控制	设置 event.decision	返回 decision 对象
异常	传播给 emit() 调用方，可能使 Run 失败	记录日志后跳过，不会中断 Run
返回值类型不符	不适用	记录警告并跳过该实现

因此用 hook 实现的安全策略不能靠抛异常阻止调用，必须返回 deny(...)。

订阅方式

PluginAgentHooks(runtime).subscribe(events) 把四个拦截 hook 接到已有或新建的事件总线上；AgentPlugins.attach(agent) 会自动订阅，同时接上 continue_prompt、steering inbox 与系统提示词。嵌入方式与生命周期要求见复用 Bub 应用。

PluginAgentHooks

class PluginAgentHooks:
    def __init__(self, runtime: HookRuntime) -> None: ...

    def subscribe(self, events: EventBus | None = None) -> EventBus: ...

    async def before_llm_call(
        self,
        request: LLMCallRequest,
        state: TurnState,
    ) -> tuple[LLMCallRequest, LLMCallDecision | None]: ...

    async def after_llm_call(
        self,
        request: LLMCallRequest,
        result: LLMCallResult,
        state: TurnState,
    ) -> None: ...

    async def before_tool_call(
        self,
        call: ToolCall,
        state: TurnState,
    ) -> tuple[ToolCall, ToolCallDecision]: ...

    async def after_tool_call(
        self,
        call: ToolCall,
        result: ToolCallResult,
        state: TurnState,
    ) -> None: ...

AgentPlugins

AgentComponents

class AgentComponents(TypedDict, total=False):
    store: TapeStore | AsyncTapeStore
    context: TapeContext
    sidecars: Sequence[TapeSidecar]

AgentPlugins

class AgentPlugins:
    def __init__(
        self,
        framework: BubFramework | None = None,
        *,
        plugins: Iterable[object] | None = None,
    ) -> None: ...

    def running(self) -> AbstractAsyncContextManager[AsyncExitStack]: ...

    def components(self) -> AgentComponents: ...

    def attach(self, agent: Agent, *, tools: bool = True) -> Agent: ...

    async def build_state(
        self,
        session_id: str,
        *,
        message: Envelope | None = None,
        agent: Agent | None = None,
    ) -> TurnState: ...