Hooks reference

​

What it gives you

​

How it works

1. prompt hooks change each system prompt the library builds: the agent’s, each task step’s, a finish check’s and a summary’s.
2. before_model hooks change the request about to go out. wrap_model hooks run around it. after_model hooks hear the answer.
3. before_tool hooks change or refuse a tool call. wrap_tool hooks run around it. after_tool hooks may replace its result.
4. before_call hooks run before the greeting. wrap_failure hooks run around the failure line.
5. on observers watch events. They never change anything.

​

The points

1. before_* and prompt hooks run first to last.
2. wrap_* hooks nest. The first one added is the outermost.
3. after_* hooks run last to first.
4. Any hook may be sync or async, except prompt, which must be a plain function, and wrap_*, which must be async.
5. A hook that raises fails its own request or tool call. A model request that fails this way fails the call, as a failed model call does. A tool call becomes a ToolError the model reads.
6. An after_tool hook that raises is logged, and the model reads the tool’s own result, because the tool already ran. An after_model hook or an observer that raises is only logged.
7. On Pipecat, an after_model hook reads each tool call’s arguments as UTF-8 JSON, like {"name": "José"}. It never sees é escapes. The Pipecat span and LiveKit show the same text.
8. ModelRequest and ToolRun are frozen Pydantic dataclasses. dataclasses.replace(request, ...) builds a new object and checks it again. A wrong type, like site=5, raises a ValidationError. An unknown argument raises TypeError. messages, tools and params are not checked, so a hook may put any JSON in them.

​

Hooks around a tool call

result = await registry.call("refund", {"order_id": "A-1"}, state, hooks=call)

1. ToolRegistry.call(..., hooks=...) runs the call’s tool hooks in this order: before_tool, then the checks and the locked values, then wrap_tool around the tool, then after_tool.
2. A before_tool hook that returns Refuse(message) stops the call. The model reads {"refused": message}, as for a value not set yet.
3. A hook that changes the arguments gets them checked against the schema again.
4. Every adapter passes the call’s CallHooks. So the hooks reach every tool of UnmuteAgent and UnmuteFlows with no more wiring.
5. A ToolRun holds name, arguments as the model wrote them, and source. The source is model, or event for a tool on an event.

​

Writing hooks as a class

class Audit:
    events = ("tool.*",)

    def before_model(self, request: ModelRequest) -> ModelRequest:
        return replace(request, params={"temperature": 0})

    def on_event(self, event: str, data: dict) -> None:
        log.info("%s %s", event, data["name"])


hooks = Hooks(Audit(), EventTool(log_call, on=["call_end"]))

1. hooks.add(obj), or Hooks(obj, ...), hooks the object in with each method named after a point.
2. An on_event(event, data) method watches the patterns in its events attribute. All events when it has none.
3. An EventTool is a hook too. See The call’s life.
4. An object with none of these raises ValueError.

​

Events

1. A pattern may hold *, like task.*.
2. A pattern that matches no event raises ValueError when you register it.
3. A sync observer runs at once. An async one runs as a task of its own, except for call.ended, which the end waits for. Another async observer still running when the process stops is cut off.

​

A suffix after every prompt

hooks.add(PromptSuffix("/no_think"))
hooks.add(PromptSuffix("/no_think", model="qwen/qwen3-32b"))

1. It reaches the agent’s prompt, every task step’s, the finish checks’ and the summaries’.
2. With model=, only prompts that model reads get it. A summarizer of another model keeps its prompt as it is.
3. It refuses blank text, more than 512 characters, and a {{...}} placeholder. Each refusal says what to do instead.
4. The native path and a ModelCall give the model the same final prompt.

​

Rules for one model

@hooks.before_model
def no_think(request: ModelRequest) -> ModelRequest | None:
    if "nemotron" not in request.model:
        return None
    return replace(request, params=request.params | {"temperature": 0})

1. request.model names the model that answers, like gpt-5.6-luna.
2. request.params are extra body fields for this request only.
3. RouterModelCall sends them. A request’s params win over the router call’s body. See SLNG.
4. A native LLM gets them as extra_body, when it is OpenAI compatible.

​

Many calls at once

1. A Hooks object holds only your functions. Many calls may share it, in one process or across threads.
2. Everything a call changes lives in its own objects: CallHooks, CallState, the adapter with the CallTasks it builds and, on Pipecat, the LLM service. Each call builds its own. Build the AgentSpec per call too when it holds a model, because the adapter closes its models when the call ends.
3. The library refuses a second call on the same object. A CallHooks binds once. CallTasks.use takes one call’s hooks. A Pipecat LLM service runs one call’s hooks.
4. A hook that keeps state, or needs to know its call, belongs to that call. Build it per call, together with the shared hooks:

SHARED = Hooks(PromptSuffix("/no_think"))


async def entrypoint(ctx: JobContext) -> None:
    hooks = Hooks(SHARED, RoomLog(ctx.room.name))  # your class, one per call
    call = CallHooks(state, hooks=hooks)

​

Hooks

​

PromptSuffix

​

Refuse

​

PromptInfo

​

ModelResponse

​

Known limits

1. wrap_model reaches only requests a ModelCall answers. A framework’s own LLM keeps its own retries and fallback.
2. On native Pipecat:
   • a hook that changes the tools is ignored, with one warning.
   • the system prompt must stay the first message, and a hook sees the service’s own prompt, without the guidance Pipecat adds to it.
   • params reach only OpenAI-compatible services.
3. On native LiveKit, params need an OpenAI-compatible plugin, and the plugin’s own extra_kwargs win.
4. Framework tools (end call, handoffs, native_tools) and prefetch lookups skip the tool hooks.
5. Under FallbackModelCall, before_model runs once. Every fallback gets the same params, and request.model names the first.
6. With slng/auto, the client does not know which model the router picks. A rule for one model then belongs in the router.

​

Troubleshooting

​

ValueError: matches no event

@hooks.on("tool.*")
def watch(event: str, data: dict) -> None: ...

​

ValueError: hooks in nowhere

​

TypeError: a prompt hook is async

​

A second call refuses the same CallHooks

call = CallHooks(state, hooks=Hooks(SHARED, RoomLog(room_name)))

​

Where to go next