tuner-core 0.1.1

Core session coordination and API client for Tuner observability SDKs.

tuner-core

Session coordination, configuration, and API client for the Tuner observability SDKs. Zero provider dependencies — every role package (tuner-stt-observer, tuner-tts-observer, tuner-llm-observer, tuner-langchain) depends on this; this package depends on nothing else in the SDK.

Installation

pip install tuner-core

tuner-core alone is enough if you're calling session._record_user_turn() / _record_agent_turn() directly (e.g. bridging a framework like LiveKit or Pipecat that owns its own provider connections). Add a role package to get automatic recording for a specific provider.

TunerConfig

Single configuration object, built from environment variables or explicit kwargs — explicit kwargs always win.

from tuner_core import TunerConfig

config = TunerConfig.from_env(
    asr_model="nova-2",
    llm_model="gpt-4o-mini",
    tts_model="sonic-2",
    cost_calculator=my_cost_fn,   # optional — receives a CallUsage, returns cents
)

Env var	Required	Purpose
TUNER_API_KEY	Yes	Bearer token (tr_api_...)
TUNER_WORKSPACE_ID	Yes	Integer workspace ID
TUNER_AGENT_ID	Yes	Agent identifier from Agent Settings
TUNER_BASE_URL	No	Defaults to https://api.usetuner.ai
TUNER_DEBUG	No	"true" to log the full payload on flush
AGENT_VERSION	No	Integer agent version tag (note: no TUNER_ prefix)

Extra metadata

Attach arbitrary key-value data to every call record:

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

extra_metadata is explicit-kwarg-only (no env var). Keys are merged into general_meta_data_raw at flush time — a key named ai_models or usage_token is reserved for the SDK's own computed values and is dropped (with a logged warning) rather than overwriting them.

Simulation Correlation using SIP call ID

If your agent runs over SIP telephony, pass the call's SIP Call-ID so Tuner can link a simulation run to the call record your agent submits. Without it, simulated calls still show up in Tuner — they just won't be tagged as simulations.

config = TunerConfig.from_env(sip_call_id=sip_id)

tuner-core doesn't talk to your telephony stack directly, so it can't read this value for you — you resolve it from your own SIP session or provider webhook and pass it in explicitly. (If you're on LiveKit or Pipecat, our dedicated SDKs for those platforms capture this automatically.)

TunerSession

One session per call. Attach handlers as you create provider connections; call flush() when the call ends.

from tuner_core import TunerSession

session = TunerSession(config=config, call_id=call_id)

handler = session.attach(SomeHandler())   # returns the handler for chaining

await session.flush()   # POSTs the assembled call payload; never raises

attach() supports two kinds of handlers — see below. It returns whatever you pass in, so adapter = session.attach(CartesiaAdapter()) works in one line.

Two handler contracts

1. Transcript collectors — extend BaseTunerHandler

For handlers that record turns as they happen: STT and TTS adapters. TunerSession.attach() detects the BaseTunerHandler subclass and injects itself via _bind(), so the handler can call _record_user_turn() / _record_agent_turn() / _record_stt_usage() / _record_tts_usage() directly.

# tuner_stt_observer.DeepgramAdapter does this
class DeepgramAdapter(BaseSTTAdapter):   # BaseSTTAdapter extends BaseTunerHandler
    def _on_transcript(self, event):
        self._record_user_turn(text=..., timestamp_ms=...)

2. LLM trace providers — implement get_segments() (coming)

For LLM framework handlers that accumulate execution data. No inheritance from tuner-core needed — TunerSession finds them via duck typing.

Useful TunerSession state

• session.last_user_end_ms — relative ms when the last user utterance ended; TTS adapters read this to compute e2e_latency.
• session.last_llm_duration_ms — most recent LLM turn's processing time; TTS adapters attach it as llm_node_ttft.
• session.disconnection_reason — settable; pass a DisconnectReason value before flush().

Full example — custom FastAPI + LangGraph + Deepgram + Cartesia

import os
import uuid
from fastapi import FastAPI, WebSocket
from cartesia import AsyncCartesia
from tuner_core import TunerConfig, TunerSession
from tuner_stt_observer import DeepgramAdapter
from tuner_tts_observer import CartesiaAdapter
from tuner_langchain import wrap_graph

app = FastAPI()

_agent = build_agent()  # your LangGraph graph

@app.websocket("/call")
async def handle_call(websocket: WebSocket):
    await websocket.accept()

    call_id = str(uuid.uuid4())
    config = TunerConfig.from_env(asr_model="nova-2", llm_model="gpt-4o-mini", tts_model="sonic-2")
    session = TunerSession(config=config, call_id=call_id)

    # STT — attach before connection.start()
    dg_connection = dg_client.listen.asyncwebsocket.v("1")
    dg_adapter = DeepgramAdapter(connection=dg_connection)
    session.attach(dg_adapter)
    await dg_connection.start(options)

    # TTS — attach once per call, used per utterance
    cartesia_client = AsyncCartesia(api_key=os.environ["CARTESIA_API_KEY"])
    cartesia_adapter = session.attach(CartesiaAdapter())

    # LLM — wraps the graph, exposes get_segments() for duck typing
    instrumented_graph = wrap_graph(_agent)
    session.attach(instrumented_graph)

    # In your agent turn loop:
    # result = await instrumented_graph.ainvoke({"messages": history})
    # agent_text = result["messages"][-1].content
    #
    # async with cartesia_adapter.track_ws(agent_text) as tracked:
    #     async for chunk in tracked(ctx.receive()):
    #         if chunk.audio:
    #             await websocket.send_bytes(chunk.audio)

    await session.flush()  # at call end

What gets captured automatically

Signal	Source	Handler
User transcript	Deepgram Transcript/UtteranceEnd events	DeepgramAdapter
Turn timestamps	Provider event timing + stream open time	DeepgramAdapter / SpeechmaticsAdapter
STT latency	User speech end → transcript delta	SpeechmaticsAdapter / DeepgramAdapter
Agent transcript	Text passed to CartesiaAdapter.track() / track_ws()	CartesiaAdapter
TTS TTFB	Synthesis request → first audio chunk	CartesiaAdapter
E2e latency	User speech end → agent first audio byte	CartesiaAdapter (reads session.last_user_end_ms)
LLM latency	Graph/chain invocation duration	wrap_graph / wrap_chain (tuner-langchain)
Tool calls + results	Graph/chain callbacks	wrap_graph / wrap_chain (tuner-langchain)

Submission

submit_call() (in client.py) never raises — failures are logged and swallowed so a Tuner outage can't crash the voice agent it's observing.

• Retries on 429/5xx and network/timeout errors; abandons immediately on other 4xx.
• Backoff: 1s, 2s, 4s + up to 500ms jitter, max_retries attempts (default 3).
• 409 (duplicate call) is treated as success and logged, not retried.

Package install matrix

Stack	Install	Status
Custom stack + LangGraph + Deepgram/Speechmatics + Cartesia	tuner-core tuner-stt-observer tuner-tts-observer tuner-langchain	✅ Supported
Custom stack, OpenAI/Anthropic LLM only	tuner-core tuner-llm-observer	✅ Supported (OpenAI adapter only — see tuner-llm-observer)
LiveKit — dedicated integration package	tuner-livekit-sdk	✅ Supported
Pipecat — dedicated integration package	tuner-pipecat-sdk	✅ Supported

Development

uv sync
uv run pytest
uv run ruff check .
uv run mypy