Pipecat Tracing

Overview

Pipecat Tracing provides deep observability into your Pipecat agent’s performance by integrating the Cekura Python SDK directly into your agent code. This integration significantly enhances the information available in the Cekura platform for end-to-end visibility over agent execution. What you get:

Complete agent-side conversation transcripts with UTC timestamps
Tool/function calls with inputs and outputs
Dual-channel audio recording (agent + user) for production monitoring
Session metadata for correlation and debugging

Video Tutorial

Watch this video to see the Pipecat tracing integration in action:

Prerequisites

A Cekura account with an API key
A Pipecat agent project

Setup

Testing
Observability

Use this setup in your test agents while running simulation calls from the Cekura platform. This mode captures transcripts only — no audio recording.

Install the Cekura Python SDK

pip install cekura[pipecat]==1.2.0

Integrate the SDK in your Pipecat agent

Add the Cekura tracer to your Pipecat agent:

Initialize the tracer with your API key and agent ID
Track your pipeline to capture transcripts and tool calls
Register task handlers for automatic cleanup

import os
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask
from pipecat.pipeline.runner import PipelineRunner
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.audio.vad.silero import SileroVADAnalyzer

from cekura.pipecat import PipecatTracer

async def run_bot(transport, runner_args):
    # Build your pipeline with aggregators
    context = LLMContext(tools=tools)
    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
        context,
        user_params=LLMUserAggregatorParams(
            vad_analyzer=SileroVADAnalyzer(),
        ),
    )

    pipeline = Pipeline([
        transport.input(),
        stt,
        user_aggregator,
        llm,
        tts,
        transport.output(),
        assistant_aggregator,
    ])

    # (1) Initialize Cekura tracer
    cekura = PipecatTracer(
        api_key=os.getenv("CEKURA_API_KEY"),
        agent_id=123,  # Your agent ID from Cekura dashboard
    )

    # (2) Track pipeline (transcript only, no audio)
    pipeline = cekura.track_pipeline(pipeline, context, runner_args=runner_args)

    task = PipelineTask(pipeline)

    # (3) Register task handlers for automatic cleanup
    task = cekura.register_task_handlers(task, transport=transport)

    # Run pipeline as usual
    runner = PipelineRunner()
    await runner.run(task)

What this does:

Captures transcripts and tool calls
Sends data to Cekura’s simulation webhook for test analysis
No audio recording — lightweight for testing

For simpler setups, you can combine all steps into one call:

task = cekura.track_and_create_task(
    pipeline, context, runner_args=runner_args, transport=transport
)

Configure Pipecat provider and enable tracing

Navigate to your agent settings in the Cekura dashboard, select Pipecat as the provider, and enable tracing:Required configuration:

Provider: Select “Pipecat” as your voice integration provider
Pipecat Cloud API Key: Your authentication key from Pipecat Cloud
Pipecat Agent Name: Your agent name in Pipecat (e.g., “my-agent”)
Tracing Enabled: Set to true to enable SDK-based tracing

Testing connection types:Configure at least one connection type for voice-based testing:

WebRTC: Direct Pipecat session connection using the credentials configured above
Telephony: Phone-based testing if your Pipecat agent is connected to a phone system (requires Contact Number)

See the Pipecat Automated Testing guide for details on all configuration fields including Agent Configuration JSON and Room Properties.

Run tests

Run tests using your preferred connection type:

WebRTC: Select WebRTC under Voice connections in the Configure Run dialog for WebRTC-based testing
Telephony: Select Telephony under Voice connections for phone-based testing

Test results will appear in the Runs section with enhanced data including transcripts and tool calls.

Use this setup in your production agents for monitoring live calls with audio recording.

Install the Cekura Python SDK

pip install cekura[pipecat]==1.2.0

Integrate the SDK in your Pipecat agent

Add the Cekura tracer to your Pipecat agent:

Initialize the tracer with your API key and agent ID
Observe your pipeline to capture transcripts, tool calls, and audio
Register task handlers for audio recording and automatic cleanup

import os
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask
from pipecat.pipeline.runner import PipelineRunner
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.audio.vad.silero import SileroVADAnalyzer

from cekura.pipecat import PipecatTracer

async def run_bot(transport, runner_args):
    # Build your pipeline with aggregators
    context = LLMContext(tools=tools)
    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
        context,
        user_params=LLMUserAggregatorParams(
            vad_analyzer=SileroVADAnalyzer(),
        ),
    )

    pipeline = Pipeline([
        transport.input(),
        stt,
        user_aggregator,
        llm,
        tts,
        transport.output(),
        assistant_aggregator,
    ])

    # (1) Initialize Cekura tracer
    cekura = PipecatTracer(
        api_key=os.getenv("CEKURA_API_KEY"),
        agent_id=123,  # Your agent ID from Cekura dashboard
    )

    # (2) Observe pipeline (transcript + audio recording)
    pipeline = cekura.observe_pipeline(pipeline, context, runner_args=runner_args)

    task = PipelineTask(pipeline)

    # (3) Register task handlers for audio recording
    task = cekura.register_task_handlers(task, transport=transport)

    # Run pipeline as usual
    runner = PipelineRunner()
    await runner.run(task)

What this does:

Captures transcripts, tool calls, and session metadata
Records dual-channel audio (agent + user) for analysis
Exports data to Cekura observability endpoint

For simpler setups, you can combine all steps into one call:

task = cekura.observe_and_create_task(
    pipeline, context, runner_args=runner_args, transport=transport
)

Monitor production calls

Once the SDK is configured in your agent code, all production calls will automatically appear in the Calls section of your Cekura dashboard with enhanced data. No additional dashboard configuration is required for observability mode.

Enhanced Data in Cekura UI

With tracing enabled, you’ll see enriched information in the Cekura platform:

Complete Transcript: Full conversation history with UTC timestamps
Tool Calls: Function call requests and responses with inputs and outputs
Session Metadata: Session IDs for correlating Cekura calls with your logs
Audio Recording: Dual-channel recordings (agent + user) for quality monitoring (observability mode only)
Custom Metadata: Additional metadata passed via custom_metadata parameter

Custom Metadata

Pass custom metadata at initialization or update it anytime before the session tears down:

# At initialization
task = cekura.observe_and_create_task(
    pipeline, context, runner_args=runner_args, transport=transport,
    custom_metadata={"bot_version": "1.0", "environment": "staging"}
)

# Or update anytime before the session ends (including post-processing handlers
# that run before the pipeline closes)
cekura.set_custom_metadata({"bot_version": "1.0", "environment": "staging"})

set_custom_metadata() can be called at any point before the pipeline session finalizes — including in cleanup or post-processing handlers that execute before the session tears down. Once the session has ended and data has been submitted to Cekura, the record is immutable and metadata cannot be updated.

Best Practices

Use the right method for your environment: Use track_pipeline() / track_and_create_task() in your test/UAT environments for simulation testing. Use observe_pipeline() / observe_and_create_task() in your production environment for monitoring live calls with audio recording.
Use environment variables for credentials: Don’t hardcode API keys in your code
Keep the SDK updated: Run pip install --upgrade cekura periodically for the latest features
Use session IDs: The SDK resolves the session ID in this order: explicit session_id parameter > runner_args.session_id > auto-generated. Pass a custom session ID to correlate Cekura data with your application logs

SDK Reference

PipecatTracer Initialization

from cekura.pipecat import PipecatTracer

tracer = PipecatTracer(
    api_key="your_api_key",         # Required: Your Cekura API key
    agent_id=123,                   # Required: Agent ID from dashboard
    host="https://api.cekura.ai",   # Optional: Custom API host
    enabled=True                    # Optional: Enable/disable tracer
)

observe_pipeline()

Adds observability to your pipeline — captures transcripts and records audio. For production monitoring.

pipeline = tracer.observe_pipeline(
    pipeline,              # Required: Pipecat Pipeline instance
    context,               # Required: LLMContext instance
    runner_args=None,      # Optional: RunnerArguments (session_id extracted if present)
    session_id=None,       # Optional: Takes precedence over runner_args.session_id
    custom_metadata=None   # Optional: Dict of custom metadata
)

track_pipeline()

Adds simulation-mode tracking to your pipeline — captures transcripts only, no audio. For testing.

pipeline = tracer.track_pipeline(
    pipeline,              # Required: Pipecat Pipeline instance
    context,               # Required: LLMContext instance
    runner_args=None,      # Optional: RunnerArguments (session_id extracted if present)
    session_id=None,       # Optional: Takes precedence over runner_args.session_id
    custom_metadata=None   # Optional: Dict of custom metadata
)

observe_and_create_task()

Single-step observability setup. Combines observe_pipeline(), PipelineTask creation, and register_task_handlers().

task = tracer.observe_and_create_task(
    pipeline,              # Required: Pipecat Pipeline instance
    context,               # Required: LLMContext instance
    runner_args=None,      # Optional: RunnerArguments for session_id
    transport=None,        # Optional: Transport for cleanup on disconnect
    session_id=None,       # Optional: Custom session identifier
    custom_metadata=None   # Optional: Dict of custom metadata
)

track_and_create_task()

Single-step simulation setup. Combines track_pipeline(), PipelineTask creation, and register_task_handlers().

task = tracer.track_and_create_task(
    pipeline,              # Required: Pipecat Pipeline instance
    context,               # Required: LLMContext instance
    runner_args=None,      # Optional: RunnerArguments for session_id
    transport=None,        # Optional: Transport for cleanup on disconnect
    session_id=None,       # Optional: Custom session identifier
    custom_metadata=None   # Optional: Dict of custom metadata
)

register_task_handlers()

Registers cleanup handlers for automatic resource management when the session ends. Only needed when using the multi-step approach.

task = tracer.register_task_handlers(
    task,              # Required: PipelineTask instance
    transport=None     # Optional: Transport for cleanup on disconnect
)

set_custom_metadata()

Update custom metadata at any point during the session.

tracer.set_custom_metadata({"key": "value"})

get_custom_metadata()

Retrieve the current custom metadata.

metadata = tracer.get_custom_metadata()

Environment variables:

CEKURA_TRACER_ENABLED="false": Disable the tracer entirely

Troubleshooting

Missing Aggregators Warning

If you see:

Cekura observability disabled: LLMUserAggregator and LLMAssistantAggregator not found in pipeline.

Your pipeline must include LLMUserAggregator and LLMAssistantAggregator for transcript capture. Add aggregators to your pipeline:

from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.audio.vad.silero import SileroVADAnalyzer

user_agg, assistant_agg = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        vad_analyzer=SileroVADAnalyzer()
    )
)

pipeline = Pipeline([
    transport.input(),
    stt,
    user_agg,      # Add user aggregator
    llm,
    tts,
    transport.output(),
    assistant_agg,  # Add assistant aggregator
])

Import Error

If you see:

ImportError: PipecatTracer requires additional dependencies. Install with: pip install cekura[pipecat]

Install the pipecat dependencies:

pip install cekura[pipecat]

Next Steps

Create custom metrics to evaluate your agents
Set up automated testing for your Pipecat agents
Explore predefined metrics

Get Started

Key Concepts

Guides

Integrations

Advanced

Pipecat Tracing

Overview

Video Tutorial

Prerequisites

Setup

Enhanced Data in Cekura UI

Custom Metadata

Best Practices

SDK Reference

PipecatTracer Initialization

observe_pipeline()

track_pipeline()

observe_and_create_task()

track_and_create_task()

register_task_handlers()

set_custom_metadata()

get_custom_metadata()

Troubleshooting

Missing Aggregators Warning

Import Error

Next Steps

Get Started

Key Concepts

Guides

Integrations

Advanced

Documentation Index

​Overview

​Video Tutorial

​Prerequisites

​Setup

​Enhanced Data in Cekura UI

​Custom Metadata

​Best Practices

​SDK Reference

​PipecatTracer Initialization

​observe_pipeline()

​track_pipeline()

​observe_and_create_task()

​track_and_create_task()

​register_task_handlers()

​set_custom_metadata()

​get_custom_metadata()

​Troubleshooting

​Missing Aggregators Warning

​Import Error

​Next Steps

Overview

Video Tutorial

Prerequisites

Setup

Enhanced Data in Cekura UI

Custom Metadata

Best Practices

SDK Reference

PipecatTracer Initialization

observe_pipeline()

track_pipeline()

observe_and_create_task()

track_and_create_task()

register_task_handlers()

set_custom_metadata()

get_custom_metadata()

Troubleshooting

Missing Aggregators Warning

Import Error

Next Steps