Startup Timing Observer

The StartupTimingObserver measures what each processor costs to get ready — its setup() and its start() together — and tracks transport connection timing. This is useful for diagnosing startup slowness and identifying initialization bottlenecks such as WebSocket connections, API authentication, or model loading.

Features

Measures per-processor readiness cost, from setup() through StartFrame propagation
Reports total pipeline startup time and per-processor breakdown
Tracks transport connection milestones (bot connected, client connected)
Emits on_startup_timing_report with processor timing data
Emits on_transport_timing_report with transport connection timing
Supports filtering to measure only specific processor types
Excludes internal pipeline processors by default

Usage

Basic Startup Monitoring

Add startup monitoring to your pipeline and handle the events:

from pipecat.observers.startup_timing_observer import StartupTimingObserver

observer = StartupTimingObserver()

@observer.event_handler("on_startup_timing_report")
async def on_startup_timing_report(observer, report):
    print(f"Total startup: {report.total_duration_secs:.3f}s")
    for timing in report.processor_timings:
        print(f"  {timing.processor_name}: {timing.duration_secs:.3f}s")

@observer.event_handler("on_transport_timing_report")
async def on_transport_timing_report(observer, report):
    if report.bot_connected_secs is not None:
        print(f"Bot connected: {report.bot_connected_secs:.3f}s")
    print(f"Client connected: {report.client_connected_secs:.3f}s")

worker = PipelineWorker(
    pipeline,
    observers=[observer],
)

Filtering Processor Types

To measure only specific processor types, pass a processor_types tuple:

from pipecat.services.stt_service import STTService
from pipecat.services.tts_service import TTSService

observer = StartupTimingObserver(
    processor_types=(STTService, TTSService)
)

Configuration

Tuple[Type[FrameProcessor], ...] | None

default:"None"

Optional tuple of processor types to measure. If None, all non-internal processors are measured. Internal pipeline processors (PipelineSource, Pipeline) are always excluded.

Event Handlers

on_startup_timing_report

Called once after the pipeline has fully started, with timing data for all measured processors.

@observer.event_handler("on_startup_timing_report")
async def on_startup_timing_report(observer, report):
    # report is a StartupTimingReport
    print(f"Total: {report.total_duration_secs:.3f}s")
    for timing in report.processor_timings:
        print(f"  {timing.processor_name}: {timing.duration_secs:.3f}s")

Report fields (StartupTimingReport):

Field	Type	Description
`start_time`	`float`	Unix timestamp when the pipeline began setting up
`total_duration_secs`	`float`	Wall-clock span from the pipeline starting to set up until it had started. Processors set up concurrently, so this is the span, not the sum of what each cost
`processor_timings`	`List[ProcessorStartupTiming]`	Per-processor timing data, in pipeline order

Processor timing fields (ProcessorStartupTiming):

Field	Type	Description
`processor_name`	`str`	The name of the processor
`start_offset_secs`	`float`	Offset from the `StartFrame` to when this processor’s `start()` began
`duration_secs`	`float`	What the processor cost to get ready: its `setup()` and its `start()` together
`setup_duration_secs`	`float`	How long `setup()` took — the part of `duration_secs` spent connecting

on_transport_timing_report

Called once when the first client connects, with transport connection timing measured from the moment the pipeline began setting up.

@observer.event_handler("on_transport_timing_report")
async def on_transport_timing_report(observer, report):
    # report is a TransportTimingReport
    if report.bot_connected_secs is not None:
        print(f"Bot connected: {report.bot_connected_secs:.3f}s")
    print(f"Client connected: {report.client_connected_secs:.3f}s")

Report fields (TransportTimingReport):

Field	Type	Description
`start_time`	`float`	Unix timestamp when the pipeline began setting up
`bot_connected_secs`	`Optional[float]`	Seconds from start of setup to `BotConnectedFrame` (SFU transports only)
`client_connected_secs`	`Optional[float]`	Seconds from start of setup to first `ClientConnectedFrame`

bot_connected_secs is only set for SFU transports (Daily, LiveKit, HeyGen, Tavus) that emit a BotConnectedFrame when the bot joins the room. Non-SFU transports (WebSocket, SmallWebRTC) will have this field set to None.

Get Started

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Pipecat Context Hub

Startup Timing Observer

Features

Usage

Basic Startup Monitoring

Filtering Processor Types

Configuration

Event Handlers

on_startup_timing_report

on_transport_timing_report

​Features

​Usage

​Basic Startup Monitoring

​Filtering Processor Types

​Configuration

​Event Handlers

​on_startup_timing_report

​on_transport_timing_report

Features

Usage

Basic Startup Monitoring

Filtering Processor Types

Configuration

Event Handlers

on_startup_timing_report

on_transport_timing_report