Pipecat (tracing) | Respan Docs

Pipecat is an open-source framework for building real-time, multimodal AI applications. It provides a pipeline architecture for voice agents, video processing, and other real-time experiences with support for speech-to-text, LLMs, and text-to-speech. Respan gives you full observability over every Pipecat turn, LLM call, STT step, TTS step, and tool execution — and gateway routing through the OpenAI-compatible Respan endpoint.

Set up Respan

Create an account at platform.respan.ai and grab an API key.

Run npx @respan/cli setup to set up with your coding agent.

Use Respan Gateway

See Pipecat gateway setup to route this integration through the Respan gateway.

Example projects

Python examples

Setup

Install packages

pip install respan-ai respan-instrumentation-pipecat

Set environment variables

export OPENAI_API_KEY="YOUR_OPENAI_API_KEY"
export RESPAN_API_KEY="YOUR_RESPAN_API_KEY"

OPENAI_API_KEY is used by Pipecat’s OpenAI service. RESPAN_API_KEY exports traces to Respan. If you use different Pipecat services, set their provider keys as usual.

Initialize and run

import asyncio
import os
from dotenv import load_dotenv
from pipecat.frames.frames import EndFrame, LLMContextFrame, LLMFullResponseEndFrame, LLMTextFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.openai.llm import OpenAILLMService
from respan import Respan
from respan_instrumentation_pipecat import PipecatInstrumentor
load_dotenv()
respan = Respan(instrumentations=[PipecatInstrumentor()])
class TextCollector(FrameProcessor):
    def __init__(self):
        super().__init__(name="text_collector", enable_direct_mode=True)
        self.text = []
        self.done = asyncio.Event()
    async def process_frame(self, frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, LLMTextFrame):
            self.text.append(frame.text)
        elif isinstance(frame, LLMFullResponseEndFrame):
            self.done.set()
        await self.push_frame(frame, direction)
async def main():
    collector = TextCollector()
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        settings=OpenAILLMService.Settings(model=os.getenv("OPENAI_MODEL", "gpt-4.1-nano")),
    )
    pipeline = Pipeline([llm, collector])
    task = PipelineTask(pipeline, cancel_on_idle_timeout=False, enable_rtvi=False)
    async def push_frames():
        await asyncio.sleep(0.05)
        context = LLMContext(
            messages=[{"role": "user", "content": "Reply in one short sentence about Pipecat tracing."}]
        )
        await task.queue_frame(LLMContextFrame(context))
        await asyncio.wait_for(collector.done.wait(), timeout=30)
        await asyncio.sleep(0.2)
        await task.queue_frame(EndFrame())
    await asyncio.gather(PipelineRunner(handle_sigint=False).run(task), push_frames())
    print("".join(collector.text))
    await asyncio.sleep(1)
asyncio.run(main())

View your trace

Open the Traces page to see your Pipecat turn with child LLM, STT, TTS, and tool spans.

Configuration

Parameter	Type	Default	Description
`api_key`	`str \| None`	`None`	Falls back to `RESPAN_API_KEY` env var.
`base_url`	`str \| None`	`None`	Falls back to `RESPAN_BASE_URL` env var.
`instrumentations`	`list`	`[]`	Plugin instrumentations to activate (e.g. `PipecatInstrumentor()`).
`customer_identifier`	`str \| None`	`None`	Default customer identifier for all spans.
`thread_identifier`	`str \| None`	`None`	Default thread identifier for all spans.
`metadata`	`dict \| None`	`None`	Default metadata attached to all spans.
`environment`	`str \| None`	`None`	Environment tag (e.g. `"production"`).

PipecatInstrumentor forwards keyword arguments to the upstream OpenInference Pipecat instrumentor, including options such as debug_log_filename and config.

Attributes

In Respan()

Set defaults at initialization — these apply to all spans.

from respan import Respan
from respan_instrumentation_pipecat import PipecatInstrumentor
respan = Respan(
    instrumentations=[PipecatInstrumentor()],
    customer_identifier="user_123",
    thread_identifier="session_abc_123",
    metadata={"service": "voice-agent", "version": "1.0.0"},
)

With propagate_attributes

Override per-request using a context scope.

from respan import Respan, propagate_attributes
from respan_instrumentation_pipecat import PipecatInstrumentor
respan = Respan(instrumentations=[PipecatInstrumentor()])
async def handle_session(user_id: str):
    with propagate_attributes(
        customer_identifier=user_id,
        thread_identifier="session_abc_123",
        metadata={"plan": "pro"},
    ):
        await runner.run(task)

Attribute	Type	Description
`customer_identifier`	`str`	Identifies the end user in Respan analytics.
`thread_identifier`	`str`	Groups related messages into a conversation.
`metadata`	`dict`	Custom key-value pairs. Merged with default metadata.

Decorators (optional)

Decorators are not required. Pipecat turns, LLM calls, STT, TTS, and tools are auto-traced by the instrumentor. Use @workflow and @task to add application-level structure around a Pipecat session.

from respan import Respan, workflow, task
from respan_instrumentation_pipecat import PipecatInstrumentor
respan = Respan(instrumentations=[PipecatInstrumentor()])
@task(name="run_pipecat_task")
async def run_pipecat_task(task):
    await runner.run(task)
@workflow(name="voice_session")
async def voice_session(task):
    await run_pipecat_task(task)