FlowManager

Overview

FlowManager orchestrates conversation flows by managing state transitions, function registration, and message handling across different LLM providers.

Configuration

All parameters are keyword-only.

worker

PipelineWorker

required

Pipeline worker instance used for queueing frames into the pipeline.

task

PipelineWorker

deprecated

Deprecated: Use worker instead. This parameter is deprecated and will be removed in 2.0.0.Pipeline worker instance used for queueing frames into the pipeline.

llm

LLMService | LLMSwitcher

required

LLM service instance or an LLMSwitcher for switching between providers at runtime. Any LLMService subclass is supported, including OpenAI-compatible services (Groq, Together, Cerebras, DeepSeek, etc.), Anthropic, Google Gemini, and AWS Bedrock.

context_aggregator

Any

required

Context aggregator for managing conversation context. Typically created using LLMContextAggregatorPair from pipecat.processors.aggregators.llm_response_universal.

context_strategy

ContextStrategyConfig

Default context strategy for managing conversation context during node transitions. Can be overridden per-node via NodeConfig.context_strategy. See ContextStrategyConfig.

transport

BaseTransport

default:"None"

Transport instance for communication (e.g., DailyTransport). When provided, accessible via the transport property in function and action handlers.

global_functions

list[FlowsFunctionSchema | FlowsDirectFunction]

default:"None"

Functions that will be available at every node. These are registered once during initialization and automatically included alongside node-specific functions. Useful for capabilities like “transfer to human” that should be accessible from any conversation state.

Properties

state

flow_manager.state -> dict[str, Any]

Shared state dictionary that persists across node transitions. Use this to store and retrieve conversation data such as user preferences, collected information, or any data that needs to be accessible across different nodes.

# Store data
flow_manager.state["user_name"] = "Alice"

# Retrieve data
name = flow_manager.state.get("user_name", "Unknown")

transport

flow_manager.transport -> BaseTransport | None

The transport instance provided during initialization, or None if not set. Use this to interact with the communication platform (e.g., mute participants, access room info).

async def my_handler(args, flow_manager):
    transport = flow_manager.transport
    if transport:
        participants = transport.participants()

current_node

flow_manager.current_node -> str | None

The identifier of the currently active conversation node. Returns None before initialization or if no node has been set.

async def my_handler(args, flow_manager):
    if flow_manager.current_node == "collecting_payment":
        await setup_secure_session(flow_manager)

worker

flow_manager.worker -> PipelineWorker

The pipeline worker instance used for frame queueing. Use this for advanced flow control such as queuing custom frames.

async def my_handler(args, flow_manager):
    from pipecat.frames.frames import TTSUpdateSettingsFrame
    await flow_manager.worker.queue_frame(
        TTSUpdateSettingsFrame(settings={"voice": "new-voice-id"})
    )

task

Deprecated: Use worker instead. This property is deprecated and will be removed in 2.0.0.

flow_manager.task -> PipelineWorker

Alias for worker. Maintained for backward compatibility.

Methods

initialize

await flow_manager.initialize(initial_node: NodeConfig | None = None) -> None

Initialize the flow manager. Must be called before any node transitions can occur.

Parameter	Type	Default	Description
`initial_node`	`NodeConfig`	`None`	Initial node configuration. Can also be set later via `set_node_from_config()`.

Raises: FlowInitializationError if initialization fails.

flow_manager = FlowManager(worker=worker, llm=llm, context_aggregator=context_aggregator)
await flow_manager.initialize(initial_node=create_initial_node())

set_node_from_config

await flow_manager.set_node_from_config(node_config: NodeConfig) -> None

Transition to a new conversation node. Used to manually trigger node transitions. The node name is taken from the name field in the config, or a UUID is generated if not provided.

Parameter	Type	Description
`node_config`	`NodeConfig`	Configuration for the new node.

Raises: FlowTransitionError if the manager is not initialized. FlowError if node setup fails.

In most cases, prefer returning the next node from a consolidated function handler instead of calling this method directly.

await flow_manager.set_node_from_config({
    "name": "collect_email",
    "task_messages": [{"role": "developer", "content": "Ask the user for their email."}],
    "functions": [collect_email_function],
})

get_current_context

flow_manager.get_current_context() -> list[dict]

Get the current conversation context as a list of messages, including developer messages, user messages, and assistant responses. Raises: FlowError if the context aggregator is not available.

messages = flow_manager.get_current_context()

register_action

flow_manager.register_action(action_type: str, handler: Callable) -> None

Register a handler for a custom action type. The handler can be either a modern handler (action, flow_manager) or a legacy handler (action). Legacy single-argument handlers are deprecated and will be removed in 2.0.0.

Parameter	Type	Description
`action_type`	`str`	String identifier for the action (e.g., `"notify_slack"`).
`handler`	`Callable`	Async function that handles the action.

async def notify_slack(action: dict, flow_manager: FlowManager):
    channel = action.get("channel", "#general")
    text = action.get("text", "")
    await slack_client.post_message(channel=channel, text=text)

flow_manager.register_action("notify_slack", notify_slack)

Once registered, the action can be used in node pre_actions or post_actions:

node_config = NodeConfig(
    task_messages=[...],
    pre_actions=[{"type": "notify_slack", "channel": "#support", "text": "New session started"}],
)

Usage

Basic Setup

from pipecat.flows import FlowManager, FlowResult, NodeConfig

async def create_initial_node() -> NodeConfig:
    return NodeConfig(
        task_messages=[
            {"role": "developer", "content": "Greet the user and ask how you can help."}
        ],
        functions=[help_function],
    )

flow_manager = FlowManager(
    worker=worker,
    llm=llm,
    context_aggregator=context_aggregator,
    transport=transport,
)
await flow_manager.initialize(initial_node=await create_initial_node())

Using Global Functions

from pipecat.flows import FlowManager, NodeConfig

async def transfer_to_human(flow_manager: FlowManager) -> tuple[None, NodeConfig]:
    """Transfer the conversation to a human agent."""
    return None, create_human_agent_node()

flow_manager = FlowManager(
    worker=worker,
    llm=llm,
    context_aggregator=context_aggregator,
    global_functions=[transfer_to_human],  # available at every node
)

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Pipecat Context Hub

Overview

Configuration

Properties

state

transport

current_node

worker

task

Methods

initialize

set_node_from_config

get_current_context

register_action

Usage

Basic Setup

Using Global Functions

​Overview

​Configuration

​Properties

​state

​transport

​current_node

​worker

​task

​Methods

​initialize

​set_node_from_config

​get_current_context

​register_action

​Usage

​Basic Setup

​Using Global Functions

Overview

Configuration

Properties

state

transport

current_node

worker

task

Methods

initialize

set_node_from_config

get_current_context

register_action

Usage

Basic Setup

Using Global Functions