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
)

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

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

Development

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