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.frames.frames import TTSSpeakFrame
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.add_workers(worker)
await runner.run()

Constructor Parameters

BasePipeline

required

The pipeline to execute.

PipelineParams

default:"PipelineParams()"

Configuration parameters for the pipeline. See PipelineParams for details.

list[BaseObserver] | None

default:"None"

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.

str | None

default:"None"

Worker name, used to address the worker on the bus. If None, a name is generated.

bool

default:"True"

Whether the worker starts out accepting bus messages. Start a worker inactive in multi-worker setups where workers take turns, so it does nothing until another worker or the application activates it.

tuple[str, ...] | None

default:"None"

Bridge configuration. None means the pipeline is not bridged. An empty tuple () accepts frames from all bridges, and a tuple of names like ("voice",) accepts frames only from those bridges. The bus comes from the runner.

tuple[type[Frame], ...] | None

default:"None"

When bridged is set, extra frame types that should not cross the bus. Lifecycle frames are always excluded.

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], ...]

Frame types that should prevent the pipeline from being considered idle. The default pairs the VAD-only UserSpeakingFrame with the turn and transcription frames a provider-driven pipeline reports instead. 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:"True"

When cancel_on_idle_timeout is also True, whether reaching the idle timeout also cancels every other worker in the WorkerRunner. Set to False for a sidecar worker that should cancel itself on idle without bringing down its peers.

float

default:"20.0"

How long to wait, in seconds, for cancellation to finish cleanly.

bool

default:"False"

Whether to enable OpenTelemetry tracing. Has no effect when OpenTelemetry isn’t installed, and turn and conversation spans also require enable_turn_tracking=True. See The OpenTelemetry guide for details.

bool

default:"True"

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

bool | None

default:"None"

Whether to automatically add RTVI support to the pipeline. None adds it unless the pipeline is bridged: a bridged worker has no client of its own, and its RTVI would report every frame a second time as it crosses the bridge. Pass True to enable RTVI on a bridged worker, or False to disable it entirely. If the pipeline already contains an RTVIProcessor and observers contains an RTVIObserver, the worker uses that pair and ignores enable_rtvi, rtvi_processor and rtvi_observer_params. If only one of the two is present, the worker logs an error and sets up no RTVI.

RTVIProcessor | None

default:"None"

The RTVI processor to add when RTVI is enabled. If None, a default RTVIProcessor is created.

RTVIObserverParams | None

default:"None"

Parameters for the RTVI observer added when RTVI is enabled.

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, counted from when every processor has finished setting up. 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

Deprecated since 1.2.0, removed in 2.0.0. Use app_resources instead.

BotCapabilities | None

default:"None"

What the bot does in the session, for fields the pipeline can’t show. Known (non-None) fields replace the ones derived from the pipeline. See capabilities.

Methods

Bot Capabilities

BotCapabilities

Property returning what the bot does in the session. It is derived from the media the pipeline’s transports send and receive and from PipelineParams.enable_metrics, then the capabilities constructor argument is applied on top. RTVI sends it to the client in the bot-ready message.BotCapabilities (from pipecat.pipeline.capabilities) has these optional fields: audio_in, audio_out, video_in, screen_in, video_out and metrics. None means unknown, so the client keeps its default; False means the bot doesn’t do it.

from pipecat.pipeline.capabilities import BotCapabilities

worker = PipelineWorker(pipeline, capabilities=BotCapabilities(video_out=True))
print(worker.capabilities.video_out)

Worker Lifecycle Management

end(), and activate_worker() with deactivate_self=True, 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. params is a WorkerParams (from pipecat.workers.base_worker) carrying the task manager. Typically called via WorkerRunner, which supplies it, rather than directly:

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

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. reason is an optional string saying why. If the pipeline hasn’t started yet, no CancelFrame is sent; the processors still setting up never start and the pipeline is cleaned up instead.

  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 to the sink, then a new probe back up to the source and a third down to the sink again, and waits for that last one to arrive. When it does, every frame queued ahead of the flush has been processed, along with any work a processor started by pushing a frame upstream. The first probe goes on the worker’s push queue, behind whatever is already waiting there. Flushing works through pipelines containing a ParallelPipeline or ServiceSwitcher.Parameters:

Parameter	Type	Default	Description
`timeout`	`float`	`5.0`	Seconds of no progress (a frame reaching the sink, or a progress report from another worker the probe crossed into) 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.frames.frames import UserStoppedSpeakingFrame
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.frames.frames import TranscriptionFrame, UserStoppedSpeakingFrame
from pipecat.processors.frame_processor import FrameDirection
from pipecat.utils.time import time_now_iso8601
frames = [
    TranscriptionFrame("user input", user_id="user", timestamp=time_now_iso8601()),
    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!")

The input transport’s on_client_connected can fire before later services have connected. Use on_pipeline_started for anything that needs every service ready. 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). It doesn’t fire when setup or the StartFrame times out (on_setup_timeout, or on_pipeline_timeout with a StartFrame). Use this event for cleanup, logging, or post-processing.

from pipecat.frames.frames import CancelFrame, EndFrame, StopFrame

@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: 10 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.

# Keep the worker running after the timeout so the prompt is heard
worker = PipelineWorker(pipeline, cancel_on_idle_timeout=False)

@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 worker is cancelled as soon as this event fires, without waiting for the handler to finish, so anything the handler queues isn’t heard. Set it to False if you want to handle idle timeouts yourself.

Get Started

Standards

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Pipecat Context Hub

Overview

Basic Usage

Constructor Parameters

Methods

Bot Capabilities

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

​Bot Capabilities

​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

Bot Capabilities

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