> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://respan.ai/docs/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://respan.ai/docs/_mcp/server.

# Pipecat (tracing)

> Trace Pipecat real-time voice and multimodal AI pipelines with Respan — auto-instrumented spans, gateway routing, and full observability.

[Pipecat](https://www.pipecat.ai/) 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](https://platform.respan.ai) and grab an [API key](https://platform.respan.ai/platform/gateway/api-keys).

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

#### Use Respan Gateway

See [Pipecat gateway setup](/docs/gateway/pipecat) to route this integration through the Respan gateway.

#### Example projects

* [Python examples](https://github.com/respanai/respan-example-projects/tree/main/python/tracing/pipecat)

## Setup

#### Install packages

```bash
pip install respan-ai respan-instrumentation-pipecat
```

#### Set environment variables

```bash
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

```python
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](https://platform.respan.ai/platform/traces) 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.

```python
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.

```python
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.

```python
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)
```