PipelineWorker

PipelineWorker manages Pipecat pipeline execution and lifecycle inside the multi-agent framework.

​

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

pipeline · BasePipeline · required

The pipeline to execute.

params · PipelineParams · default: "PipelineParams()"

Configuration parameters for the pipeline. See PipelineParams for details.

observers · list[BaseObserver] | None · default: "None"

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

clock · BaseClock · default: "SystemClock()"

Clock implementation for timing operations.

task_manager · BaseTaskManager | None · default: "None"

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

check_dangling_tasks · 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.

name · str | None · default: "None"

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

active · 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.

bridged · 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.

exclude_frames · tuple[type[Frame], ...] | None · default: "None"

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

handle_flush_frame · 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.

idle_timeout_secs · 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.

idle_timeout_frames · 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.

cancel_on_idle_timeout · bool · default: "True"

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

cancel_runner_on_idle_timeout · 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.

cancel_timeout_secs · float · default: "20.0"

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

enable_tracing · 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.

enable_turn_tracking · bool · default: "True"

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

enable_rtvi · 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.

rtvi_processor · RTVIProcessor | None · default: "None"

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

rtvi_observer_params · RTVIObserverParams | None · default: "None"

Parameters for the RTVI observer added when RTVI is enabled.

conversation_id · str | None · default: "None"

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

additional_span_attributes · dict | None · default: "None"

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

setup_timeout_secs · 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.

start_timeout_secs · 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.

processor_unusable_policy · 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.

app_resources · 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.

tool_resources · Any · default: "None" · deprecated

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

capabilities · 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

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.

run(params) · 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()

stop_when_done() · async

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

await worker.stop_when_done()

end(reason=None) · 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")

cancel(*, reason=None) · 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()

has_finished() · bool

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

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

​

Frame Management

flush_pipeline(timeout=5.0) · 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.

queue_frame(frame, direction=FrameDirection.DOWNSTREAM) · 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)

queue_frames(frames, direction=FrameDirection.DOWNSTREAM) · 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.