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

An active Tuner account
A Tuner agent with provider Pipecat
A Tuner API key, your workspace ID and the agent’s Agent Remote ID. See API keys.
Python 3.11 to 3.13 (3.14 is not yet supported due to Pipecat dependencies)
A Python project running pipecat-ai v1.0.0 or later

Overview

tuner-pipecat-sdk is an observer that captures latency signals, transcript segments, and usage metadata from your Pipecat pipeline, then sends a structured CallPayload to Tuner when the call ends. You don’t make any API calls yourself. 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.

Pass your credentials

Give the observer your Tuner API key, workspace ID, and Agent Remote ID.

Add the observer

Create an observer instance and register it on your pipeline task.

Enable simulation

Required only if you use Call Simulation. See Enable simulation.

Step 1: Install the SDK

pip install tuner-pipecat-sdk

Requirements: Python 3.11 to 3.13, pipecat-ai>=1.0.0. Do not use Python 3.14: Pipecat dependencies (onnxruntime, numba) do not yet have 3.14 wheels. Optional extras:

Extra	Install	Adds
`traces`	`pip install 'tuner-pipecat-sdk[traces]'`	OpenTelemetry trace forwarding. See OpenTelemetry traces.
`langchain`	`pip install 'tuner-pipecat-sdk[langchain]'`	`wrap_graph()` and `wrap_chain()`. See LangChain and LangGraph observability.

Step 2: Pass your credentials

The Observer does not read environment variables. Pass the credentials as constructor arguments. You can load them from your own environment variables, as below.

import os
import uuid
from tuner_pipecat_sdk import Observer

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()),
)

Parameter	Required	Description
`api_key`	✅	Tuner API key (starts with `tr_api_`)
`workspace_id`	✅	Your Tuner workspace ID (integer)
`agent_id`	✅	Agent Remote ID from Agent Settings > Agent Connection. This is a UUID, not your agent’s display name.
`call_id`	✅	A unique ID for this call. Use `uuid4()`.
`base_url`	No	API base URL (default: `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="tr_api_...",
    workspace_id=42,
    agent_id="fa4da74c-...",
    call_id=str(uuid.uuid4()),
    asr_model="deepgram/nova-3",
    llm_model="gpt-4o-mini",
    tts_model="cartesia/sonic",
)

observer.attach_turn_tracking_observer(turn_tracker)

Observer is a pipeline-level observer. Register it, its latency_observer and the turn tracker in observers=[...] on the PipelineTask, not as a processor inside 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 logs warnings and LLM/TTS metric fields are absent from the payload.

From pipecat-ai 1.3, PipelineWorker replaces PipelineTask. The examples on this page use PipelineTask.

When Pipecat emits EndFrame or CancelFrame, the observer sends the call to Tuner in the background. If the request fails, the SDK logs the failure and never raises or blocks your pipeline. If a call doesn’t show up in Tuner, check your logs for [tuner] failed to deliver call. For more examples, see the tuner-pipecat-sdk-python examples.

Enable simulation

Required for Call Simulation. Skip it otherwise.

Inbound

Tuner calls your agent over SIP.

SDK	Version
Python, `tuner-pipecat-sdk`	0.2.9 or later

Your carrier sends each call to a webhook, then opens a WebSocket for the audio. Tuner’s SIP data reaches the webhook but not the WebSocket. Forward it from the webhook, then attach it in your bot.

Add your SIP URI in Tuner

Add your agent’s SIP URI in Agent Settings > Simulation Setup. Get it from your SIP provider, then see Set up inbound simulation.

Forward the SIP data from your webhook

Read the SIP data from the carrier’s request and put it on the WebSocket.

Twilio
Telnyx

from tuner_pipecat_sdk import extract_sip, forward_to_websocket

@app.post("/twiml")
async def twiml(request):
    info = extract_sip(await request.form(), provider="twilio")
    fwd = forward_to_websocket(info, ws_url="wss://your-host/ws")
    return Response(content=fwd.twiml, media_type="application/xml")

Set the Voice URL of the Twilio SIP Domain that Tuner dials to this route. Pipecat’s built-in dev runner returns TwiML with no <Parameter> tags, so use your own webhook.

info = extract_sip(webhook_json, provider="telnyx")   # call.initiated
fwd = forward_to_websocket(info, ws_url="wss://your-host/ws")

# answer rejects custom_parameters, so answer first, then start the stream
await telnyx_post(f"/v2/calls/{call_control_id}/actions/answer", {})
await telnyx_post(f"/v2/calls/{call_control_id}/actions/streaming_start", fwd.stream)

Tuner dials the SIP subdomain of a Call Control application. Set the application’s webhook to this handler.

Attach the SIP data in your bot

In your WebSocket handler, read the SIP data from the first message and attach it to the observer. Do this before the call ends.

start = ...  # the WebSocket "start" message
observer.attach_sip(extract_sip(start, provider="twilio"))  # or "telnyx"

On Twilio, the call_data from Pipecat’s parse_telephony_websocket also works. On Telnyx, pass the raw start message, because parse_telephony_websocket drops its custom_parameters.

Twilio forwards SipCallId and X-Correlation-Id. Telnyx’s call.initiated event has no SIP Call-ID, so X-Correlation-Id links a Telnyx call. Tuner links the call on either one. You can call attach_sip more than once. Headers merge, and the first non-empty value wins.

Plivo, Exotel, Jambonz, Daily or your own SIP trunk

The SDK reads the SIP data for these carriers but doesn’t build the forwarding. You configure the carrier yourself, then attach in your bot.

Carrier	Configure	Attach
Plivo	Add `extraHeaders="SipCallId={{sip_call_id}}"` to `<Stream>`	`observer.attach_sip(extract_sip(call_data, provider="plivo"))`
Exotel	Add `custom_parameters: SipCallId=<%CallSid%>` in the Voicebot applet	`observer.attach_sip(extract_sip(call_data, provider="exotel"))`
Jambonz	SIP data arrives on the call-hook webhook, not the WebSocket. `JambonzPendingStore` holds it until the WebSocket opens. See `jambonz_server.py`.	`observer.attach_sip(extract_sip(call_hook_json, provider="jambonz"))`
Daily dial-in	Nothing	`observer.attach_sip(extract_sip(runner_args.body["dialin_settings"], provider="daily"))`
Your own SIP trunk	Nothing	`observer.attach_sip(sip_call_id=..., sip_headers={"X-Correlation-Id": ...})`

Web calls have no SIP data. Skip this section for them.

How the SIP data reaches your bot

Support another carrier

Register an object with a name and extract(payload) -> SipInfo. Add forward_to_websocket(info, ws_url) -> Forward to support forwarding.

from tuner_pipecat_sdk import Forward, SipInfo, register_provider

class AcmeProvider:
    name = "acme"

    def extract(self, payload) -> SipInfo:
        return SipInfo(sip_call_id=payload["sip_call_id"], sip_headers=payload["headers"])

    def forward_to_websocket(self, info: SipInfo, ws_url: str) -> Forward: ...

register_provider(AcmeProvider())

Older SDK versions

These methods still work in 0.2.9 and warn once. Use the replacement.

Deprecated	Use
`attach_sip_from_telephony(payload, provider=...)`	`attach_sip(extract_sip(payload, provider=...))`
`attach_sip_from_context(ctx)`	`attach_sip(extract_sip(payload, provider=...))`
`attach_sip_from_dialin(settings)`	`attach_sip(extract_sip(settings, provider="daily"))`
`attach_sip_info(sip_call_id=..., sip_headers=...)`	`attach_sip(sip_call_id=..., sip_headers=...)`
`build_sip_forwarding_twiml(form, ws_url=...)`	`forward_to_websocket(extract_sip(form, provider="twilio"), ws_url=...).twiml`

Outbound

Your agent calls a Simulated Caller at a SIP URI that Tuner sends to your trigger endpoint.

Configure the trigger endpoint in Agent Settings > Simulation Setup. See Set up outbound simulation.
Report the SIP URI your agent dialed as recipient, exactly as Tuner sent it. Reading it from the stream needs tuner-pipecat-sdk 0.2.9 or later.

If your bot gets audio from a Twilio or Telnyx media stream, put the SIP URI on the stream when you place the call. Then the bot reads it from the start message.

Twilio
Telnyx

from tuner_pipecat_sdk import SipInfo, extract_sip, forward_to_websocket

fwd = forward_to_websocket(SipInfo(recipient=dialed_sip_uri), ws_url="wss://your-host/ws", provider="twilio")
twilio_client.calls.create(to=dialed_sip_uri, from_=..., twiml=fwd.twiml)

# in the bot, on the WebSocket "start" message
observer.attach_sip(extract_sip(start, provider="twilio"))

from tuner_pipecat_sdk import SipInfo, extract_sip, forward_to_websocket

fwd = forward_to_websocket(SipInfo(recipient=dialed_sip_uri), ws_url="wss://your-host/ws", provider="telnyx")
# on call.answered for the call you placed
await telnyx_post(f"/v2/calls/{call_control_id}/actions/streaming_start", fwd.stream)

# in the bot, on the WebSocket "start" message
observer.attach_sip(extract_sip(start, provider="telnyx"))

If your bot already has the SIP URI, pass it directly:

Observer(
    ...,
    recipient=dialed_sip_uri,   # the SIP URI Tuner sent, unchanged
)

Check simulation works

Start a simulation run. See Run a simulation. When the call ends, check its result in the run. If it shows Unreported, Tuner never received a matching call from your agent. See Fix an Unreported run.

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

LangChain and LangGraph observability

If you use 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. Install the langchain extra and see LangChain / LangGraph.

Configuration options

Call type

Override the default call type label (default: "web_call"):

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

Recording URL

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

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

The URL must be a public https:// link. Tuner stores the default or any other value as sent, but treats the call as having no recording: it has no playback in call details and can’t be added to a Dataset.

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 or pass a custom string:

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. They are recorded as call 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, such as environment, region, deployment version, or tenant:

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

If APP_VERSION is not set, the observer reads GITHUB_RUN_NUMBER, CIRCLE_BUILD_NUM or BUILD_NUMBER. The value must be an integer. The constructor argument takes priority over all of them:

Observer(..., agent_version=42)

Trace forwarding

With the traces extra installed, the observer forwards Pipecat’s OpenTelemetry spans to Tuner. forward_traces defaults to True. To turn it off:

Observer(..., forward_traces=False)

See OpenTelemetry traces.

Debug mode

Log the full transcript when flushing:

Observer(..., debug=True)

Cost calculation

Provide a callable that receives the call’s usage summary (CallUsage) 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)

stt_audio_seconds is the call’s wall-clock duration in seconds, used as a proxy for STT duration.

Full example

import os
import uuid
from tuner_pipecat_sdk import Observer, extract_sip
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)

    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()),
        call_type="phone_call",
        asr_model="deepgram/nova-3",
        llm_model="gpt-4o-mini",
        tts_model="cartesia/sonic",
        disconnection_reason_resolver=lambda: _reason,
        cost_calculator=calculate_cost,
        extra_metadata={"env": "prod", "region": "us-east-1"},
    )

    # Inbound simulation: attach the SIP data (see Enable simulation)
    observer.attach_sip(extract_sip(call_data, provider="twilio"))
    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(...) or observer.wrap_chain(...). See LangChain / LangGraph.

OpenTelemetry traces

Pipecat emits OpenTelemetry spans for every conversation when tracing is on. The SDK forwards them to Tuner, and they appear as a span tree on the call’s Traces tab. See Traces for how to read one.

Requires tuner-pipecat-sdk 0.2.8 or later.

Install the tracing extras

pip install 'pipecat-ai[tracing]'          # lets Pipecat emit spans
pip install 'tuner-pipecat-sdk[traces]'    # lets the SDK forward them to Tuner

Without the traces extra, trace forwarding is skipped with a debug log, and the rest of the observer works as usual.

Turn on tracing in Pipecat

Set enable_tracing=True on your PipelineTask. This is Pipecat’s own switch for emitting spans. The SDK can’t set it for you, because you create the task after the observer.

task = PipelineTask(
    pipeline,
    params=PipelineParams(enable_metrics=True, enable_usage_metrics=True),
    observers=[observer, observer.latency_observer, turn_tracker],
    enable_tracing=True,        # required
    enable_turn_tracking=True,  # recommended: gives the trace its turn structure
)

Keep forward_traces on

forward_traces defaults to True on the Observer. The SDK sends spans to Tuner with the API key and base URL you passed to the Observer, and tags every span with your call_id. You don’t need setup_tracing(...), a TracerProvider, an exporter, or any OTEL_* environment variable.If you already call Pipecat’s setup_tracing(...) to export traces elsewhere, the SDK adds Tuner as an extra destination. If you configure an exporter yourself, use the OTLP HTTP exporter: Tuner does not accept gRPC.To stop forwarding traces, pass forward_traces=False. Pipecat’s enable_tracing is a separate switch that controls whether Pipecat produces spans at all.

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

Before you finish

A test call appears in Call Logs.
Your API key, workspace ID and Agent Remote ID come from API keys.
If you use Call Simulation: Enable simulation is done for your agent’s call direction, and one simulated call shows a result other than Unreported.

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 [tuner] failed to deliver call.

Unauthorized (401)

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

Simulated calls show as Unreported

sip_call_id is the Twilio CA… CallSid: your TwiML had no SipCallId parameter. Return forward_to_websocket(...).twiml from your webhook.
No X-Correlation-Id on a Telnyx call: the stream started without custom_parameters. Send fwd.stream as the streaming_start body.
sip_call_id is null: attach_sip never ran. Call it with the WebSocket start message.
ValueError: Unknown provider: the name is misspelled. Use a built-in name or register_provider.
See Fix an Unreported run.

wrap_graph / wrap_chain raises ImportError

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

The Traces tab never appears

enable_tracing=True is missing on the PipelineTask. Pipecat emits no spans without it.
The traces extra isn’t installed. Install tuner-pipecat-sdk[traces] and pipecat-ai[tracing].
forward_traces=False is set on the Observer.
The API key is wrong or revoked. Tuner rejects the spans with 401 or 403.

Next steps

Configure your agent

Set up call outcomes, user intents, and evals.

Custom integration

Learn about the underlying API if you need more control.

Classify calls

Define how Tuner categorizes your calls.

Alerts

Get notified when issues are detected.

Getting Started

Integrations

Evaluation

Observability

Simulation

Datasets

MCP

Guides

Resources

Video tutorial

Prerequisites

Overview

Step 1: Install the SDK

Step 2: Pass your credentials

Step 3: Add the observer

Enable simulation

Inbound

Outbound

Check simulation works

End-of-utterance (EOU) delay

LangChain and LangGraph observability

Configuration options

Full example

OpenTelemetry traces

Span reference

Webhooks

Before you finish

Troubleshooting

Next steps

Configure your agent

Custom integration

Classify calls

Alerts

​Video tutorial

​Prerequisites

​Overview

​Step 1: Install the SDK

​Step 2: Pass your credentials

​Step 3: Add the observer

​Enable simulation

​Inbound

​Outbound

​Check simulation works

​End-of-utterance (EOU) delay

​LangChain and LangGraph observability

​Configuration options

​Full example

​OpenTelemetry traces

​Span reference

​Webhooks

​Before you finish

​Troubleshooting

​Next steps

Configure your agent

Custom integration

Classify calls

Alerts

Video tutorial

Prerequisites

Overview

Step 1: Install the SDK

Step 2: Pass your credentials

Step 3: Add the observer

Enable simulation

Inbound

Outbound

Check simulation works

End-of-utterance (EOU) delay

LangChain and LangGraph observability

Configuration options

Full example

OpenTelemetry traces

Span reference

Webhooks

Before you finish

Troubleshooting

Next steps