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

1 import asyncio
2 import os
3 
4 from dotenv import load_dotenv
5 from pipecat.frames.frames import EndFrame, LLMContextFrame, LLMFullResponseEndFrame, LLMTextFrame
6 from pipecat.pipeline.pipeline import Pipeline
7 from pipecat.pipeline.runner import PipelineRunner
8 from pipecat.pipeline.task import PipelineTask
9 from pipecat.processors.aggregators.llm_context import LLMContext
10 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
11 from pipecat.services.openai.llm import OpenAILLMService
12 from respan import Respan
13 from respan_instrumentation_pipecat import PipecatInstrumentor
14 
15 load_dotenv()
16 
17 respan = Respan(instrumentations=[PipecatInstrumentor()])
18 
19 class TextCollector(FrameProcessor):
20     def __init__(self):
21         super().__init__(name="text_collector", enable_direct_mode=True)
22         self.text = []
23         self.done = asyncio.Event()
24 
25     async def process_frame(self, frame, direction: FrameDirection):
26         await super().process_frame(frame, direction)
27         if isinstance(frame, LLMTextFrame):
28             self.text.append(frame.text)
29         elif isinstance(frame, LLMFullResponseEndFrame):
30             self.done.set()
31         await self.push_frame(frame, direction)
32 
33 async def main():
34     collector = TextCollector()
35     llm = OpenAILLMService(
36         api_key=os.environ["OPENAI_API_KEY"],
37         settings=OpenAILLMService.Settings(model=os.getenv("OPENAI_MODEL", "gpt-4.1-nano")),
38     )
39 
40     pipeline = Pipeline([llm, collector])
41     task = PipelineTask(pipeline, cancel_on_idle_timeout=False, enable_rtvi=False)
42 
43     async def push_frames():
44         await asyncio.sleep(0.05)
45         context = LLMContext(
46             messages=[{"role": "user", "content": "Reply in one short sentence about Pipecat tracing."}]
47         )
48         await task.queue_frame(LLMContextFrame(context))
49         await asyncio.wait_for(collector.done.wait(), timeout=30)
50         await asyncio.sleep(0.2)
51         await task.queue_frame(EndFrame())
52 
53     await asyncio.gather(PipelineRunner(handle_sigint=False).run(task), push_frames())
54     print("".join(collector.text))
55 
56     await asyncio.sleep(1)
57     respan.flush()
58 
59 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.

1 from respan import Respan
2 from respan_instrumentation_pipecat import PipecatInstrumentor
3 
4 respan = Respan(
5     instrumentations=[PipecatInstrumentor()],
6     customer_identifier="user_123",
7     thread_identifier="session_abc_123",
8     metadata={"service": "voice-agent", "version": "1.0.0"},
9 )

With propagate_attributes

Override per-request using a context scope.

1 from respan import Respan, propagate_attributes
2 from respan_instrumentation_pipecat import PipecatInstrumentor
3 
4 respan = Respan(instrumentations=[PipecatInstrumentor()])
5 
6 async def handle_session(user_id: str):
7     with propagate_attributes(
8         customer_identifier=user_id,
9         thread_identifier="session_abc_123",
10         metadata={"plan": "pro"},
11     ):
12         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.

1 from respan import Respan, workflow, task
2 from respan_instrumentation_pipecat import PipecatInstrumentor
3 
4 respan = Respan(instrumentations=[PipecatInstrumentor()])
5 
6 @task(name="run_pipecat_task")
7 async def run_pipecat_task(task):
8     await runner.run(task)
9 
10 @workflow(name="voice_session")
11 async def voice_session(task):
12     await run_pipecat_task(task)
13     respan.flush()