PipelineWorker

Overview

PipelineWorker is the central class for managing pipeline execution. It handles the lifecycle of the pipeline, processes frames in both directions, manages 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 worker with the pipeline
worker = PipelineWorker(pipeline)

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

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

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 warn about tasks left running when the worker finishes. Only applies when the worker owns its task manager; otherwise the runner reports dangling tasks.

bool | None

default:"None"

Whether this worker answers a flush probe. Defaults to whether the pipeline is unbridged, so a bridged worker leaves the probe to travel on and be completed by the pipeline that owns the transport. A bridged worker with no such peer never completes a flush and every flush_pipeline() call waits out its timeout.

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 worker 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.

float

default:"20.0"

How long to wait for every processor to finish setting up. A processor that blocks while connecting would otherwise leave run() waiting on it forever; when the wait runs out the pipeline reports on_setup_timeout and tears down.

float

default:"20.0"

How long to wait for the StartFrame to reach the end of the pipeline. When the wait runs out the pipeline reports on_pipeline_timeout.

ProcessorUnusablePolicy

default:"ProcessorUnusablePolicy.CONTINUE"

What the pipeline does when a processor reports an error that leaves it unable to do its job, such as a service whose API key was rejected. CONTINUE (the default) reports the error and keeps running, leaving the decision to the application; END stops the pipeline gracefully and CANCEL stops it immediately. Applied once per processor, not once per failed request. See Error Handling.

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 worker finishes.

Any

default:"None"

Deprecated alias for app_resources. Use app_resources in new code.

Methods

Worker Lifecycle Management

end() and activate_worker() wait for in-flight frames to be processed before they take effect, so a closing line is heard rather than cut off and a worker handing over stops talking before the one taking over starts. A pipeline that never started, or one that has already finished, is left alone.

async

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

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

async

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

await worker.stop_when_done()

async

Requests a graceful end of the session, draining the pipeline first. Whatever this worker has already pushed reaches the end of the pipeline before the session goes away, so a closing line is heard rather than cut off.

  await worker.end(reason="done")

async

Stops the running pipeline immediately by sending a CancelFrame.

  await worker.cancel()

bool

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

if worker.has_finished(): print("Worker is complete")

Frame Management

async

Wait for in-flight frames to be processed. Pushes a flush probe downstream; the sink bounces it back upstream and the source sets its event once it completes the round-trip, signalling that every frame queued ahead of it has been processed. The probe goes on the worker’s push queue, behind whatever is already waiting there.Parameters:

Parameter	Type	Default	Description
`timeout`	`float`	`5.0`	Seconds of no frames reaching the sink before giving up. A pipeline still working keeps the wait alive; one that is stuck gives up after this much quiet.

drained = await worker.flush_pipeline(timeout=10.0)
if not drained:
    print("Pipeline gave up after timeout")

Returns: True if the pipeline drained, False if it went quiet first.

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 worker.queue_frame(TTSSpeakFrame("Hello!"))

# Push a frame upstream from the end of the pipeline
from pipecat.processors.frame_processor import FrameDirection
await worker.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 worker.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 worker.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 worker
`on_setup_timeout`	Processors never finished setting up
`on_pipeline_timeout`	A frame the worker was waiting on never arrived
`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_heartbeat_timeout`	No heartbeat received within the monitor timeout (pipeline may stall)
`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.

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

Parameters:

Parameter	Type	Description
`worker`	`PipelineWorker`	The pipeline worker 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.

@worker.event_handler("on_pipeline_finished")
async def on_pipeline_finished(worker, 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
`worker`	`PipelineWorker`	The pipeline worker instance
`frame`	`Frame`	The terminal frame (`EndFrame`, `StopFrame`, or `CancelFrame`)

on_setup_timeout

Fired when the processors never finish setting up within setup_timeout_secs. The pipeline is torn down afterwards. Takes no frame — nothing has flowed yet.

@worker.event_handler("on_setup_timeout")
async def on_setup_timeout(worker):
    print("A processor never finished connecting")

Parameters:

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

on_pipeline_timeout

Fired when a frame the worker was waiting on never arrived — a StartFrame that never reached the end of the pipeline within start_timeout_secs, or a CancelFrame that never drained. Inspect frame to tell the two apart.

@worker.event_handler("on_pipeline_timeout")
async def on_pipeline_timeout(worker, frame):
    print(f"Timed out waiting for {frame}")

Parameters:

Parameter	Type	Description
`worker`	`PipelineWorker`	The pipeline worker instance
`frame`	`Frame`	The frame the worker waited on

on_pipeline_error

Fired when an ErrorFrame reaches the pipeline worker (upstream from a processor). If the error left its processor unable to do its job, the pipeline applies its processor_unusable_policy after this handler runs.

@worker.event_handler("on_pipeline_error")
async def on_pipeline_error(worker, frame):
    print(f"Pipeline error: {frame.error}")
    if frame.processor and not frame.processor.is_usable:
        print(f"{frame.processor} can no longer do its job")

Parameters:

Parameter	Type	Description
`worker`	`PipelineWorker`	The pipeline worker 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
worker.set_reached_upstream_filter((TranscriptionFrame,))

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

Parameters:

Parameter	Type	Description
`worker`	`PipelineWorker`	The pipeline worker 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
worker.set_reached_downstream_filter((TTSAudioRawFrame,))

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

Parameters:

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

on_heartbeat_timeout

Fired when a heartbeat frame is not received within the monitor timeout period. This indicates the pipeline may be stalled or not processing frames. The event fires repeatedly every heartbeats_monitor_secs for as long as the stall persists.

@worker.event_handler("on_heartbeat_timeout")
async def on_heartbeat_timeout(worker):
    print("Heartbeat timeout detected — pipeline may be stalled")
    # Optionally cancel the pipeline or take other action
    await worker.cancel()

Parameters:

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

Heartbeat monitoring must be enabled by setting enable_heartbeats=True in PipelineParams. The timeout period is controlled by heartbeats_monitor_secs (default: 5 seconds). See PipelineParams for configuration details.

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.

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

Parameters:

Parameter	Type	Description
`worker`	`PipelineWorker`	The pipeline worker 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.

Get Started

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Pipecat Context Hub

Overview

Basic Usage

Constructor Parameters

Methods

Worker Lifecycle Management

Frame Management

Event Handlers

on_pipeline_started

on_pipeline_finished

on_setup_timeout

on_pipeline_timeout

on_pipeline_error

on_frame_reached_upstream

on_frame_reached_downstream

on_heartbeat_timeout

on_idle_timeout

​Overview

​Basic Usage

​Constructor Parameters

​Methods

​Worker Lifecycle Management

​Frame Management

​Event Handlers

​on_pipeline_started

​on_pipeline_finished

​on_setup_timeout

​on_pipeline_timeout

​on_pipeline_error

​on_frame_reached_upstream

​on_frame_reached_downstream

​on_heartbeat_timeout

​on_idle_timeout

Overview

Basic Usage

Constructor Parameters

Methods

Worker Lifecycle Management

Frame Management

Event Handlers

on_pipeline_started

on_pipeline_finished

on_setup_timeout

on_pipeline_timeout

on_pipeline_error

on_frame_reached_upstream

on_frame_reached_downstream

on_heartbeat_timeout

on_idle_timeout