Skip to main content
Functions in Pipecat Flows serve two key purposes:
  1. Process data by interfacing with external systems and APIs to read or write information
  2. 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:
  1. The LLM asks the question
  2. The user provides their answer
  3. The LLM calls the function with the answer
  4. 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 return a tuple containing:
  • Result: Data provided to the LLM for context in subsequent completions, or None. This can be any serializable value — a string, dict, etc.
  • 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[FlowResult, 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[FlowResult, 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"}, create_order_node(order)