Pipecat Flows

New to building conversational flows? Check out our Pipecat Flows guide first.

Installation

pip install pipecat-ai-flows

Core Types

FlowArgs

Dict[str, Any]

Type alias for function handler arguments.

FlowResult

TypedDict

Base type for function handler results. Additional fields can be included as needed.

FlowConfig

TypedDict

Configuration for the entire conversation flow.

NodeConfig

TypedDict

Configuration for a single node in the flow.

role_messages

List[dict]

Defines the role or persona of the LLM. Required for the initial node and optional for subsequent nodes.

# Example role messages
"role_messages": [
    {
        "role": "system",
        "content": "You are a helpful assistant..."
    }
],

task_messages

List[dict]

required

Defines the task for a given node. Required for all nodes.

# Example task messages
"task_messages": [
    {
        "role": "system", # May be `user` depending on the LLM
        "content": "Ask the user for their name..."
    }
],

context_strategy

ContextStrategyConfig

Strategy for managing context during transitions to this node.

# Example context strategy configuration
"context_strategy": ContextStrategyConfig(
    strategy=ContextStrategy.RESET_WITH_SUMMARY,
    summary_prompt="Summarize the key points discussed so far."
)

functions

List[Union[dict, FlowsFunctionSchema]]

required

LLM function / tool calls, defined either as dictionaries in the format for the specific LLM provider, or as FlowsFunctionSchema objects.

In addition to the standard fields, the function call can also contain:

handler: a custom handler that executes when the function is called.
transition_to: for static flows, specifies the next node to transition to.
transition_callback: for dynamic flows, provides a handler to determine the next node.

Note: A function cannot have both transition_to and transition_callback.

# Using native format
"functions": [
    {
        "type": "function",
        "function": {
            "name": "get_current_movies",
            "handler": get_movies,  # Optional custom handler
            "description": "Fetch movies currently playing",
            "parameters": {...},
            "transition_to": "explore_movie"  # Static transition
        },
    }
]

# Using FlowsFunctionSchema
"functions": [
    FlowsFunctionSchema(
        name="get_current_movies",
        description="Fetch movies currently playing",
        properties={...},
        required=[...],
        handler=get_movies,
        transition_to="explore_movie"
    )
]

pre_actions

List[dict]

Actions that execute before the LLM inference. For example, you can send a message to the TTS to speak a phrase (e.g. “Hold on a moment…”), which may be effective if an LLM function call takes time to execute.

# Example pre_actions
"pre_actions": [
    {
        "type": "tts_say",
        "text": "Hold on a moment..."
    }
],

post_actions

List[dict]

Actions that execute after the LLM inference. For example, you can end the conversation.

# Example post_actions
"post_actions": [
    {
        "type": "end_conversation"
    }
]

respond_immediately

bool

If set to False, the LLM will not respond immediately when the node is set, but will instead wait for the user to speak first before responding.

Defaults to True.

# Example usage
"respond_immediately": False

Function Handler Types

LegacyFunctionHandler

Callable[[FlowArgs], Awaitable[FlowResult]]

Legacy function handler that only receives arguments.

FlowFunctionHandler

Callable[[FlowArgs, FlowManager], Awaitable[FlowResult]]

Modern function handler that receives both arguments and FlowManager.

ContextStrategy

Enum

Strategy for managing conversation context during node transitions.

ContextStrategyConfig

dataclass

Configuration for context management strategy.

# Example usage
config = ContextStrategyConfig(
    strategy=ContextStrategy.RESET_WITH_SUMMARY,
    summary_prompt="Summarize the key points discussed so far."
)

FlowsFunctionSchema

class

A standardized schema for defining functions in Pipecat Flows with flow-specific properties.

You cannot specify both transition_to and transition_callback in the same function schema.

Example usage:

from pipecat_flows import FlowsFunctionSchema

# Define a function schema
collect_name_schema = FlowsFunctionSchema(
    name="collect_name",
    description="Record the user's name",
    properties={
        "name": {
            "type": "string",
            "description": "The user's name"
        }
    },
    required=["name"],
    handler=collect_name_handler,
    transition_to="next_node"
)

# Use in node configuration
node_config = {
    "task_messages": [
        {"role": "system", "content": "Ask the user for their name."}
    ],
    "functions": [collect_name_schema]
}

# Pass to flow manager
await flow_manager.set_node("greeting", node_config)

FlowManager

class

Main class for managing conversation flows, supporting both static (configuration-driven) and dynamic (runtime-determined) flows.

Methods

initialize

method

Initialize the flow with starting messages.

set_node

method

Set up a new conversation node.

For dynamic flows, the application must advance the conversation using set_node to set up the next node.
For static flows, set_node is triggered by a function call that contains a transition_to field.

register_action

method

get_current_context

method

Get the current conversation context.

Returns a list of messages in the current context, including system messages, user messages, and assistant responses.

Example usage:

# Access current conversation context
context = flow_manager.get_current_context()

# Use in handlers
async def process_response(args: FlowArgs) -> FlowResult:
    context = flow_manager.get_current_context()
    # Process conversation history
    return {"status": "success"}

State Management

The FlowManager provides a state dictionary for storing conversation data:

flow_manager.state: Dict[str, Any]

# Store data
flow_manager.state["user_age"] = 25

Usage Examples

flow_config: FlowConfig = {
    "initial_node": "greeting",
    "nodes": {
        "greeting": {
            "role_messages": [
                {
                    "role": "system",
                    "content": "You are a helpful assistant. Your responses will be converted to audio."
                }
            ],
            "task_messages": [
                {
                    "role": "system",
                    "content": "Start by greeting the user and asking for their name."
                }
            ],
            "functions": [{
                "type": "function",
                "function": {
                    "name": "collect_name",
                    "handler": collect_name_handler,     # Specify handler
                    "description": "Record user's name",
                    "parameters": {...},
                    "transition_to": "next_step"         # Specify transition
                }
            }]
        }
    }
}

# Create and initialize the FlowManager
flow_manager = FlowManager(
    task=task,
    llm=llm,
    context_aggregator=context_aggregator,
    flow_config=flow_config
)

# Initialize the flow_manager to start the conversation
await flow_manager.initialize()

Node Functions

concept

Functions that execute operations within a state.

from pipecat_flows import FlowArgs, FlowResult

async def process_data(args: FlowArgs) -> FlowResult:
    """Handle data processing within a node."""
    data = args["data"]
    result = await process(data)
    return {
        "status": "success",
        "processed_data": result
    }

# Node configuration with transition
{
    "type": "function",
    "function": {
        "name": "process_data",
        "handler": process_data,
        "description": "Process user data",
        "parameters": {
            "type": "object",
            "properties": {
                "data": {"type": "string"}
            }
        },
    }
}

Edge Functions

concept

Functions that create transitions between nodes. Use transition_to (static flow) or transition_callback (dynamic flow) to specify the target node.

# Edge function configuration
{
    "type": "function",
    "function": {
        "name": "next_step",
        "description": "Transition to next node",
        "parameters": {"type": "object", "properties": {}},
        "transition_to": "target_node"  # Required: Specify target node (or use transition_callback for dynamic flows)
    }
}

Function Properties

handler

Optional[Callable]

Async function that processes data within a node. Can be specified as:

Direct function reference
Either a Callable function or a string with __function__: prefix (e.g., "__function__:process_data") to reference a function in the main script

{
    "type": "function",
    "function": {
        "name": "process_data",
        "handler": process_data,  # Callable function
        "parameters": {...}
    }
}

transition_callback

Optional[Callable]

Handler for dynamic flow transitions. Must be an async function with one of these signatures:

# New style (recommended)
async def handle_transition(
    args: Dict[str, Any],
    result: FlowResult,
    flow_manager: FlowManager
) -> None:
    """Handle transition to next node."""
    if result.available:  # Type-safe access to result
        await flow_manager.set_node("confirm", create_confirmation_node())
    else:
        await flow_manager.set_node(
            "no_availability",
            create_no_availability_node(result.alternative_times)
        )

# Legacy style (supported for backwards compatibility)
async def handle_transition(
    args: Dict[str, Any],
    flow_manager: FlowManager
) -> None:
    """Handle transition to next node."""
    await flow_manager.set_node("next", create_next_node())

The callback receives:

args: Arguments from the function call
result: Typed result from the function handler (new style only)
flow_manager: Reference to the FlowManager instance

Example usage:

async def handle_availability_check(
    args: Dict,
    result: TimeResult,  # Typed result
    flow_manager: FlowManager
):
    """Handle availability check and transition based on result."""
    if result.available:
        await flow_manager.set_node("confirm", create_confirmation_node())
    else:
        await flow_manager.set_node(
            "no_availability",
            create_no_availability_node(result.alternative_times)
        )

# Use in function configuration
{
    "type": "function",
    "function": {
        "name": "check_availability",
        "handler": check_availability,
        "parameters": {...},
        "transition_callback": handle_availability_check
    }
}

Note: A function cannot have both transition_to and transition_callback.

Handler Signatures

Function handlers can be defined with three different signatures:

async def handler_with_flow_manager(args: FlowArgs, flow_manager: FlowManager) -> FlowResult:
    """Modern handler that receives both arguments and FlowManager access."""
    # Access state
    previous_data = flow_manager.state.get("stored_data")

    # Access pipeline resources
    await flow_manager.task.queue_frame(TTSSpeakFrame("Processing your request..."))

    # Store data in state for later
    flow_manager.state["new_data"] = args["input"]

    return {
        "status": "success",
        "result": "Processed with flow access"
    }

The framework automatically detects which signature your handler is using and calls it appropriately.

Return Types

{
    "status": "success",
    "data": "some data"  # Optional additional data
}

Provider-Specific Formats

You don’t need to handle these format differences manually - use the standard format and the FlowManager will adapt it for your chosen provider.

{
    "type": "function",
    "function": {
        "name": "function_name",
        "handler": handler,
        "description": "Description",
        "parameters": {...}
    }
}

Actions

pre_actions and post_actions are used to manage conversation flow. They are included in the NodeConfig and executed before and after the LLM completion, respectively.

Three kinds of actions are available:

Pre-canned actions: These actions perform common tasks and require little configuration.
Function actions: These actions run developer-defined functions at the appropriate time.
Custom actions: These are fully developer-defined actions, providing flexibility at the expense of complexity.

Pre-canned Actions

Common actions shipped with Flows for managing conversation flow. To use them, just add them to your NodeConfig.

tts_say

action

Speaks text immediately using the TTS service.

"pre_actions": [
    {
        "type": "tts_say",
        "text": "Processing your request..."  # Required
    }
]

end_conversation

action

Ends the conversation and closes the connection.

"post_actions": [
    {
        "type": "end_conversation",
        "text": "Goodbye!"  # Optional farewell message
    }
]

Function Actions

Actions that run developer-defined functions at the appropriate time. For example, if used in post_actions, they’ll run after the bot has finished talking and after any previous post_actions have finished.

function

action

Calls the developer-defined function at the appropriate time.

"post_actions": [
    {
        "type": "function",
        "handler": bot_turn_ended # Required
    }
]

Custom Actions

Fully developer-defined actions, providing flexibility at the expense of complexity.

Here’s the complexity: because these actions aren’t queued in the Pipecat pipeline, they may execute seemingly early if used in post_actions; they’ll run immediately after the LLM completion is triggered but won’t wait around for the bot to finish talking.

Why would you want this behavior? You might be writing an action that:

Itself just queues another Frame into the Pipecat pipeline (meaning there would no benefit to waiting around for sequencing purposes)
Does work that can be done a bit sooner, like logging that the LLM was updated

Custom actions are composed of at least:

type

str

required

String identifier for the action

handler

Callable

required

Async or sync function that handles the action

Example:

# Define custom action handler
async def custom_notification(action: dict, flow_manager: FlowManager):
    """Custom action handler."""
    message = action.get("message", "")
    await notify_user(message)

# Use in node configuration
"pre_actions": [
    {
        "type": "notify",
        "handler": send_notification,
        "message": "Attention!",
    }
]

Exceptions

FlowError

exception

Base exception for all flow-related errors.

from pipecat_flows import FlowError

try:
    await flow_manager.set_node("invalid", config)
except FlowError as e:
    print(f"Flow error: {e}")

FlowInitializationError

exception

Raised when flow initialization fails.

from pipecat_flows import FlowInitializationError

try:
    await flow_manager.initialize()
except FlowInitializationError as e:
    print(f"Initialization failed: {e}")

FlowTransitionError

exception

Raised when a state transition fails.

from pipecat_flows import FlowTransitionError

try:
    await flow_manager.set_node("next", node_config)
except FlowTransitionError as e:
    print(f"Transition failed: {e}")

InvalidFunctionError

exception

Raised when an invalid or unavailable function is called.

from pipecat_flows import InvalidFunctionError

try:
await flow_manager.set_node("node", {
  "functions": [{
    "type": "function",
    "function": {
    "name": "invalid_function"
    }
  }]
})

except InvalidFunctionError as e:
    print(f"Invalid function: {e}")

API Reference

Services

Utilities

Frameworks

Pipeline

Base Service Classes

Installation

Core Types

FlowArgs

FlowResult

FlowConfig

NodeConfig

Function Handler Types

ContextStrategy

ContextStrategyConfig

FlowsFunctionSchema

FlowManager

Methods

State Management

Usage Examples

Function Properties

Handler Signatures

Return Types

Provider-Specific Formats

Actions

Pre-canned Actions

Function Actions

Custom Actions

Exceptions

API Reference

Services

Utilities

Frameworks

Pipeline

Base Service Classes

​Installation

​Core Types

​FlowArgs

​FlowResult

​FlowConfig

​NodeConfig

​Function Handler Types

​ContextStrategy

​ContextStrategyConfig

​FlowsFunctionSchema

​FlowManager

​Methods

​State Management

​Usage Examples

​Function Properties

​Handler Signatures

​Return Types

​Provider-Specific Formats

​Actions

​Pre-canned Actions

​Function Actions

​Custom Actions

​Exceptions

Installation

Core Types

FlowArgs

FlowResult

FlowConfig

NodeConfig

Function Handler Types

ContextStrategy

ContextStrategyConfig

FlowsFunctionSchema

FlowManager

Methods

State Management

Usage Examples

Function Properties

Handler Signatures

Return Types

Provider-Specific Formats

Actions

Pre-canned Actions

Function Actions

Custom Actions

Exceptions