Writing Your Own Tools

NeMo Labs Voice Agent gives you three ways to expose a Python callable to the large language model (LLM). Read Tool Calling first for backend requirements and the config flags that gate registration.

The first two paths both hand Pipecat a DirectFunction, so they share one contract.

Prerequisites

Before you add a tool, complete the following preparation:

1. Configure a supported backend and parser by following Tool Calling.
2. Decide whether the callable is standalone, owned by a component, or needs an explicit schema.
3. Identify the pipeline entrypoint where you want to register the tool.

The Direct-Function Contract

pipecat/adapters/schemas/direct_function.py validates and introspects every direct function at registration time. Violating the first two rules raises at startup, not at call time.

1. The function must be async.
2. Its first parameter must be named exactly params (it receives a FunctionCallParams).
3. It must deliver its result exactly once, by awaiting params.result_callback(...).

Everything the LLM sees is derived automatically:

Because the docstring is the prompt, write it for the model. State when the tool should be called, when it should not be called, and what the agent should do with the result.

1. Direct Functions

The shipped example is tool_get_city_weather in nemo_voice_agent/utils/tool_calling/basic_tools.py. It pushes a TTSSpeakFrame filler (“Looking up weather data for … Please wait a moment.”) before its network call, so the user is not left in silence. The tool wraps the HTTP call in asyncio.wait_for and returns {"error": ...} through result_callback on timeout or failure.

A minimal tool of your own:

1
# my_tools.py
2
from pipecat.services.llm_service import FunctionCallParams
3
4
5
async def tool_get_store_hours(params: FunctionCallParams, store_id: str, day: str = "today"):
6
    """Look up the opening hours of a retail store.
7
8
    Call this whenever the user asks when a specific store opens or closes.
9
    Do not call it for general questions about the company.
10
11
    Args:
12
        store_id: The store identifier, for example "SF-014".
13
        day: Day of the week, or "today". Defaults to "today".
14
    """
15
    await params.result_callback({"store_id": store_id, "day": day, "opens": "09:00", "closes": "21:00"})

Register it with the pipeline by adding it to the existing register_direct_tools_to_llm call in examples/generic_voice_agent/server/server.py (guarded by llm.enable_tool_calling):

1
from my_tools import tool_get_store_hours
2
from nemo_voice_agent.utils.tool_calling.mixins import register_direct_tools_to_llm
3
4
register_direct_tools_to_llm(
5
    llm=llm,
6
    context=context,
7
    tool_mixins=[tts],
8
    tools=[tool_get_city_weather, tool_get_store_hours],
9
)

The full signature (keyword-only, from nemo_voice_agent/utils/tool_calling/mixins.py):

1
def register_direct_tools_to_llm(
2
    *,
3
    llm: OpenAILLMService,
4
    context: LLMContext,
5
    tool_mixins: list[ToolCallingMixin] = [],
6
    tools: list[DirectFunction] = [],
7
    cancel_on_interruption: bool = True,
8
) -> None: ...

Tools already present on the context are preserved — the helper appends to context.tools before calling context.set_tools(...), so calling it twice accumulates rather than replaces.

2. Component-Owned Tools

When a tool has to reach into a live pipeline component, put it on the component. Mix ToolCallingMixin (nemo_voice_agent/utils/tool_calling/mixins.py) into the service and implement setup_tool_calling, which registers each bound method:

1
from pipecat.services.llm_service import FunctionCallParams
2
3
from nemo_voice_agent.utils.tool_calling.mixins import ToolCallingMixin
4
5
6
class MyService(SomeBaseService, ToolCallingMixin):
7
    def __init__(self, **kwargs):
8
        super().__init__(**kwargs)
9
        self.setup_tool_calling()  # must be called from __init__
10
11
    def setup_tool_calling(self):
12
        self.register_direct_function("tool_set_level", self.tool_set_level)
13
14
    async def tool_set_level(self, params: FunctionCallParams, level: int):
15
        """Set the processing level of the service.
16
17
        Args:
18
            level: An integer between 0 and 10.
19
        """
20
        self._level = level
21
        await params.result_callback({"success": True, "message": f"Level set to {level}"})

The mixin surface is three members:

Name matching: the function_name you pass to register_direct_function is only a bookkeeping key. register_direct_tools_to_llm appends the callable, and Pipecat names the tool after the Python method, so a mismatched key silently has no effect on what the model sees. Keep them identical.

The canonical example is KokoroTTSService in nemo_voice_agent/pipecat/services/nemo/tts.py. It registers six voice-control methods: tool_tts_set_speed, tool_tts_reset_speed, tool_tts_speak_faster, tool_tts_speak_slower, tool_tts_set_voice, and tool_tts_reset_voice. BaseNemoTTSService calls setup_tool_calling() from its constructor, so a subclass only overrides the method. Two patterns are worth copying. tool_tts_set_voice pushes an LLMTextFrame("Just a moment.") before reloading the model. It runs the blocking reload through asyncio.to_thread so the event loop continues serving audio.

Pass the instance in tool_mixins. Anything in that list that is not a ToolCallingMixin is skipped with a warning. The example server can therefore pass tool_mixins=[tts] unconditionally even when tts.type resolves to the hosted NVIDIA service.

3. Schema Tools

StandardSchemaTool in nemo_voice_agent/utils/tool_calling/base.py is the explicit-schema path. Subclass it and implement three members. The base builds the FunctionSchema and owns delivery:

1
class StandardSchemaTool:
2
    def __init__(self, *, description: Optional[str] = None, name: Optional[str] = None): ...
3
4
    @property
5
    def properties(self) -> Dict[str, Any]: ...
6
7
    @property
8
    def required_properties(self) -> List[str]: ...
9
10
    async def _execute(self, **kwargs: Any) -> Any: ...
11
12
    async def _after_result(self, params: FunctionCallParams) -> None: ...

Key differences from the direct path:

• The tool name comes from an optional class-level name attribute, falling back to the class name — so snake_case tool names are possible here.
• _execute is pure: it takes the call arguments as keyword arguments and returns a result. It must not touch params and must not call result_callback. __call__ delivers exactly once, and converts a raised exception into {"error": ...}. tests/unit/test_tool_call_contract.py pins this.
• Falsy results are wrapped before delivery, because Pipecat rewrites any falsy result to the literal string "COMPLETED" — which a model reads as success. A bare [] from a lookup that matched nothing becomes an explicit “No matching records found.” envelope instead.
• register_schema_tools_to_llm also installs a catch-all handler for unregistered tool names that returns the list of tools available to the model.

Register with register_schema_tools_to_llm(llm, context, tools, cancel_on_interruption=True, keep_existing_tools=True, register_unknown_tool_handler=True). This is the path the evaluation domains use — refer to Authoring Evaluation Tools.

Interruption and Long-Running Tools

cancel_on_interruption defaults to True on both registration helpers, and that is what you want for an ordinary tool. Setting it to False does more than survive a barge-in: in Pipecat 1.x it also marks the tool asynchronous, so the LLM does not wait for the result. The model immediately receives a {"status": "running"} placeholder and the real payload is injected later as a developer message. Only opt in for a genuinely fire-and-forget tool. For bounded waits, use llm.function_call_timeout_secs (default 10.0). Refer to Tool Calling.

Testing Your Tool

Both paths are testable without a GPU or a running LLM: build a duck-typed stand-in for FunctionCallParams that records result_callback calls, then await the tool. tests/unit/test_runtime_basic_weather_tool.py does this for the weather tool. tests/unit/test_tool_call_contract.py does it for the schema-tool contract.

$
uv run pytest tests/unit/test_runtime_basic_weather_tool.py tests/unit/test_tool_call_contract.py -m "not gpu"

Two repository-wide requirements apply before committing. Every .py file other than __init__.py needs the SPDX/Apache header in its first 10 lines. CI fails without it. Ruff is the only formatter and linter. Run uv run ruff format my_tools.py and uv run ruff check --fix my_tools.py.

Next Steps

Continue with the runtime or reference guide that matches the tool surface you are implementing:

• Tool Calling — backend support, parser flags, and the shipped demo tools.
• Authoring Evaluation Tools — the schema-tool path in depth.
• Prompts — tuning system_prompt_suffix so the model reaches for tools at the right time.
• Server Configuration — where enable_tool_calling lives and how layering works.