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.The SDK ships two observer types depending on how your pipeline is structured:
Observer
When to use
Observer
Plain pipecat-ai pipelines using LLMContext directly
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 remote ID.
3
Add the Observer
Create the correct observer for your pipeline type, attach it, and place it after TTS.
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.
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.
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:
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 = Nonedef resolve_reason() -> str | None: return _reasonobserver = Observer(..., disconnection_reason_resolver=resolve_reason)# Set the reason when your app knows it_reason = DisconnectReason.USER_HANGUP
Works the same way on both Observer and FlowsObserver.
Model Names
Specify your ASR, LLM, and TTS models for metadata:
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 a UsageSummary 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: CallUsage) -> 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) * 100Observer(cost_calculator=calculate_cost)
import osimport uuidfrom tuner_pipecat_sdk import Observerfrom tuner_pipecat_sdk.models import DisconnectReasonfrom pipecat.observers.turn_tracking_observer import TurnTrackingObserverfrom pipecat.pipeline.task import PipelineTask, PipelineParamsasync 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 model def calculate_cost(usage: CallUsage) -> 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_context(context) 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( enable_metrics=True, enable_usage_metrics=True, ), observers=[observer.latency_observer, turn_tracker], ) _reason = DisconnectReason.USER_HANGUP
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.
