FlowsAgent

Agent with Pipecat Flows integration for structured conversations

Overview

FlowsAgent extends BaseAgent with Pipecat Flows for structured conversation flows. It manages a FlowManager for node-based conversations with functions, transitions, and actions.

On first activation the flow starts at build_initial_node(). Subsequent activations resume from build_resume_node().

Requires the flows extra: uv add "pipecat-ai-subagents[flows]"

Imported from pipecat_subagents.agents.flows_agent:

from pipecat_subagents.agents.flows_agent import FlowsAgent

class MyFlowsAgent(FlowsAgent):
    def build_llm(self):
        return OpenAILLMService(api_key="...")

    def build_initial_node(self):
        return {"name": "start", ...}

Configuration

Inherits all parameters from BaseAgent.

name · str · required

Unique name for this agent.

bus · AgentBus · required

The AgentBus for inter-agent communication.

context_aggregator · Any · required

The context aggregator pair for managing LLM conversation context, forwarded to FlowManager.

context_strategy · ContextStrategyConfig | None · default: "None"

Optional context strategy forwarded to FlowManager. See ContextStrategyConfig.

global_functions · list[FlowsFunctionSchema | FlowsDirectFunction] | None · default: "None"

Optional list of functions available at every node, forwarded to FlowManager. Methods decorated with @tool are automatically included as global functions.

active · bool · default: "False"

Whether the agent starts active. Defaults to False.

bridged · tuple[str, ...] | None · default: "()"

Bridge configuration. Defaults to an empty tuple (bridged, accepting all bridges).

Properties

Inherits all properties from BaseAgent.

flow_manager

agent.flow_manager -> FlowManager | None

The FlowManager instance, available after the pipeline task is created.

Abstract Methods

build_llm

@abstractmethod
def build_llm(self) -> LLMService

Return the LLM service for this agent.

Returns: An LLMService instance.

build_initial_node

@abstractmethod
def build_initial_node(self) -> NodeConfig

Return the initial flow node configuration. Called on first activation.

Returns: A NodeConfig describing the first node of the flow.

Methods

build_resume_node

def build_resume_node(self) -> NodeConfig

Return the node to resume from when re-entering this agent. Called on subsequent activations (after the first). Override to resume from a specific point in the flow based on flow_manager.state. Defaults to restarting from the initial node.

Returns: A NodeConfig for the resumption point.

on_activated

async def on_activated(self, args: dict | None) -> None

Initialize or resume the flow on activation. On first call, initializes the flow at build_initial_node(). On subsequent calls, sets the node from build_resume_node().

Parameter	Type	Description
args	`dict	None`	Optional activation arguments