Skip to main content
This guide covers the Python SDK for pipecat-flows. The package is available on PyPI.

​
Prerequisites

  • Tuner Active Account
  • Configured Agent with provider “Custom API” in Tuner
  • Python 3.10–3.13 (3.14 is not yet supported due to pipecat dependencies)
  • Python project running pipecat-ai v0.0.105 or later with pipecat-flows

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

Install the SDK

Add the package to your project with pip.
2

Set Your Credentials

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

Add the Observer

Create the FlowsObserver, attach it to your flow manager, and place it in your pipeline after TTS.
Estimated time: 2 minutes from start to finish

​
Step 1: Install the SDK

pip install tuner-pipecat-sdk
Requirements: Python 3.10–3.13, pipecat-ai>=0.0.105. 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. 
export TUNER_API_KEY="tr_api_..."
export TUNER_WORKSPACE_ID="123"
export TUNER_AGENT_ID="my-agent"
VariableRequiredDescription
TUNER_API_KEY✅Bearer token (starts with tr_api_)
TUNER_WORKSPACE_ID✅Your Tuner workspace ID
TUNER_AGENT_ID✅Agent identifier from Agent Settings
TUNER_BASE_URL—API base URL (default: https://api.usetuner.ai)
The Agent ID must match the identifier configured in Tuner. Find it under Agent Settings > Agent Connection > Agent ID.

​
Step 3: Add the Observer

Create the FlowsObserver, attach it to your flow manager and turn tracker, then place it after TTS in your pipeline: 
import uuid
from tuner_pipecat_sdk import FlowsObserver
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver

# Create the observer
observer = FlowsObserver(
    api_key="YOUR_TUNER_API_KEY",
    workspace_id=42,
    agent_id="my-agent",
    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",
)

# Required: attach before running the pipeline
observer.attach_flow_manager(flow_manager)
observer.attach_turn_tracking_observer(turn_tracker)

# Place observer after TTS in your pipeline
Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts,
    observer,
    transport.output(),
    context_aggregator.assistant(),
])
Enable metrics on the pipeline task so latency and usage fields are populated: 
from pipecat.pipeline.task import PipelineTask
from pipecat.pipeline.pipeline_params import PipelineParams

turn_tracker = TurnTrackingObserver()

task = PipelineTask(
    pipeline,
    params=PipelineParams(
        observers=[observer.latency_observer, turn_tracker],
        enable_metrics=True,
        enable_usage_metrics=True,
    ),
)
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.

​
Configuration Options

Override the default call type label:
FlowsObserver(..., call_type="web_call")
FlowsObserver(..., call_type="phone_call")
Provide a recording URL if available. Default is "pipecat://no-recording":
FlowsObserver(..., recording_url="https://cdn.example.com/recordings/call-123.ogg")
Specify your ASR, LLM, and TTS models for metadata:
FlowsObserver(
    ...,
    asr_model="deepgram/nova-3",
    llm_model="gpt-4o-mini",
    tts_model="cartesia/sonic",
)
Log the full transcript when flushing:
FlowsObserver(..., debug=True)

​
Full Example

import uuid
from tuner_pipecat_sdk import FlowsObserver
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.pipeline.task import PipelineTask
from pipecat.pipeline.pipeline_params import PipelineParams

turn_tracker = TurnTrackingObserver()

observer = FlowsObserver(
    api_key=os.environ["TUNER_API_KEY"],
    workspace_id=int(os.environ["TUNER_WORKSPACE_ID"]),
    agent_id="my-agent",
    call_id=str(uuid.uuid4()),
    base_url="https://api.usetuner.ai",
    call_type="web_call",
    asr_model="deepgram/nova-3",
    llm_model="gpt-4o-mini",
    tts_model="cartesia/sonic",
)

observer.attach_flow_manager(flow_manager)
observer.attach_turn_tracking_observer(turn_tracker)

pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts,
    observer,
    transport.output(),
    context_aggregator.assistant(),
])

task = PipelineTask(
    pipeline,
    params=PipelineParams(
        observers=[observer.latency_observer, turn_tracker],
        enable_metrics=True,
        enable_usage_metrics=True,
    ),
)

​
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 matches the Agent ID configured in Tuner under Agent Settings > Agent Connection.
  • 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.

​
What’s Next?

Configuring Your Agent

Set up call outcomes, user intents, and behavior checks.

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.