PipelineWorker

Overview

PipelineWorker is the central class for managing pipeline execution. It handles the lifecycle of the pipeline, processes frames in both directions, manages task cancellation, and provides event handlers for monitoring pipeline activity.

PipelineWorker builds on BaseWorker, so it can take part in a multi-worker system. See Workers for the full worker model. PipelineTask is a deprecated alias for PipelineWorker; existing code that uses it keeps working.

Basic Usage

from pipecat.pipeline.pipeline import Pipeline
from pipecat.workers.runner import WorkerRunner
from pipecat.pipeline.worker import PipelineParams, PipelineWorker

# Create a pipeline
pipeline = Pipeline([...])

# Create a task with the pipeline
task = PipelineWorker(pipeline)

# Queue frames for processing
await task.queue_frame(TTSSpeakFrame("Hello, how can I help you today?"))

# Run the pipeline
runner = WorkerRunner()
await runner.run(task)

Constructor Parameters

BasePipeline

required

The pipeline to execute.

PipelineParams

default:"PipelineParams()"

Configuration parameters for the pipeline. See PipelineParams for details.

List[BaseObserver]

default:"[]"

List of observers for monitoring pipeline execution. See Observers for details.

BaseClock

default:"SystemClock()"

Clock implementation for timing operations.

BaseTaskManager | None

default:"None"

Custom task manager for handling asyncio tasks. If None, a default TaskManager is used.

bool

default:"True"

Whether to check for processors’ tasks finishing properly.

float | None

default:"300"

Timeout in seconds before considering the pipeline idle. Set to None to disable idle detection. See Pipeline Idle Detection for details.

Tuple[Type[Frame], ...]

default:"(BotSpeakingFrame, UserSpeakingFrame)"

Frame types that should prevent the pipeline from being considered idle. See Pipeline Idle Detection for details.

bool

default:"True"

Whether to automatically cancel the pipeline task when idle timeout is reached. See Pipeline Idle Detection for details.

bool

default:"False"

Whether to enable OpenTelemetry tracing. See The OpenTelemetry guide for details.

bool

default:"True"

Whether to enable turn tracking. See The OpenTelemetry guide for details.

str | None

default:"None"

Custom ID for the conversation. If not provided, a UUID will be generated. See The OpenTelemetry guide for details.

dict | None

default:"None"

Any additional attributes to add to top-level OpenTelemetry conversation span. See The OpenTelemetry guide for details.

Any

default:"None"

Application-defined bag of resources (database handles, API clients, state, etc.) shared across tool handlers. Passed by reference to every function handler via FunctionCallParams.app_resources. The framework never copies or clears this object; the caller retains their handle and can read mutations after the task finishes.

Any

default:"None"

Deprecated alias for app_resources. Use app_resources in new code.

Methods

Task Lifecycle Management

async

Starts and manages the pipeline execution until completion or cancellation. Typically called via WorkerRunner rather than directly:

runner = WorkerRunner()
await runner.run(task)

async

Sends an EndFrame to the pipeline to gracefully stop the task after all queued frames have been processed.

await task.stop_when_done()

async

Stops the running pipeline immediately by sending a CancelFrame.

  await task.cancel()

bool

Returns whether the task has finished (all processors have stopped).

if task.has_finished(): print("Task is complete")

Frame Management

async

Queues a single frame to be pushed through the pipeline.Downstream frames are pushed from the beginning of the pipeline. Upstream frames are pushed from the end of the pipeline.Parameters:

Parameter	Type	Default	Description
`frame`	`Frame`	(required)	The frame to be processed
`direction`	`FrameDirection`	`FrameDirection.DOWNSTREAM`	The direction to push the frame

# Push a frame downstream (default behavior)
await task.queue_frame(TTSSpeakFrame("Hello!"))

# Push a frame upstream from the end of the pipeline
from pipecat.processors.frame_processor import FrameDirection
await task.queue_frame(UserStoppedSpeakingFrame(), direction=FrameDirection.UPSTREAM)

async

Queues multiple frames to be pushed through the pipeline.Downstream frames are pushed from the beginning of the pipeline. Upstream frames are pushed from the end of the pipeline.Parameters:

Parameter	Type	Default	Description
`frames`	`Iterable[Frame] \| AsyncIterable[Frame]`	(required)	An iterable or async iterable of frames
`direction`	`FrameDirection`	`FrameDirection.DOWNSTREAM`	The direction to push the frames

# Push frames downstream (default behavior)
frames = [TTSSpeakFrame("Hello!"), TTSSpeakFrame("How are you?")]
await task.queue_frames(frames)

# Push frames upstream from the end of the pipeline
from pipecat.processors.frame_processor import FrameDirection
frames = [TranscriptionFrame("user input"), UserStoppedSpeakingFrame()]
await task.queue_frames(frames, direction=FrameDirection.UPSTREAM)

Event Handlers

PipelineWorker provides event handlers for monitoring pipeline lifecycle and frame flow. Register handlers using the @event_handler decorator.

Event	Description
`on_pipeline_started`	Pipeline has started processing
`on_pipeline_finished`	Pipeline reached a terminal state
`on_pipeline_error`	An error frame reached the pipeline task
`on_frame_reached_upstream`	A filtered frame type reached the pipeline source
`on_frame_reached_downstream`	A filtered frame type reached the pipeline sink
`on_idle_timeout`	No activity detected within the idle timeout period

on_pipeline_started

Fired when the StartFrame has been processed by all processors in the pipeline. This indicates the pipeline is fully initialized and running.

@task.event_handler("on_pipeline_started")
async def on_pipeline_started(task, frame):
    print("Pipeline is running!")

Parameters:

Parameter	Type	Description
`task`	`PipelineWorker`	The pipeline task instance
`frame`	`StartFrame`	The start frame that was processed

on_pipeline_finished

Fired after the pipeline reaches any terminal state. This includes normal completion (EndFrame), explicit stop (StopFrame), or cancellation (CancelFrame). Use this event for cleanup, logging, or post-processing.

@task.event_handler("on_pipeline_finished")
async def on_pipeline_finished(task, frame):
    if isinstance(frame, EndFrame):
        print("Pipeline ended normally")
    elif isinstance(frame, CancelFrame):
        print("Pipeline was cancelled")
    elif isinstance(frame, StopFrame):
        print("Pipeline was stopped")

Parameters:

Parameter	Type	Description
`task`	`PipelineWorker`	The pipeline task instance
`frame`	`Frame`	The terminal frame (`EndFrame`, `StopFrame`, or `CancelFrame`)

on_pipeline_error

Fired when an ErrorFrame reaches the pipeline task (upstream from a processor). If the error is fatal, the pipeline will be cancelled after this handler runs.

@task.event_handler("on_pipeline_error")
async def on_pipeline_error(task, frame):
    print(f"Pipeline error: {frame.error}")
    if frame.fatal:
        print("Fatal error — pipeline will be cancelled")

Parameters:

Parameter	Type	Description
`task`	`PipelineWorker`	The pipeline task instance
`frame`	`ErrorFrame`	The error frame with error details

on_frame_reached_upstream

Fired when a frame of a registered type reaches the pipeline source (the start of the pipeline). You must configure which frame types trigger this event using set_reached_upstream_filter() or add_reached_upstream_filter().

from pipecat.frames.frames import TranscriptionFrame

# Configure which frame types to monitor
task.set_reached_upstream_filter((TranscriptionFrame,))

@task.event_handler("on_frame_reached_upstream")
async def on_frame_reached_upstream(task, frame):
    print(f"Frame reached upstream: {frame}")

Parameters:

Parameter	Type	Description
`task`	`PipelineWorker`	The pipeline task instance
`frame`	`Frame`	The frame that reached the pipeline source

This event only fires for frame types you’ve explicitly registered. By default, no frame types are monitored. This is for efficiency — checking every frame would be wasteful when you typically only care about specific types.

on_frame_reached_downstream

Fired when a frame of a registered type reaches the pipeline sink (the end of the pipeline). You must configure which frame types trigger this event using set_reached_downstream_filter() or add_reached_downstream_filter().

from pipecat.frames.frames import TTSAudioRawFrame

# Configure which frame types to monitor
task.set_reached_downstream_filter((TTSAudioRawFrame,))

@task.event_handler("on_frame_reached_downstream")
async def on_frame_reached_downstream(task, frame):
    print(f"Frame reached downstream: {frame}")

Parameters:

Parameter	Type	Description
`task`	`PipelineWorker`	The pipeline task instance
`frame`	`Frame`	The frame that reached the pipeline sink

on_idle_timeout

Fired when no activity frames (as specified by idle_timeout_frames) have been received within the idle timeout period. See Pipeline Idle Detection for configuration details.

@task.event_handler("on_idle_timeout")
async def on_idle_timeout(task):
    print("Pipeline has been idle too long")
    await task.queue_frame(TTSSpeakFrame("Are you still there?"))

Parameters:

Parameter	Type	Description
`task`	`PipelineWorker`	The pipeline task instance

If cancel_on_idle_timeout is True (the default), the pipeline will be automatically cancelled after this handler runs. Set it to False if you want to handle idle timeouts yourself.

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Overview

Basic Usage

Constructor Parameters

Methods

Task Lifecycle Management

Frame Management

Event Handlers

on_pipeline_started

on_pipeline_finished

on_pipeline_error

on_frame_reached_upstream

on_frame_reached_downstream

on_idle_timeout

​Overview

​Basic Usage

​Constructor Parameters

​Methods

​Task Lifecycle Management

​Frame Management

​Event Handlers

​on_pipeline_started

​on_pipeline_finished

​on_pipeline_error

​on_frame_reached_upstream

​on_frame_reached_downstream

​on_idle_timeout

Overview

Basic Usage

Constructor Parameters

Methods

Task Lifecycle Management

Frame Management

Event Handlers

on_pipeline_started

on_pipeline_finished

on_pipeline_error

on_frame_reached_upstream

on_frame_reached_downstream

on_idle_timeout