Pipecat Integration Guide

June 19, 2026 · View on GitHub

This document explains how to take a flow built in the visual editor and run it inside a Pipecat application.

Flow Lifecycle

Design – Build and validate the flow inside the editor. Nodes map directly to Pipecat NodeConfig objects.
Export – Use the toolbar to export either:
- flow.json (schema-compliant Pipecat Flow JSON)
- <flow_name>_flow.py (generated Python scaffolding produced by lib/codegen/pythonGenerator.ts)
Implement handlers – Fill in the TODO sections in the generated Python file (or write your own implementation if consuming JSON manually).
Run in Pipecat – Load the generated node factories, instantiate a FlowManager, and call initialize with the initial node when your transport connects.

Export Formats

JSON

flow.json is the canonical data model used in the editor and contains:

meta, context, global_functions
nodes[] with Pipecat NodeConfig fields (messages, functions, actions, context strategy, respond_immediately)
Function-level routing via next_node_id or decision
Visualization edges derived from the routing data

Use JSON if you want to plug the editor into a custom runtime, build your own generator, or version the declarative representation of the flow.

Generated Python

Selecting Export → Export Python downloads a ready-to-run scaffold targeting pipecat-flows 1.3.0+:

One create_<node_id>_node() factory per node
Direct functions — async def name(flow_manager, ...) stubs whose schema is derived automatically from the type hints and docstring. JSON-Schema constraints (enum / minimum / maximum / pattern) are folded into the docstring Args: section.
Result types as plain TypedDicts (the deprecated FlowResult is no longer used)
A @flows_tool_options(...) decorator when a function sets call options (cancel-on-interruption and/or a per-tool timeout)
Decision routing rendered as Python if / elif / else blocks
Optional context strategy wiring (adds ContextStrategyConfig imports only when needed)
Placeholder FlowManager setup showing where to plug in your Pipecat pipeline and transport events

Each direct function takes flow_manager first, followed by its parameters as typed arguments, and returns (result, next_node) — where result is any JSON-serializable value (or None) and next_node is the NodeConfig to transition to (or None).

Using the Generated Python

Install Pipecat Flows (and your preferred transports/services):

pip install pipecat pipecat-ai-flows

Save the generated file (e.g., food_ordering_flow.py) somewhere importable by your bot.
Wire it into your Pipecat entrypoint:

from pipecat.transports.base_transport import BaseTransport
from pipecat_flows import FlowManager

from food_ordering_flow import (
    create_initial_node,
    # ...any other helpers you want to reference directly
)

flow_manager = FlowManager(
    worker=worker,  # PipelineWorker; `task=` is deprecated
    llm=llm,
    context_aggregator=context_aggregator,
    transport=transport,
    # global_functions=[...],  # uncomment if your flow defines them
)

@transport.event_handler("on_client_connected")
async def on_client_connected(transport: BaseTransport, client):
    await flow_manager.initialize(create_initial_node())

Implement the direct functions generated in the file. Each receives flow_manager plus its typed parameters, so you can:
- Read the user-provided arguments directly as named parameters
- Store intermediate data in flow_manager.state
- Decide what node to visit next by returning its NodeConfig (or rely on decisions/next_node_id)

Decision Handling

If a function in the editor contains a decision:

The generated direct function includes the action block (your expression, evaluated server-side) and returns Any as the result.
Conditions become if/elif/else checks that route to create_<node_id>_node() functions.
The editor also stores optional decision_node_position so re-importing Python-exported JSON maintains the layout of helper decision nodes in the UI.

Using JSON Directly

If you prefer to consume the JSON yourself:

JSON Field	Pipecat Usage
`node.id`	`NodeConfig.name`
`node.data.role_messages`	`NodeConfig.role_message` (collapsed to a string)
`node.data.task_messages`	`NodeConfig.task_messages`
`node.data.functions[]`	Direct functions (auto-derived schema)
`node.data.functions[].next_node_id`	Return value `(…, create_<id>_node())`
`node.data.functions[].decision`	Direct function body with `if/elif/else` routing
`node.data.pre_actions` / `post_actions`	Passed directly into `NodeConfig`
`node.data.context_strategy`	Adds `ContextStrategyConfig`
`global_functions`	Optional functions registered on every node
`edges`	Visualization only (derived from the routing fields)

The TypeBox schema is documented in docs/SCHEMA.md.

Validation & Tooling

JSON exports are validated via Ajv and custom graph checks before they leave the editor.
Python exports run the same validation first; generation is blocked until the flow passes.
Re-importing either JSON or edited Python (converted back to JSON) keeps decision positions, function metadata, and context strategy aligned with the UI.