Type alias for function handler arguments.

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

Configuration for the entire conversation flow.

Configuration for a single node in the flow.

# Example name
"name": "greeting"

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

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

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

# Using provider-specific dictionary format
"functions": [
    {
        "type": "function",
        "function": {
            "name": "get_current_movies",
            "handler": get_movies,
            "description": "Fetch movies currently playing",
            "parameters": {...}
        },
    }
]

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

# Using direct functions (auto-configuration)
"functions": [get_movies]

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

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

# Example usage
"respond_immediately": False

Legacy function handler that only receives arguments.

Returns either:

• A FlowResult (⚠️ deprecated)
• A “consolidated” result tuple (result, next node) where:
  • result is an optional FlowResult
  • next node is an optional NodeConfig (for dynamic flows) or string (for static flows)

Modern function handler that receives both arguments and FlowManager.

Returns either:

• A FlowResult (⚠️ deprecated)
• A “consolidated” result tuple (result, next node) where:
  • result is an optional FlowResult
  • next node is an optional NodeConfig (for dynamic flows) or string (for static flows)

Function that is meant to be passed directly into a NodeConfig rather than into the handler field of a function configuration.

It must be an async function with flow_manager: FlowManager as its first parameter.

It must return a ConsolidatedFunctionResult, which is a tuple (result, next node) where:

• result is an optional FlowResult
• next node is an optional NodeConfig (for dynamic flows) or string (for static flows)

Strategy for managing conversation context during node transitions.

Configuration for context management strategy.

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

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

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
)

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

# Pass to flow manager
await flow_manager.set_node_from_config(node_config)

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

Initialize the flow with starting messages.

Set up a new conversation node programmatically (dynamic flows only).

• In dynamic flows, the application advances the conversation using set_node to set up each next node.
• In static flows, set_node is triggered under the hood when a node contains a transition_to field.

Set up a new conversation node programmatically (dynamic flows only).

Note that this method should only be used in rare circumstances. Most often, you should:

• Prefer “consolidated” function handlers that return a tuple (result, next node), which implicitly sets up the next node
• Prefer passing your initial node to FlowManager.initialize()

Register a handler for a custom action type.

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) -> tuple[FlowResult, str]:
    context = flow_manager.get_current_context()
    # Process conversation history
    return {"status": "success"}, "next"

Functions that execute operations within a single conversational state, without switching nodes.

from pipecat_flows import FlowArgs, FlowResult

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

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

Functions that specify a transition between nodes (optionally processing data first).

from pipecat_flows import FlowArgs, FlowResult

async def next_step(args: FlowArgs) -> tuple[None, str]:
    """Specify the next node to transition to."""
    return None, "target_node" # Return NodeConfig instead of str for dynamic flows

# Function configuration
{
    "type": "function",
    "function": {
        "name": "next_step",
        "handler": next_step,
        "description": "Transition to next node",
        "parameters": {"type": "object", "properties": {}}
    }
}

Async function that processes data within a node and/or specifies the next node (more details here). 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": {...}
    }
}

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_from_config(create_confirmation_node())
    else:
        await flow_manager.set_node_from_config(
            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_from_config(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_from_config(create_confirmation_node())
    else:
        await flow_manager.set_node_from_config(
            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.

Speaks text immediately using the TTS service.

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

Ends the conversation and closes the connection.

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

Calls the developer-defined function at the appropriate time.

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

String identifier for the action

Async or sync function that handles the action

Base exception for all flow-related errors.

from pipecat_flows import FlowError

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

Raised when flow initialization fails.

from pipecat_flows import FlowInitializationError

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

Raised when a state transition fails.

from pipecat_flows import FlowTransitionError

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

Raised when an invalid or unavailable function is specified.

from pipecat_flows import InvalidFunctionError

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

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