Skip to main content
This guide covers both the Python SDK (tuner-livekit-sdk) and the Node.js SDK (@usetuner.ai/livekit-sdk). Use the language switcher below to view the guide for your stack.

Video Tutorial

Prerequisites

  • Tuner Active Account
  • Configured Agent with provider “Custom API” in Tuner
  • A LiveKit Agents project, either:

Overview

The Tuner LiveKit SDK automatically captures session data from your LiveKit agent and sends it to Tuner when the call ends — no manual API calls required. It ships in two flavours that share the same wire format and feature set: a Python package and a Node.js package.
1

Install the SDK

Add the package to your project.
2

Set Your Credentials

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

Add the Plugin

Drop two lines into your entrypoint and you’re done.
Estimated time: 2 minutes from start to finish

Step 1: Install the SDK

pip install tuner-livekit-sdk
Requirements: Python ≥ 3.10, livekit-agents >= 1.4, aiohttp >= 3.9

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="fa4da74c-..."
VariableRequiredDescription
TUNER_API_KEYBearer token (starts with tr_api_)
TUNER_WORKSPACE_IDYour Tuner workspace ID
TUNER_AGENT_IDAgent Remote ID from Agent Settings → Agent Connection (UUID — not your agent’s display name)
TUNER_BASE_URLAPI base URL (default: https://api.usetuner.ai)
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.

Step 3: Add the Plugin

Import TunerPlugin and add it right after creating your AgentSession — before calling session.start():
from tuner import TunerPlugin

async def entrypoint(ctx: JobContext):
    session = AgentSession(...)
    TunerPlugin(session, ctx)          # wires itself automatically
    await session.start(...)
That’s it. The plugin listens to session events and submits call data to Tuner when the session ends.

SIP correlation for simulation

If you plan to run Call Simulation against this agent, the plugin needs one extra value: the SIP Call-ID of the inbound call. Tuner uses it to match the simulation call it just dialled with the session your agent 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.
LiveKit attaches the SIP Call-ID to the SIP participant as the sip.callIDFull attribute. Read it after the room is connected and pass it to TunerPlugin as sip_correlation_id.
Requires tuner-livekit-sdk >= 0.1.5. The sip_correlation_id argument was added in 0.1.5.

Helper: extract the SIP correlation ID

Add this helper to your entrypoint module. It scans the room’s remote participants for the SIP participant and returns its sip.callIDFull attribute.
from livekit import rtc
from livekit.agents import JobContext


def _extract_sip_correlation_id(ctx: JobContext) -> str | None:
    for participant in ctx.room.remote_participants.values():
        if participant.kind != rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
            continue
        attributes = dict(getattr(participant, "attributes", {}) or {})
        sip_call_id_full = attributes.get("sip.callIDFull")
        if isinstance(sip_call_id_full, str) and sip_call_id_full:
            return sip_call_id_full
        return None
    return None
For web calls (no SIP participant), this returns None and TunerPlugin falls back to its normal behaviour.

Wire it into your entrypoint

The SIP participant only becomes visible after the agent joins the room. Call ctx.connect() first, then read the correlation ID, then pass it to TunerPlugin:
async def entrypoint(ctx: JobContext):
    await ctx.connect()                                     # 1. connect first
    sip_correlation_id = _extract_sip_correlation_id(ctx)   # 2. then read

    session = AgentSession(...)

    TunerPlugin(
        session, ctx,
        sip_correlation_id=sip_correlation_id,              # 3. pass it in
        # ... your other TunerPlugin options
    )

    await session.start(...)
If ctx.connect() runs after you try to read participants, ctx.room.remote_participants will be empty and sip_correlation_id will be None. Simulated calls will still complete, but they won’t be linked to their Tuner simulation rows.

Disconnection Reasons

Tuner captures why each call ended. User disconnects and errors are tracked automatically. To record agent-initiated hangups, pass reason="agent_hangup" to ctx.shutdown() when ending the call programmatically:
get_job_context().shutdown(reason="agent_hangup")
Without this, agent hangups are indistinguishable from dropped calls in your analytics.
@function_tool
async def end_call(self, context: RunContext):
    """End the call once the user has been helped."""
    await context.session.shutdown(drain=True)
    get_job_context().shutdown(reason="agent_hangup")
drain=True lets any in-flight audio finish before the session closes. Omit it for an immediate cutoff.

LangChain / LangGraph observability

If your agent uses LangGraph or LangChain for its logic layer, you can capture node transitions, tool calls, and timing alongside the standard session data. The plugin handles all the wiring and flushes the events automatically when the session ends.

Extra install

pip install tuner-langchain

Attach to a LangGraph graph

Save the plugin reference, then call wrap_graph() and pass the result into langchain.LLMAdapter. The handler records what the graph does on every invocation.
from tuner import TunerPlugin

async def entrypoint(ctx: JobContext):
    session = AgentSession(...)
    plugin = TunerPlugin(session, ctx)      # save the reference

    wrapped_graph = plugin.wrap_graph(my_graph)

    agent = Agent(
        instructions="",
        llm=langchain.LLMAdapter(
            wrapped_graph,
            stream_mode="custom",
            config={"configurable": {"thread_id": thread_id}},
        ),
    )
    await session.start(agent=agent, ...)

Attach to a plain LangChain chain

For plain chains (no graph), use wrap_chain() instead:
wrapped_chain = plugin.wrap_chain(my_chain)

agent = Agent(
    instructions="",
    llm=langchain.LLMAdapter(
        wrapped_chain,
        stream_mode="messages",
        config={"configurable": {"thread_id": thread_id}},
    ),
)

Privacy controls

By default, node inputs, outputs, system prompts, and tool payloads are all captured. Pass a CaptureConfig to suppress what you don’t need:
from tuner_langchain import CaptureConfig

wrapped_graph = plugin.wrap_graph(
    my_graph,
    capture=CaptureConfig(
        node_instructions=False,   # system prompts
        tool_inputs=False,         # tool parameters
    )
)
Tool error output is always captured regardless of CaptureConfig. Errors are needed for debugging and are not treated as sensitive.

Configuration Options

The plugin auto-detects the call type (phone_call for SIP participants, web_call otherwise). Override it explicitly when needed:
TunerPlugin(session, ctx, call_type="phone_call")
TunerPlugin(session, ctx, call_type="web_call")
The phone number or SIP URI of the callee. Optional — pass it when you want to record who the call was directed to:
TunerPlugin(session, ctx, recipient="+15551234567")
TunerPlugin(session, ctx, recipient="sip:+15551234567@provider.com")
Tuner requires a recording_url for every call. Provide a resolver function that returns the URL. If you don’t provide one, the plugin submits "pending" as a placeholder.
# Static / pre-known URL
async def my_resolver(room_name: str, job_id: str) -> str:
    return f"https://cdn.example.com/recordings/{job_id}.ogg"

TunerPlugin(session, ctx, recording_url_resolver=my_resolver)
# LiveKit Egress → S3
async def egress_resolver(room_name: str, job_id: str) -> str:
    url = await my_egress_db.get_recording_url(room_name)
    return url or "pending"

TunerPlugin(session, ctx, recording_url_resolver=egress_resolver)
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) -> float:
    llm_cost  = usage.llm_prompt_tokens     * 0.000_003
    llm_cost += usage.llm_completion_tokens * 0.000_015
    tts_cost  = usage.tts_characters_count  * 0.000_030
    stt_cost  = usage.stt_audio_duration    * 0.000_006
    return llm_cost + tts_cost + stt_cost

TunerPlugin(session, ctx, cost_calculator=calculate_cost)
Attach arbitrary key-value data to every call record:
TunerPlugin(
    session, ctx,
    extra_metadata={
        "env": "production",
        "region": "us-east-1",
        "deployment": "v2.3.1",
    },
)
TunerPlugin(
    session, ctx,
    timeout_seconds=15.0,   # per-request timeout (default: 30.0)
    max_retries=5,          # retries on 5xx / 429 / network errors (default: 3)
)
Track which version of your agent handled each call — useful when you update a prompt, swap a model, or change your pipeline:
AGENT_VERSION=42 python agent.py start
Tuner reads it automatically. Bump the number on every deployment.Override in code (takes priority over the env var):
TunerPlugin(
    session, ctx, agent_version=42,...
)
Useful for local development or test environments:
import os

TunerPlugin(
    session, ctx,
    enabled=os.getenv("ENV") == "production",
)

Full Example

import os
from livekit import rtc
from livekit.agents import JobContext, AgentSession
from tuner import TunerPlugin


def _extract_sip_correlation_id(ctx: JobContext) -> str | None:
    for participant in ctx.room.remote_participants.values():
        if participant.kind != rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
            continue
        attributes = dict(getattr(participant, "attributes", {}) or {})
        sip_call_id_full = attributes.get("sip.callIDFull")
        if isinstance(sip_call_id_full, str) and sip_call_id_full:
            return sip_call_id_full
    return None


def calculate_cost(usage) -> float:
    return (
        usage.llm_prompt_tokens     * 0.000_003
        + usage.llm_completion_tokens * 0.000_015
        + usage.tts_characters_count  * 0.000_030
    )


async def get_recording_url(room_name: str, job_id: str) -> str:
    return await my_storage.get_url(job_id) or "pending"


async def entrypoint(ctx: JobContext):
    session = AgentSession(...)

    await ctx.connect()
    sip_correlation_id = _extract_sip_correlation_id(ctx)

    TunerPlugin(
        session,
        ctx,
        api_key=os.environ["TUNER_API_KEY"],
        workspace_id=int(os.environ["TUNER_WORKSPACE_ID"]),
        agent_id=os.environ["TUNER_AGENT_ID"],
        call_type="phone_call",
        recording_url_resolver=get_recording_url,
        cost_calculator=calculate_cost,
        sip_correlation_id=sip_correlation_id,
        extra_metadata={"env": "prod", "region": "us-east-1"},
        timeout_seconds=20.0,
        max_retries=3,
        enabled=True,
    )

    await session.start(...)

Troubleshooting

  • Verify that TUNER_AGENT_ID is the Agent Remote ID from Agent Settings → Agent Connection (not your agent’s display name).
  • Confirm TUNER_WORKSPACE_ID is correct.
  • Check your application logs for any error messages from the plugin.
  • Ensure TUNER_API_KEY starts with tr_api_ and is valid.
  • Confirm the API key belongs to the correct workspace.
  • You haven’t provided a recording_url_resolver. Add one that returns the actual recording URL for each call.
  • If using LiveKit Egress, ensure the recording has finished processing before the resolver is called.

What’s Next?

Configuring Your Agent

Set up call outcomes, user intents, and evals.

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.