Hooks

​

Add hooks to a call

from unmute_core.call import CallHooks
from unmute_core.hooks import Hooks
from unmute_core.livekit import UnmuteAgent

hooks = Hooks()


async def entrypoint(ctx: JobContext) -> None:
    call = CallHooks(state, hooks=hooks)
    agent = UnmuteAgent(spec, call=call)

from unmute_core.call import CallHooks
from unmute_core.hooks import Hooks
from unmute_core.pipecat import UnmuteFlows

hooks = Hooks()


async def bot(runner_args: RunnerArguments) -> None:
    call = CallHooks(state, hooks=hooks)
    flows = UnmuteFlows(spec, llm=llm, context=context, aggregators=aggregators, call=call)

1. Many calls may share one Hooks. It holds only your functions.
2. A CallHooks binds to one call. Build a new one for each call.
3. A hook may be sync or async, except a prompt hook, which must be a plain function, and a wrap_* hook, which must be async.

​

Add a line after every prompt

from unmute_core.hooks import PromptSuffix

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

1. The suffix goes after the agent’s prompt, every task step’s, the finish checks’ and the summaries’, after a blank line.
2. With model=, only prompts that model reads get it.
3. It refuses blank text, more than 512 characters, and a {{...}} placeholder.

​

Change a model request

from dataclasses import replace

from unmute_core.model import ModelRequest


@hooks.before_model
def calm(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. Return None to keep the request as it is.
2. request.params are extra body fields for this request only.
3. wrap_model hooks run around the request, and after_model hooks hear the answer.

​

Refuse or change a tool call

from unmute_core.hooks import Refuse, ToolRun


@hooks.before_tool
def no_refunds_at_night(run: ToolRun) -> Refuse | None:
    if run.name == "refund" and closed_now():
        return Refuse("Refunds open at 9.")
    return None

1. Refuse(message) stops the call. The tool never runs, and the model reads the message.
2. Return a changed ToolRun to change the arguments. They are checked against the schema again.
3. wrap_tool hooks run around the tool, as async (run, handler). Call await handler(run) to run it.

​

Run code when a tool ends

@hooks.after_tool
def audit_refund(run: ToolRun, result: object) -> None:
    if run.name == "refund":
        audit.write(run.arguments, result)

1. Return None to keep the result. Return something else to replace what the model reads.
2. An after_tool hook that raises is logged, and the model reads the tool’s own result.

​

Watch call events

from unmute_core.call import CallEnd


@hooks.on("tool.*")
def log_tool(event: str, data: dict) -> None:
    logger.info("%s %s", event, data["name"])


@hooks.on("call.ended")
async def report(event: str, end: CallEnd) -> None:
    await crm.save(end.reason, end.values)

1. A pattern may hold *, like task.*. A pattern that matches no event raises ValueError.
2. Observers only watch. One that raises is only logged.
3. The end of the call waits for call.ended observers, up to end_timeout_s.

​

Where the model hooks run

​

Run tool hooks on a plain agent

from unmute_core.livekit import function_tools

agent = Agent(instructions=PROMPT, tools=function_tools(registry, state, hooks=call))

from unmute_core.pipecat import tools_schema

tools = tools_schema(registry, state, hooks=call)

1. hooks=call runs the tool hooks, and fires the tool.* events for your tools.
2. Framework tools skip the tool hooks: end call, handoffs and native_tools. So do prefetch lookups.

​

Where to go next