Tools reference

​

Use it in your agent

1. Build the tools once per call. They read that call’s state, and what the call carries from state.call.
2. Every option on this page works in all four kinds of agent: UnmuteAgent, a plain LiveKit Agent, UnmuteFlows and a plain Pipecat pipeline. The sections below say where one kind of agent needs one more line.
3. call is the call’s CallHooks. It runs your tool hooks and lets background webhooks finish when the call ends. state and call are needed only when a tool reads them. On a plain agent, without state, the tools take the state of call.
4. A reply line says dates and times in the call’s language. function_tools and tools_schema take language= and timezone= for it. On UnmuteAgent and UnmuteFlows, the spec’s AgentSpec(language=, timezone=) sets them.

​

What it handles for you

​

Inputs and outputs

from datetime import date
from typing import Annotated, Literal

from pydantic import BaseModel, Field

from unmute_core.state import Phone
from unmute_core.tools import local_tool


class Booking(BaseModel):
    reference: str
    day: date


def book_table(
    day: Annotated[date, Field(description="The day the caller wants.")],
    party: Annotated[int, Field(ge=1, le=12)] = 2,
    seating: Literal["inside", "terrace"] = "inside",
    phone: Phone | None = None,
) -> Booking:
    """Book a table."""
    return Booking(reference="B-12", day=day)


tool = local_tool(book_table)

1. local_tool builds a Pydantic model from the type hints. Its JSON Schema is what the model reads.
2. Field(description=...) tells the model what an argument is. Limits like ge and le go into the schema too.
3. Each call is checked before your function runs. Your function gets real Python types: a date, an int, a clean Phone.
4. A bad argument goes back to the model with what to fix, like “party: 40 is greater than the maximum of 12”. The model asks the caller or calls again.
5. Every parameter needs a type. A parameter with none raises TypeError that names it. *args and **kwargs are refused too.
6. The tool name is the function name, and the description is the docstring up to its Args: section. name= and description= change them.

​

Outputs

def find_order(order_id: str) -> Order:  # the return annotation is the output
    ...


def find_raw(order_id: str) -> dict:  # a function that returns a dict names it
    ...


local_tool(find_raw, output=Order)

1. local_tool reads the output from the return annotation. A function that returns a dict passes output=. With neither, local_tool raises TypeError: “name its output”.
2. A result that succeeded is checked against the output when the tool returns. The model then reads the checked result, output.model_dump(mode="json"). Defaults are filled, and a field the model does not declare is dropped, unless the model sets extra="allow".
3. A result that does not fit fails the call. The model reads “find_order returned a result that does not fit Order: status: … Do not claim success.” The log names the tool and the field, never a value.
4. A result that reports an error is never checked: an error or a refused with text in it, or "ok": false. For a failure in your own function, raise ToolError.
5. Done is the output of a tool that only runs: {"ok": true}. A webhook that answers with an empty body returns it. Accepted is the output of a webhook that does not wait for its answer. Both come from unmute_core.tools.
6. The wrap_tool and after_tool hooks see the checked result. What a hook puts in its place is not checked again.
7. The rule covers every library tool. A framework tool passed as native_tools= is the framework’s own object, and the library does not check it.

​

One input model

class OrderQuery(BaseModel):
    order_id: Annotated[str, Field(description="The order id the caller reads out.")]


async def lookup_order(query: OrderQuery) -> Order:
    return await orders.get(query.order_id)


tool = local_tool(lookup_order)

​

Inputs and outputs of a webhook

webhook_tool("lookup_order", "Look up an order by its id.", URL, input=OrderQuery, output=Order)

1. Use precise types: date, Literal, Phone, and Field limits. Each one is a check you do not write.
2. Give every argument a description. It is the only place the model learns what the argument means.
3. Type the result’s dates and choices too: date, time, Literal. A reply line then says them as a person does.
4. Raise ToolError for a failure the model should read. Do not return an error dict from a function typed to return a model.

​

What a tool does around its call

1. With several announce lines, the agent never says the same one twice in a row.
2. A failed tool never ends the call. The model reads the error and goes on.
3. interruption matters on Pipecat, where cancel is the default. continue lets the tool finish. The library keeps Pipecat’s own guidance about background tools out of the prompt, so the model reads the same prompt as on LiveKit. That holds only while the library’s tools are the only background tools. Register your own tool with cancel_on_interruption=False and Pipecat’s guidance comes back for all of them. LiveKit never cancels a running tool. When the reply that called it is cut off, the model reads that the tool is still running, and its result comes in a later reply.
4. A value that is not one of the listed words raises a Pydantic ValidationError. It names the field and the allowed words. An unknown argument raises TypeError.

​

A tool the caller talks over

1. Use interruption="continue" for a tool that changes something: a booking, a payment, a transfer.
2. Leave provider_default for a lookup the model can simply ask again.

• UnmuteFlows keeps Pipecat’s background-tool text out of the prompt by itself.
• A plain Pipecat pipeline needs attach_call(..., llm=llm) for that. Without it, Pipecat adds its own text about background tools to the prompt.
• On LiveKit, both kinds of agent need nothing more.

​

Announce lines with values

local_tool(
    check_day,
    announce="Let me check {{day}} for {{brand}}. Call {{call.session_id}}.",
    announce_wait=True,
)

1. {{value}} reads a State value or an argument of this call. {{call.name}} reads what the call carries, like {{call.session_id}}.
2. The line is filled as it is said.
3. If a line reads a value that is not set, the call is refused. Nothing is said and the tool does not run.
4. With announce_wait=True, LiveKit waits for the speech to end, and the caller cannot cut the line off. Pipecat waits for the bot to stop speaking, at most 30 seconds. On Pipecat, user_mute_strategies(protect=["tool_calls"]) keeps the caller from cutting the line.
5. A realtime model that speaks itself skips the line and words its own. See Realtime models.

1. Announce only a tool the caller would notice waiting for. A fast lookup needs no line.
2. Keep each line short, and give two or three for a tool the agent calls often.
3. Use announce_wait=True only when the line must come before the action, like “Putting you through now.”

​

A result the model does not read

local_tool(
    save_order,
    show_result=False,
    result_instructions="Tell the caller the order is saved. Do not read out the number.",
)

1. With show_result=False, the model reads {"ok": true} for a result that succeeded.
2. A failure or a refusal is always shown. The model never claims a success. A result is a failure when it has an error with text in it, a refused with text in it, or "ok": false. {"error": null} and an empty error are a success.
3. Task code still reads the full result. Finish rules and ToolReply lines work as before.
4. result_instructions is added to the description as "<description>\n\nUse the result like this: <instructions>".
5. ToolRegistry.for_model(name, result) gives what the model reads. The adapters call it.
6. hidden=("read_back",) leaves only those fields out of a result that succeeded. Use it for a line code speaks from the result, like a read-back: a model that can see the words may say them itself. assign rows and spoken lines still read the field. A field the output model does not have fails when the tool is built.

​

End the call

hang_up = local_tool(hang_up_handler, effect="ends_conversation")

from unmute_core.pipecat import end_call_tool, reply_text

tool = end_call_tool("Call it when the caller is done.", "Goodbye!", reply_text(llm))

​

A call that is refused

{"refused": "day not set yet. Get it from the caller, then call this tool again."}

1. The tool does not run, and nothing is said. The model reads the message and acts on its hint.
2. If a step must still confirm the value, the hint is “Run first: it confirms with the caller”.
3. Else, if a task saves the value, the hint is “Run first: it saves ”. The first task that saves it wins.
4. Else the hint is “Get it from the caller”, with the value’s description from the State model in brackets when it has one.
5. Every announce line of the tool is filled before one is picked. So every line’s values must be set, even when another line needs none.
6. A refused call counts as a failure. ends_conversation never fires on it.
7. In code, ToolRefused is a ToolError. LiveKit and Pipecat send the same JSON.

1. In your own tool, raise ToolError with the words the model should read, like “No table is free that day. Offer another day.”
2. Let any other error raise. The model reads that the tool failed, and the log keeps the details.
3. Use show_result=False for a result the caller must not hear. See A result the model does not read.

​

The schema the model reads

1. An optional value whose default is null loses its null branch and that default. Leaving it out means the same thing. A value without that default keeps null, because there null means something, like “clear this value”.
2. format and pattern become words in the description. Some providers refuse them, and a format only this library knows, like spelled, tells a model nothing.
3. Local $refs are inlined.

1. A null branch on every optional value makes the model choose between leaving a value out, sending null and guessing. Small models often guess.
2. A keyword the provider does not support can make it refuse the call, or return no call at all. The caller then hears silence.
3. A format name the model has never seen, like spelled, tells it nothing. Plain words do.

​

A required argument left out or empty

{"refused": "birth_date is missing. Ask the caller for it, then call lookup_policy again.",
 "missing": ["birth_date"]}

1. A required argument that is absent, or text that is blank, is missing. The tool does not run. It raises ArgumentsMissing, a ToolRefused.
2. The model reads this refusal, not a format error like “expected a day written year-month-day”.
3. When the reply that made the call already said something, no model reply follows. That reply asked the caller, so the caller answers next. replies_to(result, spoke=...) from unmute_core.tools holds this rule.
4. When the reply said nothing, the model replies to the refusal and asks.
5. A step’s closing tool with confirm lines refuses the same way, before it asks anything.
6. On Pipecat, tools_schema cannot tell whether the reply spoke, so the model always replies there. So does OpenAI’s live service, which cannot take a result without replying.

​

Save a result and speak it

from unmute_core.tools import Assign, ToolRegistry, ToolReply

registry = ToolRegistry(
    [local_tool(find_order)],
    state=State,
    assign={
        "find_order": [
            Assign(variable="order", path=()),
            Assign(variable="order_number", path=("order_id",)),
        ]
    },
    replies=[
        ToolReply(tool="find_order", when={"status": ["held"]}, say=["{{result.order_id}} is held."]),
        ToolReply(tool="find_order", state={"vip": [True]}, say=["{{name}}, it arrives {{result.arrives}}."]),
        ToolReply(tool="find_order", say=["Your order arrives {{result.arrives}}."]),
    ],
)

1. assign maps a tool to the State values its result saves. path walks the result by field names. path=() saves the whole result into a State value typed as the output model, like order: Order | None = None.
2. The values of one result are saved in one batch, or not at all. A save the State refuses is logged and told to the sinks as tool.save_refused. The model still reads the result, because the tool ran. A tool’s save never confirms a value.
3. replies are tried in order. The first one that matches the result and fills is spoken. The model reads the result and is asked for no reply.
4. when maps a result field to the values that call for the line. state maps a State value to its values the same way. A reply with neither is the fixed reply. A line whose value is missing does not play, and the next reply is tried.
5. A line reads {{result.<field>}}, nested like {{result.order.id}}, and {{variable}} from the State, nested like {{account.amount_words}}. Several lines are alternatives, said in turn.
6. A value is said by its output field’s JSON Schema format, in the call’s language. A date reads “tomorrow, October 16” with a timezone and “Friday, October 16, 2026” without one. A date-time adds the time, a time reads “9:00 AM”, a phone is grouped, and a Literal or Enum value is said as words. Any other value is said as it is.
7. The registry checks every row when it is built. A tool, a result field or a State value that does not exist, or a result value whose type cannot fill the State value, fails with a message that names the fix.
8. A reply on a tool that ends the call is refused: the call is over before the line could play. Say a goodbye with the tool’s announce and announce_wait=True.
9. A result that reports an error saves nothing and speaks nothing. Inside a task step, a tool that does not end the step still saves its registry assign rows at once, so a step that asks several questions keeps each answer as it comes. The step’s own replies are spoken, then the tool’s own, and the step’s own assign saves when the step ends.
10. A when value on a Literal, bool or Enum field must be one the field can hold. A typo like "shiped" fails when the registry is built, with the values the field holds.
11. Each line code speaks is told to the sinks as tool.replied, so a trace shows which turns cost no model reply.

​

Replies on the tool itself

class Noted(BaseModel):
    status: Literal["noted", "future_day"] = "noted"
    complete: bool
    next_question: str = ""


local_tool(
    note_facts,
    replies=[
        ToolReply(when={"status": ["future_day"]}, say=["That day hasn't happened yet. On which day did it happen?"]),
        ToolReply(when={"complete": [False]}, say=["{{result.next_question}}"]),
    ],
)

1. A tool’s own reply leaves out tool. One that names another tool is refused.
2. The place’s replies are tried first: the registry’s on the main agent, the TaskSpec’s in a step. Then the tool’s own, in order.
3. A reply on a ToolRegistry or a TaskSpec still names its tool.

​

Check a rule before the tool runs

def outside_window(day: date) -> Recorded | None:
    if day < today():
        return Recorded(status="past_day")
    return None


local_tool(
    ledger.record_promise,
    precheck=outside_window,
    replies=[ToolReply(when={"status": ["past_day"]}, say=["That day has already gone. Which other day would suit you?"])],
)

1. The precheck is a plain function, not async. It takes some of the tool’s parameters by name, with the same types, or the whole input model when the tool takes one. Locked values are filled.
2. It returns an instance of the tool’s output model, the outcome, or None to go ahead. An outcome is the tool’s result: the tool does not run, and no announce plays.
3. The tool’s replies say the outcome, like any result. A when value is checked against the output’s Literal at build time.
4. On a closing tool with confirm, the precheck runs before the read-back. A wrong value is never read back, and the turn costs one model request: the call. It runs again on the caller’s yes, since the state may have changed.
5. An outcome must not match the tool’s finish rule. It is a refusal, not a success.

1. Use a reply for a line that must come out word for word, like a booking’s day and time.
2. Save the whole output with path=() when later prompts read several of its fields, like {{order.status}}.
3. Set language= and timezone= on the spec, or on function_tools and tools_schema, so “tomorrow” means the business’s tomorrow.

​

Locked arguments

webhook_tool(..., locked={"customer_id": "{{customer.id}}", "call": "{{call.session_id}}"})

1. The model never sees a locked argument. A value it sends is replaced.
2. {{name}} and {{name.field}} read the call state. The tool is refused, with a hint, while the value is unset.
3. {{call.name}} reads what the call carries, like {{call.remote_number}} or {{call.carrier_call_id}}. It reads state.call, so every adapter has it.
4. A tool that locks an empty call value is refused: this tool reads call.carrier_call_id, which this call does not carry.
5. {{@name}} no longer exists. A tool that uses it fails when it is built, and the message names the new spelling. {{@call_id}} is now {{call.carrier_call_id}} or {{call.session_id}}. {{@phone_number}} is {{call.remote_number}}. {{@transcript}} is Transcript().
6. A ToolRegistry built with state= checks each locked path when the registry is built. The path must exist, and its type must fit the parameter.
7. A path may name a whole shape or a list, like "{{customer}}". Lock it as the only text, to a parameter that is an object, an array or has no type. The tool then gets the object or the list.
8. A shape or a list inside a longer text, like "Customer: {{customer}}", fails at build.

1. Lock every value the caller does not choose: customer ids, account numbers, the call id, the brand.
2. Build the ToolRegistry with state=. A typo in a path then stops the agent at startup, not in the middle of a call.

​

Send the transcript

from unmute_core.tools import Transcript

webhook_tool(
    ..., locked={"transcript": Transcript(max_messages=50), "reason": "{{call.end_reason}}"}
)

​

Webhooks

webhook_tool(
    "notify_crm",
    "Tell the CRM about the booking.",
    "https://crm.example.com/hooks/booking",
    input=BookingNote,
    output=Accepted,
    auth=WebhookAuth.api_key("CRM_KEY", header="X-API-Key"),
    body="envelope",
    wait_for_response=False,
)

​

webhook_tool

from unmute_core.tools import WebhookOptions

options: WebhookOptions = {"timeout": 5, "wait_for_response": False}
webhook_tool("notify_crm", "Tell the CRM.", URL, input=BookingNote, output=Accepted, **options)

1. The URL is checked when the tool is built. The scheme must be https, or http with allow_private_network. The host must be there. User info, a fragment and a port outside 1 to 65535 are refused.
2. When the webhook answers with an error, the model reads the status and a short excerpt of the body. The excerpt is the error, message and code of a JSON body, or the first 300 characters of the text.
3. The tool events keep only the status. They never carry text that another service sent. In code, ToolError(message, event_text=...) sets the two apart.

1. A GET tool whose input or locked names a query value of the URL fails at build. Rename the argument or drop the value from the URL.
2. A call whose argument clashes with a URL query value is refused with an error that names the argument.
3. With auth, a fixed Authorization header fails at build, in any letter case. auth sends the credential.

​

WebhookAuth

async def fetch_token() -> str:
    return await vault.read("crm-token")


auth = WebhookAuth.bearer(token=fetch_token)
auth = WebhookAuth.api_key(header="X-Key", token=fetch_token)

​

Time and size limits

webhook_tool(..., timeout=5)
webhook_tool(..., wait_for_response=False)

1. timeout is one deadline for the whole request: DNS, the connection, every redirect and the answer. It is 10 seconds by default.
2. The request and the answer may each be 64 KiB at most. A larger one fails the tool call, and the model reads the error.
3. With wait_for_response=False, the model reads {"ok": true, "accepted": true} at once. The request goes on in the background. The tool’s output is then Accepted.

1. Keep the 10 second default. Lower it for a lookup the caller waits on.
2. Use wait_for_response=False for a notification nobody waits for, like telling a CRM about a booking.

​

Private addresses

webhook_tool(..., "http://localhost:8000/hooks/booking", allow_private_network=True)

1. allow_private_network is off by default.
2. Off, the webhook refuses private, loopback and link-local addresses. It refuses plain http too.
3. On, it accepts them.

1. Keep it off in production.
2. Turn it on only for your own internal service, or for localhost while you develop.
3. Never turn it on when someone else writes the URL.

​

Built-in tools

from unmute_core.tools import current_datetime_tool, user_phone_number_tool

registry = ToolRegistry([current_datetime_tool("Europe/Madrid"), user_phone_number_tool()])

1. Give current_datetime_tool an IANA zone, like Europe/Madrid. Another name raises ValueError.
2. call_type is inbound or outbound, and null when the call has no direction.
3. When the call has no phone number, user_phone_number_tool returns {"available": false, "phone_number": null, "call_type": null, "reason": ...}.
4. Both take name= and description= to change what the model reads. The defaults are current_datetime and user_phone_number.
5. The tool reads state.call, so the adapter needs nothing more.

​

Watch tool calls

def show(stage: str, data: dict) -> None:
    print(stage, data)


registry.listen(show)

1. The sink gets names and timing only. It never gets argument values. source is model, or event for a tool on an event.
2. ms is the time the tool ran. A wait for the announce line is not in it. A cancelled call ends with tool.failed and the error cancelled.
3. A sink that raises is logged. It never breaks the call. A sink the registry already holds is not added again.
4. only() and with_tools(*tools) keep the sinks.
5. In a call, use the call’s hooks instead. See Hooks.

​

Advanced

​

Rules for a hand-written schema

1. definitions and $id are refused. Use $defs and local #/$defs/... references.
2. format and the notes are hints. Nothing checks them.
3. nullable is not JSON Schema, so it is refused. Write "type": ["<type>", "null"] instead.
4. The root of the parameters is a plain object schema. It may hold only $schema, $defs, type ("object"), properties, required, additionalProperties, description, title, $comment and x- keys. Frameworks send only the root properties and required, so another root keyword, like oneOf, anyOf, allOf, not, patternProperties, propertyNames, minProperties, maxProperties or if, would be lost on one of them. It fails when the tool is built. Move it into a property, or wrap the choice in one property. Nested schemas may use every keyword in the table.

1. Build the schema from a Pydantic model with model_json_schema(), so it cannot drift from your code.
2. Keep the root a plain object of named properties. Put a choice, like oneOf, inside one property.

​

Arguments the schema does not name

1. The model may send an argument the schema does not list. The call is refused with a message that names it: “bogus is not an argument of lookup. Call lookup again without it.” The model reads this and calls again.
2. A schema stays open when it sets additionalProperties to true or to a schema, or when it has no properties. It then keeps the extra arguments.
3. A custom ToolExecutor with checks_extra_arguments = True gets the extra arguments. It then refuses them in its own words.

​

Ask the registry

"lookup_order" in registry  # True when the registry has a tool of this name
registry.keeps_running("lookup_order")  # True when its interruption is "continue"

1. name in registry tests for a tool by name.
2. keeps_running(name) says whether the caller’s speech must leave a running call of the tool alone. An unknown name gives False, so the call is cancelled.
3. The adapters use both. You need them only when you write your own adapter.

​

Your own kind of tool

from typing import Any

from unmute_core.state import CallState
from unmute_core.tools import ToolError, ToolSpec


class ServiceTool:
    """Runs a tool through your own tool service."""

    def __init__(self, client: ServiceClient, tool_id: str) -> None:
        self._client = client
        self._tool_id = tool_id

    async def run(
        self,
        arguments: dict[str, Any],
        state: CallState | None,
        *,
        context: dict[str, object] | None = None,
    ) -> object:
        result = await self._client.invoke(self._tool_id, arguments, context=context)
        if not result.ok:
            raise ToolError(result.message_for_the_model)
        return result.data


spec = ToolSpec("book", "Book a table.", PARAMETERS, Booking, ServiceTool(client, "book"))

1. run gets the model’s arguments, with the locked values filled and checked against the schema. It must be an async method. A ToolSpec refuses an executor with no async run.
2. Raise ToolError with the words the model should read. Any other error becomes “failed before completing. Do not claim success.”
3. Add a context keyword to run to get the event context. Without it, run gets only the arguments and the state.
4. Add validate(arguments) -> dict to refuse bad input before a step asks the caller to confirm it.
5. Set checks_extra_arguments = True on the class to get arguments the schema does not name. Without it, the registry refuses them first.
6. The executor never changes what the model sees. The output model, the announce line, the effect and the interruption rule stay on the ToolSpec, the same for every kind of tool. PARAMETERS is the JSON Schema of the arguments, and Booking the output model.

​

Troubleshooting

​

A tool call fails with “does not fit”

​

A tool call returns refused and nothing is said

​

A webhook call fails with “could not authenticate”

export ORDERS_TOKEN="..."

​

A webhook to localhost is refused

webhook_tool(..., "http://localhost:8000/hooks/booking", allow_private_network=True)

​

Where to go next