LiveKit Agents

  LiveKit Agents (realtime pipeline)        Naive
  ─────────────────────────────────        ─────
  caller speaks ─▶ STT ─▶ LLM (reasons)
                          │  tool call (naive_cards_create)
                          ▼  mcp.MCPServerHTTP (SSE + headers)
  AgentSession ── /mcp/sse/sess_… ────────────▶ per-user session
    (Authorization: Bearer nv_sess_…)           │  AccountKit-gated, scoped to one user
                                                ▼
                                         connect GitHub · issue a $50 card · run a capability
                                                │
                                         sensitive? → 202 pending_approval (human-in-the-loop)
                          ┌─────────────────────┘
                          ▼
  LLM ─▶ TTS ─▶ "I've queued that card for approval." ─▶ caller hears it

pip install "livekit-agents[silero,turn-detector]" mcp httpx

export NAIVE_API_KEY=nv_sk_live_...
export LIVEKIT_URL=wss://your-project.livekit.cloud
export LIVEKIT_API_KEY=...
export LIVEKIT_API_SECRET=...

from livekit.agents.llm import mcp
from livekit.agents.llm.mcp import MCPToolResultContext

def spoken_result(ctx: MCPToolResultContext):
    """Shape what the voice agent says when Naive queues an approval."""
    for part in ctx.result.content:
        text = getattr(part, "text", "")
        if "pending_approval" in text:
            return "That action needs approval — I've queued it for a human to confirm."
    return ctx.result  # let non-gated results flow through unchanged

naive = mcp.MCPServerHTTP(
    url=m["url"],
    transport_type="sse",
    headers=m["headers"],
    allowed_tools=["naive_connections_connect", "naive_cards_create"],  # least privilege
    tool_result_resolver=spoken_result,
)

{
  "status": "pending_approval",
  "approval_id": "65589c8b-e033-4a65-b16c-379211c94429",
  "action_type": "cards.create",
  "primitive": "cards",
  "title": "Issue virtual card \"Ads budget\""
}

def list_pending(user_id: str) -> list[dict]:
    r = httpx.get(
        f"{NAIVE_API}/v1/users/{user_id}/approvals",
        headers=AUTH,
        params={"status": "pending"},
    )
    r.raise_for_status()
    return r.json()["approvals"]

def approve(user_id: str, approval_id: str) -> dict:
    r = httpx.post(
        f"{NAIVE_API}/v1/users/{user_id}/approvals/{approval_id}/approve",
        headers=AUTH,
    )
    r.raise_for_status()
    return r.json()

# Find what the agent queued for Alice, then approve (or deny) it.
for a in list_pending("<user_id>"):
    # ...show a["title"] / a["action_type"] to a human in your UI...
    approve("<user_id>", a["id"])  # API replays cards.create → real card

import json

@server.rtc_session()
async def entrypoint(ctx: JobContext):
    await ctx.connect()

    # One dispatched job == one caller. Carry the tenant on the job metadata.
    meta = json.loads(ctx.job.metadata or "{}")
    user_id = meta["naive_user_id"]

    m = mint_session(user_id)["mcp"]  # fresh, per-user session
    naive = mcp.MCPServerHTTP(
        url=m["url"],
        transport_type="sse",
        headers=m["headers"],
        allowed_tools=["naive_connections_connect", "naive_cards_create"],
    )

    agent = Agent(instructions="...", mcp_servers=[naive])
    session = AgentSession(
        vad=silero.VAD.load(),
        stt=inference.STT("deepgram/nova-3", language="multi"),
        llm=inference.LLM("openai/gpt-4.1-mini"),
        tts=inference.TTS("cartesia/sonic-3"),
    )
    await session.start(agent=agent, room=ctx.room)

# Same agent definition, different caller — isolated identity, spend, and approvals each time.