tuner-core 0.1.0

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)

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