nemo_voice_agent.utils.tool_calling | NeMo Labs Voice Agent

Submodules

Package Contents

Classes

Name	Description
`StandardSchemaTool`	Base class for all standard tools with FunctionSchema.

Functions

Name	Description
`get_standard_schema_tool`	Get a schema tool for evaluation by name.
`list_standard_schema_tools`	List all schema tools for evaluation.
`register_schema_tools_to_llm`	Register standard schema tools to the LLM.
`register_standard_schema_tool`	Class decorator that registers a tool class into ALL_STANDARD_SCHEMA_TOOLS.

Data

ALL_STANDARD_SCHEMA_TOOLS

API

class nemo_voice_agent.utils.tool_calling.base.StandardSchemaTool(
    description: typing.Optional[str] = None,
    name: typing.Optional[str] = None
)

Base class for all standard tools with FunctionSchema.

description

= description if description is not None else ''

properties

Dict[str, Any]

Return the properties for the tool.

An example of the properties:

properties = {
    "location": {
        "type": "string",
        "description": "The city and state, e.g. San Francisco, CA",
    },
    "format": {
            "type": "string",
            "enum": ["celsius", "fahrenheit"],
            "description": "The temperature unit to use.",
    },
}

required_properties

List[str]

Return the required properties for the tool.

An example of the required properties:

required_properties = ["location", "format"]

schema

FunctionSchema

Return the FunctionSchema for the tool. Refer to https://docs.pipecat.ai/guides/learn/function-calling#using-the-standard-schema-recommended for more details.

An example of the FunctionSchema:

schema = FunctionSchema(
    name="get_current_weather",
    description="Get the current weather in a location",
    properties={
        "location": {
            "type": "string",
            "description": "The city and state, e.g. San Francisco, CA",
        },
        "format": {
            "type": "string",
            "enum": ["celsius", "fahrenheit"],
            "description": "The temperature unit to use.",
        },
    },
    required=["location", "format"]
)

nemo_voice_agent.utils.tool_calling.base.StandardSchemaTool.__call__(
    params: pipecat.services.llm_service.FunctionCallParams
) -> None

async

Pipecat entry point. Owns everything framework-facing.

This is the only place a tool result is delivered. Subclasses implement _execute, which takes the call arguments as plain keyword arguments and returns a plain result — it never touches params and never delivers.

Keeping delivery here makes double-delivery structurally impossible. Previously _execute received params and each implementation called params.result_callback itself, while this method also delivered _execute’s return value — so every call sent its real result and then a second, spurious None for the same tool_call_id. Pipecat <1.0 ignored the duplicate; 1.x tracks in-flight tool calls and rejects it with “tool_call_id … is not running”, which strands the aggregator’s deferred context push and stops the LLM being re-invoked with the tool output.

nemo_voice_agent.utils.tool_calling.base.StandardSchemaTool._after_result(
    params: pipecat.services.llm_service.FunctionCallParams
) -> None

async

Hook for side effects that must run after the result is delivered.

Default is a no-op. Used by the end-conversation tools, which emit an exit message once the LLM has been given the tool result.

nemo_voice_agent.utils.tool_calling.base.StandardSchemaTool._execute(
    kwargs: typing.Any = {}
) -> typing.Any

async

Run the tool and RETURN its result. Pure logic, no framework types.

Receives the LLM-supplied call arguments as keyword arguments and returns the result to send back. Do not import or touch FunctionCallParams, and do not deliver the result yourself — __call__ owns delivery and calls it exactly once.

Raising is fine: __call__ converts an exception into a structured {"error": ...} result so the LLM gets a usable signal and the aggregator is never left waiting.

An example of a get_current_weather tool:

async def _execute(self, location: str, format: str = "celsius") -> dict:
    return {"location": location, "format": format}

nemo_voice_agent.utils.tool_calling.get_standard_schema_tool(
    name: str,
    kwargs = {}
) -> nemo_voice_agent.utils.tool_calling.base.StandardSchemaTool

Get a schema tool for evaluation by name.

nemo_voice_agent.utils.tool_calling.list_standard_schema_tools() -> typing.List[nemo_voice_agent.utils.tool_calling.base.StandardSchemaTool]

List all schema tools for evaluation.

nemo_voice_agent.utils.tool_calling.base.register_schema_tools_to_llm(
    llm: pipecat.services.openai.llm.OpenAILLMService,
    context: pipecat.processors.aggregators.llm_context.LLMContext,
    tools: typing.List[nemo_voice_agent.utils.tool_calling.base.StandardSchemaTool],
    cancel_on_interruption: bool = True,
    keep_existing_tools: bool = True,
    register_unknown_tool_handler: bool = True
) -> None

Register standard schema tools to the LLM. Args: llm: The LLM service to use. context: The LLM context to use. tools: The list of tools to register. cancel_on_interruption: Whether to cancel an in-flight tool call when the user interrupts. Keep the True default for ordinary synchronous tools. In pipecat >=1.0 this flag does double duty: False also marks the tool asynchronous, meaning the LLM does not wait for the result — it continues the conversation immediately, the tool message it receives is a {"status": "running"} placeholder, and the real payload is injected later as a developer message. Only pass False for a genuinely long-running, fire-and-forget tool whose result the model must not block on. (In pipecat 0.x the flag meant only “survive interruption”; there is no longer a way to get that behaviour without also opting into the async protocol.) keep_existing_tools: Whether to keep the existing tools in the context. register_unknown_tool_handler: When True (default), registers a catch-all handler under function_name=None so any hallucinated tool call gets a structured error result instead of wedging the aggregator. See _unknown_tool_handler docstring for the deadlock chain this avoids. Disable only if you’ve already registered a custom catch-all via llm.register_function(function_name=None, ...).

nemo_voice_agent.utils.tool_calling.register_standard_schema_tool(
    cls
)

Class decorator that registers a tool class into ALL_STANDARD_SCHEMA_TOOLS.

Usage: @register_standard_schema_tool class MyTool: name = “my_tool” …

The tool is keyed by cls.name if it exists, otherwise cls.name.

nemo_voice_agent.utils.tool_calling.ALL_STANDARD_SCHEMA_TOOLS = {}