FlowManager

Overview

FlowManager orchestrates conversation flows by managing state transitions, function registration, and message handling across different LLM providers. It drives both forms of flow: nodes built in Python, and nodes a Flow produced from a flow config.

Placeholder Rendering

On entering a node, the manager fills {{ key }} placeholders in the node’s prompts from state, before anything is sent to the LLM. It renders the node’s role_message, the content of each of its task_messages, and the text of any tts_say action. Rendering happens on every entry, not once at load, so a prompt sees whatever handlers have stored by then — including after a context reset, and again each time a node is re-entered. {{ order.size }} walks into a stored mapping, values are rendered with str(), and \{{ key }} is left as the literal {{ key }}. A placeholder naming a key that is not in state raises FlowError on entry.

Configuration

All parameters are keyword-only.

PipelineWorker

required

Pipeline worker instance used for queueing frames into the pipeline.

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.

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.

Any

required

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

ContextStrategyConfig

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

BaseTransport

default:"None"

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

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.With a flow config, pass flow.global_functions, which holds the config’s global_functions already resolved to your handlers. It is a fresh list each time, so you can extend it with functions defined in code.

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())

With a flow config, the starting node comes from the flow:

flow = Flow(FlowConfig.from_file("flow.yaml"), handlers=handlers)

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

Set any state the config’s prompts refer to as {{ key }} before calling initialize(), since the initial node is rendered as it is entered.

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, 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
)

Get Started

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Pipecat Context Hub

Overview

Placeholder Rendering

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

​Placeholder Rendering

​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

Placeholder Rendering

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