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. Startup also loads the framework’s deferred imports, which no processor accounts for, so the report carries what that cost alongside the per-processor timings and the whole span is attributable.

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")
    if report.warmup:
        print(f"  warmup: {report.warmup.blocking_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 time from the pipeline starting to set up until it had started
`setup_phase_secs`	`float`	How long getting every processor ready took. Setup runs concurrently and warming runs alongside it
`start_phase_secs`	`float`	How long the `StartFrame` took to travel the pipeline. It reaches processors one after another
`processor_timings`	`List[ProcessorStartupTiming]`	Per-processor timing data, in pipeline order
`warmup`	`StartupWarmupTiming \| None`	What warming the framework’s deferred imports cost, or `None` when the pipeline started without warming

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. Processors are set up concurrently, so this overlaps every other processor’s
`start_duration_secs`	`float`	How long the processor spent on the `StartFrame`. The frame reaches processors one after another, so this adds up

Warmup timing fields (StartupWarmupTiming):

Field	Type	Description
`duration_secs`	`float`	How long warming took, in seconds
`blocking_duration_secs`	`float`	The part of warming the pipeline waited on once every processor had finished setting up. Warming runs alongside setup, so a pipeline whose processors take longer to connect than warming takes to load pays nothing for it

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