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 and 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 and 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",
)

Extra Metadata

Attach arbitrary key-value data to every call record. Useful for environment, region, deployment version, tenant, or anything you want to filter and evaluate against later:

    Observer(
        ...,
        extra_metadata={
            "env": "production",
            "region": "us-east-1",
            "deployment": "v2.3.1",
        },
    )

Metadata can also be used as an eval input — see Evaluate agents with per-call instructions.

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,
        extra_metadata={"env": "prod", "region": "us-east-1"},
    )

    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.

OpenTelemetry traces

Pipecat already emits OpenTelemetry spans for every conversation. Point its exporter at Tuner and those spans appear as a span tree on the matching call, under a Traces tab. See OpenTelemetry Traces for how to read one.

This is optional and separate from the observer above. Trace export is configured through Pipecat’s own tracing setup.

Point your exporter at Tuner

export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT="https://api.usetuner.ai/api/v1/traces"
export OTEL_EXPORTER_OTLP_HEADERS="authorization=Bearer tr_api_..."

Use the same TUNER_API_KEY you set in Step 2. It identifies your workspace, so nothing else needs configuring.

Use the signal-specific variable, OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, and give the full URL. The generic OTEL_EXPORTER_OTLP_ENDPOINT breaks this setup in two ways: the exporter appends /v1/traces to it, which does not resolve to Tuner’s path, and it is shared with metrics and logs, so repointing it would send those to Tuner as well.

If you already send traces to your own observability backend, leave that variable alone and fan out with an OpenTelemetry Collector instead of switching endpoints.

Switch to the HTTP exporter

Tuner accepts OTLP over HTTP with protobuf encoding, which is the default for the HTTP exporter. gRPC is not supported.

# Correct
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

# Not supported by Tuner
# from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

Pipecat’s own tracing example uses the gRPC exporter. If you started from that example, this import is the one line you need to change.

Tag the trace with your call id

Tuner matches a trace to a call on a tuner.call_id span attribute. Its value has to be the same call_id you pass to the Observer, so reuse the variable rather than generating a second UUID.

from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
from pipecat.utils.tracing.setup import setup_tracing

call_id = str(uuid.uuid4())

observer = Observer(
    api_key="YOUR_TUNER_API_KEY",
    workspace_id=42,
    agent_id="fa4da74c-...",
    call_id=call_id,
)

setup_tracing(service_name="my-agent", exporter=OTLPSpanExporter())

worker = PipelineWorker(
    pipeline,
    params=PipelineParams(enable_metrics=True),
    enable_tracing=True,
    additional_span_attributes={"tuner.call_id": call_id},
)

Pipecat applies additional_span_attributes to the root conversation span only. That is enough, because Tuner correlates at the trace level and applies the id to every span in the trace.

Pass service_name as well. Without it, spans arrive with service.name set to unknown_service, which makes traces from different agents hard to tell apart.

Check that it worked

Open the call in Tuner. A Traces tab appears on the call details page once spans have been received and matched. Correlation completes when the call itself is ingested, so the tab can take a moment to appear after a call ends.

Span reference

The span names come from Pipecat, so they are the same ones you would see in any other OpenTelemetry backend. Service spans nest under turns, under one conversation span:

conversation
└── turn
    ├── stt
    ├── llm
    └── tts

Realtime models such as Gemini Live and OpenAI Realtime replace the llm span with llm_setup, llm_response, and llm_tool_call or llm_tool_result. See Pipecat’s OpenTelemetry documentation for the full set, and OpenTelemetry Traces for how to read the tree in Tuner.

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

The Traces tab never appears

In order of likelihood:

tuner.call_id does not match the call_id you passed to the Observer. They have to be identical strings. This is the most common cause by a wide margin.
The attribute is missing entirely. Nothing in the trace carries it, so there is nothing for Tuner to match on.
The gRPC exporter is still in use. Requests are rejected with 415. Pipecat’s own example uses gRPC, so check that import first.
The API key is wrong or revoked. Requests are rejected with 401 or 403.

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 and LangGraph observability

Configuration Options

Full Example

OpenTelemetry traces

Span reference

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 and LangGraph observability

​Configuration Options

​Full Example

​OpenTelemetry traces

​Span reference

​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 and LangGraph observability

Configuration Options

Full Example

OpenTelemetry traces

Span reference

Webhooks

Troubleshooting

What’s Next?