Connecting to Pipecat

This guide covers the Python SDK for pipecat-ai. 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 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 below.

Install the SDK

Add the package to your project with pip.

Set Your Credentials

Configure your Tuner API key, workspace ID, and agent remote ID.

Add the Observer

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:

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.

Environment Variables
Inline in Code

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

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:

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:

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.

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.

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

The SDK handles SIP-field extraction internally. Pass the raw payload your server already has with one line per provider:

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:

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:

observer.attach_sip_from_dialin(runner_args.body["dialin_settings"])

If you already have the values (custom SIP trunk, FreeSWITCH, Asterisk, etc.):

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.

Configuration Options

Call Type

Override the default call type label:

Observer(..., call_type="web_call")
Observer(..., call_type="phone_call")

Recipient

The phone number or SIP URI of the callee. Optional — pass it when you want to record who the call was directed to:

Observer(..., recipient="+15551234567")
Observer(..., recipient="sip:+15551234567@provider.com")

Recording URL

Provide a recording URL if available. Default is "pipecat://no-recording":

Observer(..., recording_url="https://cdn.example.com/recordings/call-123.ogg")

Disconnection Reason

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

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

Model Names

Specify your ASR, LLM, and TTS models for metadata:

Observer(
    ...,
    asr_model="deepgram/nova-3",
    llm_model="gpt-4o-mini",
    tts_model="cartesia/sonic",
)

Agent Version

Track which version of your agent handled each call. Set APP_VERSION in your environment:

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:

Observer(..., agent_version=42)

Debug Mode

Log the full transcript when flushing:

Observer(..., debug=True)

Cost Calculation

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.

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

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.

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.

Troubleshooting

No matching distribution found for onnxruntime

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.

Failed to build numba / Cannot install on Python 3.14

Same as above: use Python 3.12 or 3.13.

Calls don't appear in Tuner

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.

Unauthorized (401)

Ensure TUNER_API_KEY starts with tr_api_ and is valid.
Confirm the API key belongs to the correct workspace.

wrap_graph / wrap_chain raises ImportError

The langchain extra isn’t installed. Run pip install tuner-pipecat-sdk[langchain].

What’s Next?

Configuring Your Agent

Set up call outcomes, user intents, and evals.

Custom Integration

Learn about the underlying API if you need more control.

Classifying Calls

Define how Tuner categorizes your calls.

Real-Time Alerts

Get notified when issues are detected.

Getting Started

Integrations

Evaluation

Observability

Simulation

MCP

Guides

Resources

Video Tutorial

Prerequisites

Overview

Step 1: Install the SDK

Step 2: Set Your Credentials

Step 3: Add the Observer

End-of-Utterance (EOU) Delay

SIP / Telephony Calls

LangChain / LangGraph observability

Configuration Options

Full Example

Webhooks

Troubleshooting

What’s Next?

Configuring Your Agent

Custom Integration

Classifying Calls

Real-Time Alerts

​Video Tutorial

​Prerequisites

​Overview

​Step 1: Install the SDK

​Step 2: Set Your Credentials

​Step 3: Add the Observer

​End-of-Utterance (EOU) Delay

​SIP / Telephony Calls

​LangChain / LangGraph observability

​Configuration Options

​Full Example

​Webhooks

​Troubleshooting

​What’s Next?

Configuring Your Agent

Custom Integration

Classifying Calls

Real-Time Alerts

Video Tutorial

Prerequisites

Overview

Step 1: Install the SDK

Step 2: Set Your Credentials

Step 3: Add the Observer

End-of-Utterance (EOU) Delay

SIP / Telephony Calls

LangChain / LangGraph observability

Configuration Options

Full Example

Webhooks

Troubleshooting

What’s Next?