Types

Type definitions for the conversation flow system.

This module defines the core types used throughout the flow system:

  • FlowResult: Function return type

  • FlowArgs: Function argument type

  • NodeConfig: Node configuration type

  • FlowsFunctionSchema: A uniform schema for function calls in flows

These types provide structure and validation for flow configurations and function interactions.

class pipecat_flows.types.FlowResult[source]

Bases: TypedDict

Optional convention TypedDict for status/error results.

Deprecated since version 1.x.0: FlowResult is no longer required or referenced by any handler type in the library, and Pipecat’s upstream function-call-result contract is Any. Define your own TypedDict if you want a structured result, or return any JSON-serializable value. FlowResult will be removed in 2.0.0.

Parameters:

  • status – Status of the function execution.

  • error – Optional error message if execution failed.

status: str
error: str
pipecat_flows.types.FlowArgs

Type alias for function handler arguments.

Each invocation gets its own dict, so handlers may mutate it freely without affecting Pipecat’s internal state.

Note

In 2.0.0 this alias is planned to widen to Mapping[str, Any] to align with Pipecat’s typing of FunctionCallParams.arguments. Handlers that only read args will be unaffected; handlers that mutate args will need to keep the annotation as dict[str, Any] explicitly.

Example:

{
    "user_name": "John",
    "age": 25,
    "preferences": {"color": "blue"}
}

alias of dict[str, Any]

pipecat_flows.types.LegacyActionHandler

Legacy action handler type that only receives the action dictionary.

Deprecated since version 1.x.0: Use FlowActionHandler ((action, flow_manager)) instead. Will be removed in 2.0.0.

Parameters:

action – Dictionary containing action configuration and parameters.

Example:

async def simple_handler(action: dict):
    await notify(action["text"])

alias of Callable[[dict[str, Any]], Awaitable[None]]

pipecat_flows.types.FlowActionHandler

Modern action handler type that receives both action and flow_manager.

Parameters:

  • action – Dictionary containing action configuration and parameters.

  • flow_manager – Reference to the FlowManager instance.

Example:

async def advanced_handler(action: dict, flow_manager: FlowManager):
    await flow_manager.transport.notify(action["text"])

alias of Callable[[dict[str, Any], FlowManager], Awaitable[None]]

class pipecat_flows.types.ActionConfig[source]

Bases: TypedDict

Configuration for an action.

Parameters:

  • type – Action type identifier (e.g. “tts_say”, “notify_slack”).

  • handler – Callable to handle the action.

  • text – Text for tts_say action.

Note

Additional fields are allowed and passed to the handler.

type: Required[str]
handler: Callable[[dict[str, Any]], Awaitable[None]] | Callable[[dict[str, Any], FlowManager], Awaitable[None]]
text: str
class pipecat_flows.types.ContextStrategy(*values)[source]

Bases: Enum

Strategy for managing context during node transitions.

Parameters:

  • APPEND – Append new messages to existing context (default).

  • RESET – Reset context with new messages only.

  • RESET_WITH_SUMMARY –

    Reset context but include an LLM-generated summary.

    Deprecated since version 1.0.0: Use Pipecat’s native context summarization instead. To trigger on-demand summarization during a node transition, push an LLMSummarizeContextFrame in a pre-action. See https://docs.pipecat.ai/guides/fundamentals/context-summarization Will be removed in 2.0.0.

APPEND = 'append'
RESET = 'reset'
RESET_WITH_SUMMARY = 'reset_with_summary'
class pipecat_flows.types.ContextStrategyConfig(strategy: ContextStrategy, summary_prompt: str | None = None)[source]

Bases: object

Configuration for context management.

Parameters:

  • strategy – Strategy to use for context management.

  • summary_prompt –

    Required prompt text when using RESET_WITH_SUMMARY.

    Deprecated since version 1.0.0: Deprecated along with RESET_WITH_SUMMARY. Use LLMContextSummaryConfig.summarization_prompt instead. Will be removed in 2.0.0.

strategy: ContextStrategy
summary_prompt: str | None = None
class pipecat_flows.types.NodeConfig[source]

Bases: TypedDict

Configuration for a single node in the flow.

Parameters:

  • task_messages – List of message dicts defining the current node’s objectives.

  • name – Name of the node, useful for debug logging when returning a next node from a “consolidated” function.

  • role_message – The bot’s role/personality as a plain string, sent as the LLM’s system instruction via LLMUpdateSettingsFrame. When provided, the system instruction persists across node transitions until a new node explicitly sets role_message again.

  • role_messages –

    Deprecated list-of-dicts format for the bot’s role/personality.

    Deprecated since version 0.0.24: Use role_message (str) instead. Will be removed in 2.0.0.

  • functions – List of FlowsFunctionSchema definitions or direct functions whose definitions are automatically extracted from their signatures.

  • pre_actions – Actions to execute before LLM inference.

  • post_actions – Actions to execute after LLM inference.

  • context_strategy – Strategy for updating context during transitions.

  • respond_immediately – Whether to run LLM inference as soon as the node is set (default: True).

Example:

{
    "role_message": "You are a helpful assistant...",
    "task_messages": [
        {
            "role": "developer",
            "content": "Ask the user for their name..."
        }
    ],
    "functions": [...],
    "pre_actions": [...],
    "post_actions": [...],
    "context_strategy": ContextStrategyConfig(strategy=ContextStrategy.APPEND),
    "respond_immediately": true,
}
task_messages: Required[list[dict]]
name: str
role_message: str
role_messages: list[dict[str, Any]]
functions: list[FlowsFunctionSchema | Callable[[...], Awaitable[tuple[Any, NodeConfig | None]]]]
pre_actions: list[ActionConfig]
post_actions: list[ActionConfig]
context_strategy: ContextStrategyConfig
respond_immediately: bool
pipecat_flows.types.ConsolidatedFunctionResult

Return type for “consolidated” functions.

Return type for “consolidated” functions that do either or both of: - doing some work - specifying the next node to transition to after the work is done

The first tuple element is the function-call result delivered to the LLM. Any JSON-serializable value is accepted (matching Pipecat’s upstream FunctionCallResultCallback contract). Pass None to signal a transition-only handler; FlowManager substitutes an acknowledgement result.

alias of tuple[Any, NodeConfig | None]

pipecat_flows.types.ZeroArgFunctionHandler

Function handler that takes no arguments.

Deprecated since version 1.x.0: Use FlowFunctionHandler ((args, flow_manager)) instead. Will be removed in 2.0.0.

Returns:

Any JSON-serializable value, or a ConsolidatedFunctionResult tuple to also specify the next node.

alias of Callable[[], Awaitable[Any]]

pipecat_flows.types.LegacyFunctionHandler

Legacy function handler that only receives arguments.

Deprecated since version 1.x.0: Use FlowFunctionHandler ((args, flow_manager)) instead. Will be removed in 2.0.0.

Parameters:

args – Dictionary of arguments from the function call.

Returns:

Any JSON-serializable value, or a ConsolidatedFunctionResult tuple to also specify the next node.

alias of Callable[[dict[str, Any]], Awaitable[Any]]

pipecat_flows.types.FlowFunctionHandler

Modern function handler that receives both arguments and flow_manager.

Parameters:

  • args – Dictionary of arguments from the function call.

  • flow_manager – Reference to the FlowManager instance.

Returns:

Any JSON-serializable value, or a ConsolidatedFunctionResult tuple to also specify the next node.

alias of Callable[[dict[str, Any], FlowManager], Awaitable[Any]]

pipecat_flows.types.FunctionHandler = collections.abc.Callable[[], collections.abc.Awaitable[typing.Any]] | collections.abc.Callable[[dict[str, typing.Any]], collections.abc.Awaitable[typing.Any]] | collections.abc.Callable[[dict[str, typing.Any], 'FlowManager'], collections.abc.Awaitable[typing.Any]]

Union type for function handlers supporting 0-arg, legacy, and modern patterns.

pipecat_flows.types.FlowsDirectFunction

Type alias for “direct” functions with automatic metadata extraction.

“Direct” functions have their definition automatically extracted from the function signature and docstring. This can be used in NodeConfig directly, in lieu of a FlowsFunctionSchema or function definition dict.

Expected shape:

async def f(flow_manager: FlowManager, **params) -> ConsolidatedFunctionResult:
    ...

where **params are any named parameters described by the function’s docstring.

This is defined as Callable[..., ...] rather than a Protocol because Python’s Protocol system cannot express “any concrete named-parameter list” against **kwargs: Any — a function with named params like llm: str is not structurally compatible with a **kwargs: Any protocol signature. Runtime validation of the expected shape happens in FlowsDirectFunctionWrapper.validate_function().

alias of Callable[[…], Awaitable[tuple[Any, NodeConfig | None]]]

class pipecat_flows.types.FlowsFunctionSchema(name: str, description: str, properties: dict[str, Any], required: list[str], handler: Callable[[], Awaitable[Any]] | Callable[[dict[str, Any]], Awaitable[Any]] | Callable[[dict[str, Any], FlowManager], Awaitable[Any]] | None = None, cancel_on_interruption: bool = False, timeout_secs: float | None = None)[source]

Bases: object

Function schema with Flows-specific properties.

This class extends standard function schemas with additional fields for Pipecat Flows integration including handler assignment and transition logic.

Parameters:

  • name – Name of the function.

  • description – Description of the function.

  • properties – Dictionary defining parameter types and descriptions.

  • required – List of required parameter names.

  • handler – Function handler to process the function call.

  • cancel_on_interruption – Whether to cancel this function call when an interruption occurs. Defaults to False.

  • timeout_secs – Optional per-tool timeout in seconds, overriding the global function_call_timeout_secs. Defaults to None (use global timeout).

name: str
description: str
properties: dict[str, Any]
required: list[str]
handler: Callable[[], Awaitable[Any]] | Callable[[dict[str, Any]], Awaitable[Any]] | Callable[[dict[str, Any], FlowManager], Awaitable[Any]] | None = None
cancel_on_interruption: bool = False
timeout_secs: float | None = None
to_function_schema() → FunctionSchema[source]

Convert to a standard FunctionSchema for use with LLMs.

Returns:

FunctionSchema without flow-specific fields.

pipecat_flows.types.flows_direct_function(*, cancel_on_interruption: bool = False, timeout_secs: float | None = None) → Callable[[Callable], Callable][source]

Decorator to attach additional metadata to a Pipecat direct function.

This metadata can be used, for example, to store the additional arguments that should be used when registering the function with the Pipecat service.

Parameters:

  • cancel_on_interruption – Whether to cancel the function call when the user interrupts. Defaults to False.

  • timeout_secs – Optional per-tool timeout in seconds, overriding the global function_call_timeout_secs. Defaults to None (use global timeout).

Returns:

A decorator that attaches the metadata to the function.

Example:

@flows_direct_function(cancel_on_interruption=False, timeout_secs=30)
async def long_running_task(flow_manager: FlowManager, query: str):
    '''Perform a long-running task that should not be cancelled on interruption.'''
    # ... implementation
    return {"status": "complete"}, None
class pipecat_flows.types.FlowsDirectFunctionWrapper(function: Callable)[source]

Bases: BaseDirectFunctionWrapper

Wrapper around a FlowsDirectFunction for metadata extraction and invocation.

The wrapper:

  • extracts metadata from the function signature and docstring

  • generates a corresponding FunctionSchema

  • helps with function invocation

classmethod special_first_param_name() → str[source]

Get the special first parameter name for Flows direct functions.

Returns:

The string “flow_manager” which is expected as the first parameter.

classmethod validate_function(function: Callable) → None[source]

Validate the function signature and docstring.

Parameters:

function – The function to validate.

Raises:

InvalidFunctionError – If the function does not meet the requirements.

async invoke(args: Mapping[str, Any], flow_manager: FlowManager)[source]

Invoke the wrapped function with the provided arguments.

Parameters:

  • args – Arguments to pass to the function.

  • flow_manager – FlowManager instance for function execution context.

Returns:

The result of the function call.

pipecat_flows.types.get_or_generate_node_name(node_config: NodeConfig) → str[source]

Get the node name from configuration or generate a UUID if not set.

Parameters:

node_config – Node configuration dictionary.

Returns:

Node name from config or generated UUID string.