Function Calling

Function calling (also known as tool calling) allows LLMs to request information from external services and APIs during conversations. This extends your voice AI bot’s capabilities beyond its training data to access real-time information and perform actions.

Pipeline Integration

Function calling works seamlessly within your existing pipeline structure. The LLM service handles function calls automatically when they’re needed:

pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),     # Collects user transcriptions
    llm,                          # Processes context, calls functions when needed
    tts,
    transport.output(),
    context_aggregator.assistant(), # Collects function results and responses
])

Function call flow:

User asks a question requiring external data
LLM recognizes the need and calls appropriate function
Your function handler executes and returns results
LLM incorporates results into its response
Response flows to TTS and user as normal

Context integration: Function calls and their results are automatically stored in conversation context by the context aggregators, maintaining complete conversation history.

Understanding Function Calling

Function calling allows your bot to access real-time data and perform actions that aren’t part of its training data. For example, you could give your bot the ability to:

Check current weather conditions
Look up stock prices
Query a database
Control smart home devices
Schedule appointments

Here’s how it works:

You define functions the LLM can use and make them available to the LLM service used in your pipeline
When needed, the LLM requests a function call
Your application executes any corresponding functions
The result is sent back to the LLM
The LLM uses this information in its response

Implementation

1. Define a tool

A tool needs two things: a handler — the code to run when the LLM calls the tool — and a schema that describes the tool to the LLM (its name, what it does, and its parameters) so the model knows it exists and how to call it. The preferred way to define a tool is with a direct function: a single async function that is both the handler and the schema. Pipecat auto-derives the tool’s metadata — name, description, parameter properties (with their descriptions), and which parameters are required — from the function’s signature and docstring. The first parameter is always params (a FunctionCallParams); the tool’s own arguments follow. Document each argument in a Google-style docstring.

from pipecat.services.llm_service import FunctionCallParams

async def get_current_weather(params: FunctionCallParams, location: str, format: str):
    """Get the current weather.

    Args:
        location: The city and state, e.g. "San Francisco, CA".
        format: The temperature unit to use. Must be either "celsius" or "fahrenheit". Infer this from the user's location.
    """
    weather_data = {"conditions": "sunny", "temperature": "75"}
    await params.result_callback(weather_data)

The direct-function schema generator doesn’t yet map Literal types to a JSON-schema enum. Express enum-like constraints in the docstring prose instead (e.g. ‘Must be either “celsius” or “fahrenheit”’), as shown above. If you need a strict enum in the schema, use the verbose FunctionSchema pattern.

2. Add the tool to the context

List your direct functions in LLMContext(tools=[...]):

from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair

context = LLMContext(tools=[get_current_weather, get_restaurant_recommendation])
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)

The bot’s personality (e.g. “You are a helpful assistant”) is set via system_instruction in the LLM service’s Settings, not as a context message. Tools are automatically converted to the correct format for your LLM provider through adapters.

3. Create the pipeline

Include your LLM service in the pipeline:

# Create the pipeline
pipeline = Pipeline([
    transport.input(),     # Input from the transport
    stt,                   # STT processing
    user_aggregator,       # User context aggregation
    llm,                   # LLM processing
    tts,                   # TTS processing
    transport.output(),    # Output to the transport
    assistant_aggregator,  # Assistant context aggregation
])

Per-Tool Options with `@tool_options`

By default, a direct function is cancelled if the user interrupts, and it uses the LLM service’s global timeout. To override either, decorate the function with @tool_options. The decorator only attaches call options — the schema is still auto-derived — so decorated functions can stay at module level.

import asyncio

from pipecat.adapters.schemas.direct_function import tool_options
from pipecat.services.llm_service import FunctionCallParams

@tool_options(cancel_on_interruption=False, timeout_secs=30)
async def get_current_weather(params: FunctionCallParams, location: str, format: str):
    """Get the current weather.

    Args:
        location: The city and state, e.g. "San Francisco, CA".
        format: The temperature unit to use. Must be either "celsius" or "fahrenheit". Infer this from the user's location.
    """
    # Simulate a long-running API call.
    await asyncio.sleep(20)
    await params.result_callback({"conditions": "nice", "temperature": "75"})

Options:

cancel_on_interruption (default True): When True, the call is cancelled if the user interrupts. When False, the call is treated as asynchronous — see below.
timeout_secs (default None): Per-tool timeout in seconds. Overrides the global function_call_timeout_secs for this function. Use a longer timeout for slow operations (e.g. database queries) or a shorter one for quick lookups.

@tool_options also sets call options on the handler of a FunctionSchema tool, not just a direct function.

On an LLMWorker, mark tool methods with @tool instead. It applies the same options and marks the method for automatic collection as one of the worker’s own tools.

Synchronous vs. asynchronous calls

With cancel_on_interruption=True (the default), the call is synchronous: the LLM waits for the result before generating its next response. This ensures the LLM has complete information before responding. With cancel_on_interruption=False, the call is asynchronous: the LLM continues the conversation immediately without waiting. Once the result returns, it’s injected back into the context as a developer message, triggering a new LLM inference at that point. This enables truly non-blocking calls where the conversation proceeds while the function runs in the background. Async calls can also send intermediate updates before their final result.

Async function call cancellation

For async functions (cancel_on_interruption=False), you can also enable model-directed cancellation:

llm = OpenAILLMService(
    api_key="your-api-key",
    enable_async_tool_cancellation=True,
)

When enable_async_tool_cancellation=True and at least one async function is available, Pipecat automatically adds the built-in cancel_async_tool_call tool and supporting system instructions. The LLM can call that tool to cancel a stale in-progress async function call — for example, when the user changes their request before a long-running lookup completes.

Changing Tools Mid-Conversation

To change the set of tools the LLM can use during a session, push an LLMSetToolsFrame. Its tools field takes the same things as LLMContext(tools=[...]) — a list of direct functions and/or FunctionSchema objects. Whatever you pass becomes the LLM’s new tool set.

from pipecat.frames.frames import LLMSetToolsFrame
from pipecat.processors.aggregators.llm_context import NOT_GIVEN

# Make get_current_weather the only tool the LLM can call
await worker.queue_frame(LLMSetToolsFrame(tools=[get_current_weather]))

# Clear all tools
await worker.queue_frame(LLMSetToolsFrame(tools=NOT_GIVEN))

Tools Across Service Switches

When you use an LLMSwitcher to swap LLM providers mid-session, the tools you list in LLMContext(tools=[...]) are available on whichever provider is active. You define them once for the whole switcher.

from pipecat.pipeline.llm_switcher import LLMSwitcher

llm_switcher = LLMSwitcher(llms=[llm_openai, llm_google])

# Tools in the context are available on whichever provider is active
context = LLMContext(tools=[get_current_weather, get_restaurant_recommendation])
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)

Advanced: Defining Tools with `FunctionSchema`

Direct functions cover most cases. Reach for the verbose FunctionSchema pattern when you need explicit control over the schema — for example, a strict enum constraint (which the direct-function generator doesn’t yet emit) — or when the tool’s handler isn’t shaped like a direct function. A FunctionSchema spells out the tool’s name, description, and parameters by hand. Pass the handler that runs when the LLM calls the tool as the schema’s handler, then list the schema in LLMContext(tools=[...]) — exactly as you would a direct function.

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.services.llm_service import FunctionCallParams

async def fetch_weather_from_api(params: FunctionCallParams):
    weather_data = {"conditions": "sunny", "temperature": "75"}
    await params.result_callback(weather_data)

weather_function = 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",
            # A strict enum — the kind of explicit control a direct function can't yet express.
            "enum": ["celsius", "fahrenheit"],
            "description": "The temperature unit to use.",
        },
    },
    required=["location", "format"],
    handler=fetch_weather_from_api,  # bundle the handler on the schema
)

# List the schema in the context, just like a direct function.
context = LLMContext(tools=[weather_function])

To override the handler’s default call options, decorate it with @tool_options — the same decorator direct functions use, with the same synchronous vs. asynchronous semantics:

from pipecat.adapters.schemas.direct_function import tool_options

@tool_options(cancel_on_interruption=False, timeout_secs=30)
async def fetch_weather_from_api(params: FunctionCallParams):
    ...

These schemas behave just like direct functions everywhere else in this guide — swap them mid-conversation with an LLMSetToolsFrame, and they keep working across an LLMSwitcher’s providers.

Registering a handler manually

Bundling the handler on the schema (above) is the recommended approach. If you’d rather keep the handler separate, list a handler-free FunctionSchema in the context as usual and register its handler by name:

# weather_function here is the same schema, just defined without handler=.
context = LLMContext(tools=[weather_function])
llm.register_function("get_current_weather", fetch_weather_from_api)

To remove the tool, un-advertise it with an LLMSetToolsFrame; call llm.unregister_function(...) only afterward, since unregistering a still-advertised tool leaves the LLM able to call a handler that’s no longer there. This is uncommon — bundling keeps a tool and its handler together — but the option is there when you need to manage registration directly.

Provider-Specific Custom Tools

For normal function calling, prefer standard_tools with FunctionSchema or direct functions so Pipecat can convert them to each provider’s native format. When a provider has tools that don’t fit Pipecat’s standard function schema, add those provider-native definitions through ToolsSchema.custom_tools. These custom tools are passed only to the matching adapter and are appended to the converted standard tools.

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema

# Standard function converted by Pipecat
weather_function = FunctionSchema(
    name="get_current_weather",
    description="Get the current weather",
    properties={"location": {"type": "string"}},
    required=["location"],
)

# Provider-native tool appended only for OpenAI-family adapters.
# This object must match the target OpenAI API you are using.
provider_tool = {"type": "tool_search"}

tools = ToolsSchema(
    standard_tools=[weather_function],
    custom_tools={AdapterType.OPENAI: [provider_tool]},
)

Raw provider-native tool lists are not the normal LLMContext path. Some lower-level adapter code still preserves non-ToolsSchema tools for legacy or direct provider-specific paths, but LLMContext(tools=...) validates tools as a ToolsSchema. Use custom_tools as the provider-specific escape hatch while staying in the universal context flow.

For normal callable functions, use direct functions or FunctionSchema instead of provider-native function definitions. Today, custom_tools is supported for OpenAI-family adapters and Gemini. Anthropic standard functions should be represented with FunctionSchema.

Function Handler Details

FunctionCallParams

Every function handler receives a FunctionCallParams object containing all the information needed for execution:

@dataclass
class FunctionCallParams:
    function_name: str                          # Name of the called function
    tool_call_id: str                           # Unique identifier for this call
    arguments: Mapping[str, Any]                # Arguments from the LLM
    llm: LLMService                             # Reference to the LLM service
    context: LLMContext                         # Current conversation context
    result_callback: FunctionCallResultCallback # Return results here
    app_resources: Any                          # Application-defined resources shared across tool calls

Using the parameters:

async def example_function_handler(params: FunctionCallParams):
    # Access function details
    print(f"Called function: {params.function_name}")
    print(f"Call ID: {params.tool_call_id}")

    # Extract arguments
    location = params.arguments["location"]

    # Access LLM context for conversation history
    messages = params.context.messages

    # Access shared resources (database, API clients, etc.)
    if params.app_resources:
        db = params.app_resources.database
        user_id = params.app_resources.current_user_id

    # Use LLM service for additional operations
    await params.llm.push_frame(TTSSpeakFrame("Looking up weather data..."))

    # Return results
    await params.result_callback({"conditions": "nice", "temperature": "75"})

See the API reference for complete details.

params.tool_resources is a deprecated alias for params.app_resources. Use app_resources in new code.

Handler Structure

Your function handler should:

Receive necessary arguments, either:
- From params.arguments
- Directly from function arguments, if using direct functions
Process data or call external services
Return results via params.result_callback(result)

async def fetch_weather_from_api(params: FunctionCallParams):
    try:
        # Extract arguments
        location = params.arguments.get("location")
        format_type = params.arguments.get("format", "celsius")

        # Call external API
        api_result = await weather_api.get_weather(location, format_type)

        # Return formatted result
        await params.result_callback({
            "location": location,
            "temperature": api_result["temp"],
            "conditions": api_result["conditions"],
            "unit": format_type
        })
    except Exception as e:
        # Handle errors
        await params.result_callback({
            "error": f"Failed to get weather: {str(e)}"
        })

When function handlers need access to shared resources like database connections, API clients, or application state, you can pass them via app_resources when creating the PipelineWorker. These resources are then accessible in every function handler via params.app_resources.

from dataclasses import dataclass
from pipecat.pipeline.worker import PipelineWorker
from pipecat.services.llm_service import FunctionCallParams

# Define your application resources
@dataclass
class AppResources:
    database: DatabaseConnection
    api_client: WeatherAPIClient
    user_id: str

# Create your resources
resources = AppResources(
    database=db_connection,
    api_client=weather_client,
    user_id="user-123"
)

# Pass resources to the pipeline worker
worker = PipelineWorker(
    pipeline,
    app_resources=resources
)

# Access resources in function handlers
async def query_user_preferences(params: FunctionCallParams):
    # Access shared resources
    db = params.app_resources.database
    user_id = params.app_resources.user_id

    # Query database with shared connection
    prefs = await db.query("SELECT * FROM preferences WHERE user_id = ?", user_id)

    await params.result_callback(prefs)

Key points:

Resources are passed by reference — the caller retains their handle and can read mutations after the task finishes
The framework never copies or clears the app_resources object
All function handlers in the pipeline share the same app_resources instance
Useful for database connections, API clients, caches, or any shared state

PipelineWorker(tool_resources=...) and FunctionCallParams.tool_resources are deprecated aliases retained for compatibility. Prefer PipelineWorker(app_resources=...) and params.app_resources.

Advanced: Controlling Function Call Behavior

When returning results from a function handler, you can control how the LLM processes those results using a FunctionCallResultProperties object passed to the result callback.

Properties

FunctionCallResultProperties provides fine-grained control over LLM execution:

@dataclass
class FunctionCallResultProperties:
    run_llm: bool | None = None                 # Whether to run LLM after this result
    on_context_updated: Callable | None = None  # Callback when context is updated
    is_final: bool = True                       # Whether this is the final result

Property options:

run_llm=True: Run LLM after function call (default behavior)
run_llm=False: Don’t run LLM after function call (useful for chained calls)
on_context_updated: Async callback executed after the function result is added to context
is_final=False: Treat this as an intermediate result for an async function call. Only use this for async functions (cancel_on_interruption=False)

Skip LLM execution (run_llm=False) when you have back-to-back function calls. If you skip a completion, you must manually trigger one from the context aggregator.

See the API reference for complete details.

Example Usage

from pipecat.frames.frames import FunctionCallResultProperties
from pipecat.services.llm_service import FunctionCallParams

async def fetch_weather_from_api(params: FunctionCallParams):
    # Fetch weather data
    weather_data = {"conditions": "sunny", "temperature": "75"}

    # Don't run LLM after this function call
    properties = FunctionCallResultProperties(run_llm=False)

    await params.result_callback(weather_data, properties=properties)

async def query_database(params: FunctionCallParams):
    # Query database
    results = await db.query(params.arguments["query"])

    async def on_update():
        await notify_system("Database query complete")

    # Run LLM after function call and notify when context is updated
    properties = FunctionCallResultProperties(
        run_llm=True,
        on_context_updated=on_update
    )

    await params.result_callback(results, properties=properties)

Intermediate Results for Async Functions

Async function calls can send progress updates before their final result. Make the function async with @tool_options(cancel_on_interruption=False), then call params.result_callback(..., properties=FunctionCallResultProperties(is_final=False)) for each intermediate update. Finish with a normal params.result_callback(...).

from pipecat.adapters.schemas.direct_function import tool_options
from pipecat.frames.frames import FunctionCallResultProperties
from pipecat.services.llm_service import FunctionCallParams

@tool_options(cancel_on_interruption=False)
async def track_delivery(params: FunctionCallParams):
    """Track a delivery, reporting each status update until it arrives."""
    await params.result_callback(
        {"status": "picked_up"},
        properties=FunctionCallResultProperties(is_final=False),
    )

    await params.result_callback(
        {"status": "nearby"},
        properties=FunctionCallResultProperties(is_final=False),
    )

    await params.result_callback({"status": "delivered"})

Intermediate results are injected into the LLM context as async-tool developer messages. They do not close the function call; the call remains in progress until the final result is sent.

Key Takeaways

Function calling extends LLM capabilities beyond training data to real-time information
Context integration is automatic - function calls and results are stored in conversation history
Direct functions are the preferred approach - one async function is both schema and handler; list it in LLMContext(tools=[...]) or add it via LLMSetToolsFrame to make it available. When you need explicit schema control, use a FunctionSchema with its handler bundled in
Async function calls are opt-in - set cancel_on_interruption=False for deferred results, intermediate updates, and optional async-tool cancellation
Pipeline integration is seamless - functions work within your existing voice AI architecture
Advanced control available - fine-tune LLM execution and monitor function call lifecycle

What’s Next

Now that you understand function calling, let’s explore how to configure text-to-speech services to convert your LLM’s responses (including function call results) into natural-sounding speech.

Text to Speech

Learn how to configure speech synthesis in your voice AI pipeline

​Pipeline Integration

​Understanding Function Calling

​Implementation

​1. Define a tool

​2. Add the tool to the context

​3. Create the pipeline

​Per-Tool Options with @tool_options

​Synchronous vs. asynchronous calls

​Async function call cancellation

​Changing Tools Mid-Conversation

​Tools Across Service Switches

​Advanced: Defining Tools with FunctionSchema

​Registering a handler manually

​Provider-Specific Custom Tools

​Function Handler Details

​FunctionCallParams

​Handler Structure

​Sharing Resources with app_resources

​Advanced: Controlling Function Call Behavior

​Properties

​Example Usage

​Intermediate Results for Async Functions

​Key Takeaways

​What’s Next

Text to Speech

Pipeline Integration

Understanding Function Calling

Implementation

1. Define a tool

2. Add the tool to the context

3. Create the pipeline

Per-Tool Options with `@tool_options`

Synchronous vs. asynchronous calls

Async function call cancellation

Changing Tools Mid-Conversation

Tools Across Service Switches

Advanced: Defining Tools with `FunctionSchema`

Registering a handler manually

Provider-Specific Custom Tools

Function Handler Details

FunctionCallParams

Handler Structure

Sharing Resources with app_resources

Advanced: Controlling Function Call Behavior

Properties

Example Usage

Intermediate Results for Async Functions

Key Takeaways

What’s Next