Functions

Functions in Pipecat Flows serve two key purposes:

Process data by interfacing with external systems and APIs to read or write information
Progress the conversation by transitioning between nodes in your flow

How Functions Work

When designing your nodes, clearly define the task in the task_messages and reference the available functions. The LLM will use these functions to complete the task and signal when it’s ready to move forward. For example, if your node’s job is to collect a user’s favorite color:

The LLM asks the question
The user provides their answer
The LLM calls the function with the answer
The function processes the data and determines the next node

Function Definition

Flows provides a universal FlowsFunctionSchema that works across all LLM providers:

from pipecat_flows import FlowsFunctionSchema

record_favorite_color_func = FlowsFunctionSchema(
    name="record_favorite_color_func",
    description="Record the color the user said is their favorite.",
    required=["color"],
    handler=record_favorite_color_and_set_next_node,
    properties={"color": {"type": "string"}},
)

Function Handlers

Each function has a corresponding handler where you implement your application logic and specify the next node:

async def record_favorite_color_and_set_next_node(
    args: FlowArgs, flow_manager: FlowManager
) -> tuple[str, NodeConfig]:
    """Function handler that records the color then sets the next node.

    Here "record" means print to the console, but any logic could go here:
    Write to a database, make an API call, etc.
    """
    print(f"Your favorite color is: {args['color']}")
    return args["color"], create_end_node()

Handler Return Values

Function handlers can return any JSON-serializable value (string, dict, list, etc.) that gets passed to the LLM. To also specify the next node, return a tuple:

Result: Data provided to the LLM for context in subsequent completions, or None. Any JSON-serializable value is accepted.
Next Node: The NodeConfig for Flows to transition to next, or None

Some handlers may not want to transition conversational state, in which case you can return None for the next node. Other handlers may only want to transition conversational state without doing other work, in which case you can return None for the result.

Direct Functions

For more concise code, you can optionally use Direct Functions where the function definition and handler are combined in a single function. The function signature and docstring are automatically used to generate the function schema:

async def record_favorite_color(
    flow_manager: FlowManager,
    color: str
) -> tuple[str, NodeConfig]:
    """Record the color the user said is their favorite.

    Args:
        color: The user's favorite color.
    """
    print(f"Your favorite color is: {color}")
    return color, create_end_node()

# Use directly in NodeConfig
node_config = {
    "functions": [record_favorite_color]
}

This approach eliminates the need for separate FlowsFunctionSchema definitions while maintaining the same functionality. To control interruption behavior, use the @flows_direct_function decorator:

from pipecat_flows import flows_direct_function

@flows_direct_function(cancel_on_interruption=False)
async def long_running_lookup(
    flow_manager: FlowManager,
    order_id: str
) -> tuple[dict, NodeConfig]:
    """Look up an order that should not be cancelled if the user speaks.

    Args:
        order_id: The order ID to look up.
    """
    order = await db.get_order(order_id)
    return {"status": "success", "order": order}, create_order_node(order)

Get Started

Guides

Examples

Migration

How Functions Work

Function Definition

Function Handlers

Handler Return Values

Direct Functions

​How Functions Work

​Function Definition

​Function Handlers

​Handler Return Values

​Direct Functions

How Functions Work

Function Definition

Function Handlers

Handler Return Values

Direct Functions