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

pipeline · BasePipeline · required

The pipeline to execute.

params · PipelineParams · default: "PipelineParams()"

Configuration parameters for the pipeline. See PipelineParams for details.

observers · List[BaseObserver] · default: "[]"

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.

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], ...] · default: "(BotSpeakingFrame, UserSpeakingFrame)"

Frame types that should prevent the pipeline from being considered idle. 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.

enable_tracing · bool · default: "False"

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

enable_turn_tracking · bool · default: "True"

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

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

run() · async

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

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

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() · async

Stops the running pipeline immediately by sending a CancelFrame.

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

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