How Custom Tools Are Registered and Executed in Dograh’s Voice Pipeline

Dograh’s voice pipeline enables dynamic tool integration by using the CustomToolManager class to fetch tool definitions from the database, convert them into LLM-compatible function schemas, and register asynchronous handlers that execute HTTP requests, end calls, or perform calculations when the LLM invokes them.

The open-source Dograh platform (dograh-hq/dograh) powers voice AI workflows through its PipecatEngine, which orchestrates real-time conversations and integrates external capabilities via custom tools. Understanding how custom tools are registered and executed in the voice pipeline is essential for building voice agents that dynamically call external APIs or control call flow based on LLM decisions.

Tool Discovery and Schema Generation

When a workflow node is activated, the engine receives a list of tool UUIDs from node.data.tool_uuids. The CustomToolManager fetches the corresponding ToolModel records from the database and transforms them into LLM-compatible schemas.

In api/services/workflow/pipecat_engine_custom_tools.py, the get_tool_schemas() method converts each database tool into a FunctionSchema using tool_to_function_schema:

schemas = await custom_tool_manager.get_tool_schemas(tool_uuids)
await self.llm.update_function_schemas(schemas)

This step ensures the LLM knows the function names, parameters, and descriptions required for valid function calling before any handlers are invoked.

Handler Registration Process

Once schemas are exposed to the LLM, CustomToolManager.register_handlers() binds executable Python functions to each tool. This process iterates over the tool UUIDs, creates concrete async handlers via _create_handler(), and registers them with the LLM runtime alongside timeout configurations.

From pipecat_engine_custom_tools.py:

# Inside PipecatEngine when a node is entered

tool_uuids = node.data.tool_uuids or []          # e.g. ["c1a2…", "d3e4…"]

await self._custom_tool_manager.register_handlers(tool_uuids)

The registration binds the handler to the LLM with timeout protection:

handler, timeout_secs = self._create_handler(tool, function_name)
self._engine.llm.register_function(
    function_name, handler, timeout_secs=timeout_secs
)

The manager supports four distinct handler types:

• Calculator tools: Registered once via _register_calculator_handler() using the built-in safe calculator implementation.
• HTTP-API tools: Created by _create_http_tool_handler() for external REST API calls.
• End-call tools: Created by _create_end_call_handler() to terminate conversations gracefully.
• Transfer-call tools: Created by _create_transfer_call_handler() to route calls to external numbers.

Executing HTTP-API Tools

When the LLM decides to invoke an HTTP tool, the registered handler executes the full request lifecycle. The handler manages pre-tool audio messages, resolves preset parameters, performs authentication, and dispatches the HTTP request.

The handler implementation in api/services/workflow/tools/custom_tool.py follows this pattern:

async def http_tool_handler(function_call_params: FunctionCallParams) -> None:
    # (pre‑tool custom message)

    await play_custom_message_if_configured(tool)

    # Execute the HTTP request

    result = await execute_http_tool(
        tool=tool,
        arguments=function_call_params.arguments,
        call_context_vars=self._engine._call_context_vars,
        gathered_context_vars=self._engine._gathered_context,
        organization_id=await self.get_organization_id(),
    )

    # Return the result to the LLM

    await function_call_params.result_callback(result)

The execute_http_tool function handles credential injection via build_auth_header, resolves preset parameters through _resolve_preset_parameters, and sends the request using httpx.AsyncClient.

Handling End-Call Actions

End-call tools allow the LLM to terminate conversations while optionally recording disposition reasons and playing goodbye messages. The handler registered by _create_end_call_handler() manages this cleanup process.

From the source in pipecat_engine_custom_tools.py:

async def end_call_handler(function_call_params: FunctionCallParams) -> None:
    config = tool.definition.get("config", {})
    # Optionally record a reason

    if config.get("endCallReason"):
        reason = function_call_params.arguments.get("reason", "end_call_tool")
        self._engine._gathered_context["call_disposition"] = reason

    # Tell the LLM the tool succeeded

    await function_call_params.result_callback(
        {"status": "success", "action": "ending_call"},
        properties=FunctionCallResultProperties(run_llm=False),
    )

    # Play goodbye message (if any) then terminate the call

    if await self._play_config_message(config):
        await self._engine.end_call_with_reason(
            EndTaskReason.END_CALL_TOOL_REASON.value, abort_immediately=False
        )
    else:
        await self._engine.end_call_with_reason(
            EndTaskReason.END_CALL_TOOL_REASON.value, abort_immediately=True
        )

The run_llm=False property prevents the LLM from generating additional text after the termination signal, while abort_immediately controls whether to wait for audio playback completion.

The Calculator Tool

Dograh includes a built-in calculator tool for safe arithmetic operations that executes within the pipeline without external HTTP calls. This tool is registered once during initialization via _register_calculator_handler() and implemented in api/services/workflow/tools/calculator.py as safe_calculator.

Key Source Files and Architecture

The custom tool system spans several modules that work together to enable dynamic function calling:

• api/services/workflow/pipecat_engine_custom_tools.py: Contains the CustomToolManager class that orchestrates schema generation, handler creation, and LLM registration.
• api/services/workflow/tools/custom_tool.py: Provides tool_to_function_schema for schema conversion, _resolve_preset_parameters for argument resolution, and execute_http_tool for HTTP dispatch.
• api/services/workflow/pipecat_engine.py: The main engine that instantiates CustomToolManager and invokes registration methods when workflow nodes activate.
• api/services/workflow/tools/calculator.py: Implements the safe_calculator function for built-in arithmetic operations.

Summary

• Tool discovery begins when the PipecatEngine reads tool_uuids from workflow node data.
• Schema generation converts database ToolModel records into LLM-compatible FunctionSchema objects via CustomToolManager.get_tool_schemas().
• Handler registration binds async functions to the LLM using register_handlers(), supporting HTTP APIs, end-call actions, and call transfers with configurable timeouts.
• Execution flow handles pre-tool messages, HTTP requests with credential injection via build_auth_header, and result callbacks through function_call_params.result_callback.
• Termination control allows end-call tools to record disposition reasons, play goodbye messages, and abort immediately or gracefully using EndTaskReason.END_CALL_TOOL_REASON.

Frequently Asked Questions

How does Dograh convert database tool definitions into LLM function schemas?

The CustomToolManager.get_tool_schemas() method in pipecat_engine_custom_tools.py fetches ToolModel records from the database and transforms them using tool_to_function_schema from custom_tool.py. This produces FunctionSchema objects that define the function name, description, and parameter specifications required by the LLM for valid function calling.

What happens when the LLM decides to call a custom tool during a conversation?

When the LLM emits a function call, the Pipecat runtime invokes the pre-registered async handler associated with that tool name. For HTTP tools, this triggers execute_http_tool() which resolves arguments using _resolve_preset_parameters, injects authentication headers, makes the request via httpx.AsyncClient, and returns the result through function_call_params.result_callback().

Can custom tools play audio messages before executing their primary action?

Yes. Handlers can trigger pre-tool messages using play_custom_message_if_configured(). For end-call tools specifically, the _play_config_message() method plays configured goodbye audio before terminating the call, allowing the abort_immediately parameter to control whether to wait for playback completion or end the call instantly.

Where are tool handlers registered in the voice pipeline lifecycle?

Handlers are registered when the PipecatEngine enters a workflow node containing tools. The engine calls await self._custom_tool_manager.register_handlers(tool_uuids) with the UUIDs specified in node.data.tool_uuids, binding each tool to the LLM instance via self._engine.llm.register_function() with appropriate timeout values.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}