> ## Documentation Index
> Fetch the complete documentation index at: https://docs.usetuner.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Connecting to Pipecat
> Connect your Pipecat pipeline to Tuner with the Observer SDK. Call data is automatically captured and sent to Tuner when each call ends.
This guide covers the **Python SDK** for [pipecat-ai](https://github.com/pipecat-ai/pipecat). The package is available on PyPI. A **Node.js SDK** is coming soon.
## Video Tutorial
## Prerequisites
* Tuner Active Account
* Configured Agent with provider "Custom API" in Tuner
* Python 3.11–3.13 (3.14 is not yet supported due to pipecat dependencies)
* Python project running [pipecat-ai](https://github.com/pipecat-ai/pipecat) v1.0.0 or later
## Overview
The **tuner-pipecat-sdk** is a lightweight observer that captures flow transitions, latency signals, transcript segments, and usage metadata from your Pipecat pipeline, then sends a structured `CallPayload` to Tuner when the call ends — no manual API calls required. If your agent's LLM step is a LangChain runnable or LangGraph graph, the same observer also captures node transitions, tool calls, and token usage from that layer — see [LangChain / LangGraph observability](#langchain-%2F-langgraph-observability) below.
Add the package to your project with pip.
Configure your Tuner API key, workspace ID, and agent remote ID.
Create an observer instance and attach it to pipecat.
**Estimated time:** 2 minutes from start to finish
## Step 1: Install the SDK
For pipecat-ai pipelines:
```shell theme={null}
pip install tuner-pipecat-sdk
```
**Requirements:** Python 3.11–3.13, `pipecat-ai>=1.0.0`. Do not use Python 3.14 — pipecat dependencies (`onnxruntime`, `numba`) do not yet have 3.14 wheels.
## Step 2: Set Your Credentials
You can configure credentials via environment variables or pass them directly in code.
```shell theme={null}
export TUNER_API_KEY="tr_api_..."
export TUNER_WORKSPACE_ID="123"
export TUNER_AGENT_ID="fa4da74c-..."
```
| Variable | Required | Description |
| -------------------- | -------- | ----------------------------------------------------------------------------------------------------- |
| `TUNER_API_KEY` | ✅ | Bearer token (starts with `tr_api_`) |
| `TUNER_WORKSPACE_ID` | ✅ | Your Tuner workspace ID |
| `TUNER_AGENT_ID` | ✅ | **Agent Remote ID** from **Agent Settings → Agent Connection** (UUID — not your agent's display name) |
| `TUNER_BASE_URL` | — | API base URL (default: `https://api.usetuner.ai`) |
```python theme={null}
Observer(
api_key="tr_api_...",
workspace_id=123,
agent_id="fa4da74c-...",
call_id=str(uuid.uuid4()),
base_url="https://api.usetuner.ai",
)
```
Use the **Agent Remote ID** from **Agent Settings → Agent Connection** — not your agent's display name. It is a UUID (for example `fa4da74c-...`). Copy **your** value from the Tuner dashboard.
## Step 3: Add the Observer
Create an `Observer` instance and attach turn tracking:
```python theme={null}
import uuid
from tuner_pipecat_sdk import Observer
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
turn_tracker = TurnTrackingObserver()
observer = Observer(
api_key="YOUR_TUNER_API_KEY",
workspace_id=42,
agent_id="fa4da74c-...",
call_id=str(uuid.uuid4()),
base_url="https://api.usetuner.ai",
asr_model="deepgram/nova-3",
llm_model="gpt-4o-mini",
tts_model="cartesia/sonic",
)
observer.attach_turn_tracking_observer(turn_tracker)
```
Attach tuner observer, latency\_observer and turn\_tracker to the Pipeline:
```python theme={null}
from pipecat.pipeline.task import PipelineTask, PipelineParams
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
observers=[observer, observer.latency_observer, turn_tracker],
)
```
Without `enable_metrics` and `enable_usage_metrics`, the observer will log warnings and LLM/TTS metric fields will be absent from the payload.
For more examples, see the [tuner-pipecat-sdk-python examples](https://github.com/usetuner/tuner-pipecat-sdk-python/tree/main/examples).
## End-of-Utterance (EOU) Delay
Wire your user context aggregator to the observer to capture **how long the
pipeline took to decide the user was done talking** — the gap between the
user going silent and the turn actually closing. This appears on every user
turn in the transcript view as `eou_delay`, `eou_confidence`, and
`eou_reason`.
```python theme={null} theme={null}
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
)
observer.attach_user_aggregator(user_aggregator)
```
`eou_reason` tells you *which* signal closed the turn, so `eou_delay` means
something different depending on the value:
| `eou_reason` | What closed the turn | `eou_delay` measures |
| ----------------- | ------------------------------------------------------------- | ---------------------------------------------------------------------------------- |
| `silence_timeout` | Local VAD silence threshold | ms from the last VAD-detected speech-end to the turn closing |
| `model_verdict` | A turn-completion model (e.g. Smart Turn) decided | the model's own end-to-end decision time; `eou_confidence` carries its probability |
| `stt_endpoint` | Server-side STT (e.g. Deepgram Flux) decided on its own clock | not populated — the decision happens server-side, independent of local VAD |
All three fields are omitted from a turn's metadata when nothing decided
that turn's end.
## SIP / Telephony Calls
If you plan to run [Call Simulation](/docs/simulation/overview) against this agent, the observer needs the **SIP Call-ID** of the inbound call. Tuner uses it to match the simulation call it dialled with the call your observer syncs back — without it, simulated calls show up as ordinary production traffic.
This applies to **inbound** simulation, where Tuner dials your agent. For **outbound** simulation, pass the dialed SIP URI as `recipient` instead, see [Outbound Simulation](/docs/simulation/outbound).
The SDK handles SIP-field extraction internally. Pass the raw payload your server already has with **one line per provider**:
```python theme={null}
observer.attach_sip_from_telephony(payload, provider="twilio")
observer.attach_sip_from_telephony(payload, provider="telnyx")
observer.attach_sip_from_telephony(payload, provider="plivo")
observer.attach_sip_from_telephony(payload, provider="exotel")
observer.attach_sip_from_telephony(payload, provider="jambonz")
```
Built-in provider strings: `"twilio"`, `"telnyx"`, `"plivo"`, `"exotel"`, `"jambonz"`. For an unlisted provider, pass a callable extractor instead:
```python theme={null}
def my_extractor(payload):
return payload["my_id"], payload.get("my_headers")
observer.attach_sip_from_telephony(payload, provider=my_extractor)
```
For Daily PSTN/SIP dial-in:
```python theme={null}
observer.attach_sip_from_dialin(runner_args.body["dialin_settings"])
```
If you already have the values (custom SIP trunk, FreeSWITCH, Asterisk, etc.):
```python theme={null}
observer.attach_sip_info(sip_call_id="...", sip_headers={...})
```
For non-SIP (web) calls, skip all of the above — the observer falls back to its normal behaviour and SIP fields are omitted from the payload entirely.
## LangChain / LangGraph observability
Using LangGraph or LangChain for your LLM step? The observer can capture node
transitions, tool calls, and token usage from inside that runnable alongside the
standard call data. See [LangChain / LangGraph](/docs/integrations/custom-stack/langchain#pipecat).
## Configuration Options
Override the default call type label:
```python theme={null}
Observer(..., call_type="web_call")
Observer(..., call_type="phone_call")
```
The phone number or SIP URI of the callee. Optional — pass it when you want to record who the call was directed to:
```python theme={null}
Observer(..., recipient="+15551234567")
Observer(..., recipient="sip:+15551234567@provider.com")
```
Provide a recording URL if available. Default is `"pipecat://no-recording"`:
```python theme={null}
Observer(..., recording_url="https://cdn.example.com/recordings/call-123.ogg")
```
Record why a call ended by passing a `disconnection_reason_resolver` callable. The resolver is called at flush time and should return a string or `None`.
Use the built-in constants from `DisconnectReason`:
| Constant | Value |
| ------------------------------- | ---------------- |
| `DisconnectReason.USER_HANGUP` | `"user_hangup"` |
| `DisconnectReason.AGENT_HANGUP` | `"agent_hangup"` |
| `DisconnectReason.ERROR` | `"error"` |
| `DisconnectReason.TIMEOUT` | `"timeout"` |
| `DisconnectReason.UNKNOWN` | `"unknown"` |
```python theme={null}
from tuner_pipecat_sdk.models import DisconnectReason
_reason = None
def resolve_reason() -> str | None:
return _reason
observer = Observer(..., disconnection_reason_resolver=resolve_reason)
# Set the reason when your app knows it
_reason = DisconnectReason.USER_HANGUP
```
Specify your ASR, LLM, and TTS models for metadata:
```python theme={null}
Observer(
...,
asr_model="deepgram/nova-3",
llm_model="gpt-4o-mini",
tts_model="cartesia/sonic",
)
```
Track which version of your agent handled each call. Set `APP_VERSION` in your environment:
```shell theme={null}
APP_VERSION=42 python agent.py start
```
Using GitHub Actions or CircleCI? Tuner reads `GITHUB_RUN_NUMBER` / `CIRCLE_BUILD_NUM` automatically. Manual override via constructor takes priority:
```python theme={null}
Observer(..., agent_version=42)
```
Log the full transcript when flushing:
```python theme={null}
Observer(..., debug=True)
```
Provide a callable that receives the call's usage summary and returns the call cost in USD cents. Implement the method that matches your pricing plan; the snippet below is illustrative.
```python theme={null}
def calculate_cost(usage) -> float:
# OpenAI gpt-4o-mini pricing (per token / per character / per second)
llm_cost = (usage.llm_prompt_tokens or 0) * 0.000_000_150
llm_cost += (usage.llm_completion_tokens or 0) * 0.000_000_600
# OpenAI TTS-1: $15 per 1M characters
tts_cost = (usage.tts_characters or 0) * 0.000_015
# OpenAI gpt-4o-transcribe: $6 per 1M audio seconds
stt_cost = usage.stt_audio_seconds * 0.000_006
return (llm_cost + tts_cost + stt_cost) * 100 # USD -> cents
Observer(..., cost_calculator=calculate_cost)
```
## Full Example
```python theme={null}
import os
import uuid
from tuner_pipecat_sdk import Observer
from tuner_pipecat_sdk.models import DisconnectReason
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask, PipelineParams
async def run_bot(call_data, context, transport, stt, llm, tts, context_aggregator):
# call_data is the payload from your telephony WebSocket handler
# e.g. _, call_data = await parse_telephony_websocket(websocket)
# Read the SIP Call-ID from the provider payload — the field name varies
# Twilio example: forward it as a customParameter (see SIP section above)
sip_call_id = call_data.get("customParameters", {}).get("SipCallId")
turn_tracker = TurnTrackingObserver()
_reason = None
# update the rates to match your models
def calculate_cost(usage) -> float:
# OpenAI gpt-4o-mini pricing (per token / per character / per second)
llm_cost = (usage.llm_prompt_tokens or 0) * 0.000_000_150
llm_cost += (usage.llm_completion_tokens or 0) * 0.000_000_600
# OpenAI TTS-1: $15 per 1M characters
tts_cost = (usage.tts_characters or 0) * 0.000_015
# OpenAI gpt-4o-transcribe: $6 per 1M audio seconds
stt_cost = usage.stt_audio_seconds * 0.000_006
return (llm_cost + tts_cost + stt_cost) * 100
observer = Observer(
api_key=os.environ["TUNER_API_KEY"],
workspace_id=int(os.environ["TUNER_WORKSPACE_ID"]),
agent_id="fa4da74c-...",
call_id=str(uuid.uuid4()),
base_url="https://api.usetuner.ai",
call_type="phone_call",
asr_model="deepgram/nova-3",
llm_model="gpt-4o-mini",
tts_model="cartesia/sonic",
sip_call_id=sip_call_id,
disconnection_reason_resolver=lambda: _reason,
cost_calculator=calculate_cost,
)
observer.attach_turn_tracking_observer(turn_tracker)
pipeline = Pipeline([
transport.input(),
stt,
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
])
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
observers=[observer, observer.latency_observer, turn_tracker],
)
_reason = DisconnectReason.USER_HANGUP
```
Using LangChain or LangGraph for your LLM step instead of a native pipecat LLM
service? Swap `llm` for a `LangchainProcessor` wrapping `observer.wrap_graph(...)`
/ `observer.wrap_chain(...)` — see [LangChain / LangGraph](/docs/integrations/custom-stack/langchain#pipecat).
If you prefer not to extract the field yourself, call `observer.attach_sip_from_telephony(call_data, provider="twilio")` after creating the observer — the SDK reads the SIP Call-ID from the payload for you.
## Webhooks
Tuner can notify your systems when it has something to report on a call — see [Webhooks](/docs/integrations/webhooks).
## Troubleshooting
**Python 3.14**: Pipecat pins `onnxruntime` versions that have no 3.14 wheels. Switch to **Python 3.12 or 3.13** and create a new venv.
Same as above: use **Python 3.12 or 3.13**.
* Verify that `agent_id` is the **Agent Remote ID** from **Agent Settings → Agent Connection** (not your agent's display name).
* Confirm `workspace_id` is correct.
* Ensure `enable_metrics=True` and `enable_usage_metrics=True` are set on `PipelineParams`.
* Check your application logs for any error messages from the observer.
* Ensure `TUNER_API_KEY` starts with `tr_api_` and is valid.
* Confirm the API key belongs to the correct workspace.
The `langchain` extra isn't installed. Run `pip install tuner-pipecat-sdk[langchain]`.
## What's Next?
Set up call outcomes, user intents, and evals.
Learn about the underlying API if you need more control.
Define how Tuner categorizes your calls.
Get notified when issues are detected.