Developer Docs

Everything here reflects the real, live system — every command on this page has been run against production and verified to work exactly as shown.

1. Payment and credit

One fixed tier: $28/month. No metering, no custom amounts — fiatAmount is a fixed server-side constant, never read from your request. See Pricing for the checkout form.

Credit unit: 1 credit = 1 US cent. $28 = 2800 credits.

Billing runs in 6-day cycles, not calendar months. Each cycle costs 560 credits (2800 ÷ 560 = 5 cycles ≈ one $28 payment's worth of runway). The first 560-credit charge happens atomically at the moment your instance successfully starts running — not retroactively at the end of the cycle — so there's no free trial period hiding in the billing math. Every cycle after that is checked and charged automatically, in advance, every 6 days.

A free trial may also be available — new accounts can be granted a one-time 560-credit bonus (enough for exactly one 6-day trial instance) via the signup form on the homepage. This is independently toggleable and isn't always open; if it's closed, /signup returns {"status": "trial_closed"} and checkout (paid) is the reliable path in.

Multi-instance: one AWS account can run up to 4 instances at once, sharing one credit pool. If your account has more than one instance due for a cycle charge at the same time and the shared balance can't cover all of them, the oldest instance is protected first — newest instance(s) are the ones deprovisioned if credit runs short.

Standing balance cap: at most 22,400 credits (2 months × 4 instances) can be held prepaid at once — enforced before any payment is even created, so a rejected top-up never becomes a payment you'd need refunded.

Running out of credit: when a cycle's charge can't be covered, that instance is deprovisioned automatically — the underlying server, its static IP, and its DNS record are all torn down. Top up again via Pricing any time before or after.

2. Setting up and installation

Your AWS account is your login — every authenticated call below is signed with your own AWS credentials (SigV4), the same mechanism the aws CLI already uses. No separate API key or password to manage.

Step 1 — Get an account and credit

Submit the checkout form on Pricing (email + 12-digit AWS account number) and pay. This registers your account (if it doesn't exist yet) and, once the payment settles, credits it automatically — no manual step on your end.

Step 2 — Provision your instance

All authenticated calls need a SigV4-signing tool. awscurl is the simplest option:

pip install awscurl

export AWS_ACCESS_KEY_ID=<your access key>
export AWS_SECRET_ACCESS_KEY=<your secret key>

awscurl --service execute-api --region ap-southeast-1 -X POST "https://api.niuty.com/create-instance"

Returns once the instance is actually running (usually 15-90 seconds):

{"status": "creating", "instance_id": "<id>", "livekit_url": "wss://<id>.rt.niuty.com"}

Choosing a region: pass ?region=EU or ?region=SG (case-insensitive) to pick where your instance runs. Omit it and you get the platform default. Any other value returns a 400 listing the currently valid regions - more are planned, but only ones that are actually live are ever accepted.

awscurl --service execute-api --region ap-southeast-1 -X POST "https://api.niuty.com/create-instance?region=SG"

Audio-only vs. audio+video: if your agent only ever publishes/subscribes audio tracks, either region works equally well. If you're streaming real video alongside audio (e.g. a camera-equipped robot, or anything using /session?enableCamera=1), choose EU (?region=EU) — it's provisioned with meaningfully more real-time throughput headroom and is the better fit for audio+video workloads today. SG is tuned specifically for audio-only agents and may throttle a video track under sustained real use.

Once your own agent process is up and listening for a job on this instance, you can talk to it directly from your browser — no frontend of your own required yet — via Test Your Agent.

/session is the same underlying flow (audio and video, for an agent that also publishes a video track) as a single shareable link instead of a page you paste a token into - not linked from the site nav, and deliberately not search-indexed, since it's meant to be sent directly to one person rather than browsed to. /test-token's response already includes a ready-to-use sessionUrl field (already URL-encoded) - just copy that, no need to build it by hand. Pass expiresInHours to /test-token if you want the link to stay valid longer than the default 1 hour (up to 7 days) - useful for an evergreen demo link rather than a one-off test.

Opening the link connects immediately, full-screen-capable, no paste step - whoever you send it to just sees/hears the agent right away.

Add &audioOnly=1 to skip video entirely - swaps the fullscreen video stage for a minimal "connected, talk into your microphone" view, same as Test Your Agent but via the single-link flow instead of paste. /session also always exposes a two-way data-channel bridge (window.NiutyBridge/window.AndroidApp, both modes, video or audio-only) - this is what makes it embeddable inside your own native app instead of just a page a human opens directly; see Embedding in a native Android app below.

Tune the browser's own mic processing with &echoCancellation=, &noiseSuppression=, and &autoGainControl= (each 1/0 or true/false) - these map directly to WebRTC's own capture constraints, applied when /session publishes your microphone. Any you omit are left at the browser's own default (normally on). Useful if you're using a mic that already does its own noise reduction (e.g. a lavalier) and don't want WebRTC's processing fighting it - for example, keep echo cancellation on for TV-speaker feedback while turning off noise suppression and AGC so they don't distort the input:

https://niuty.com/session?token=<token>&livekitUrl=<livekitUrl>&noiseSuppression=0&autoGainControl=0

Pick a specific mic with &audioDeviceId=<id> — for deterministic device selection on fixed kiosk hardware (e.g. an external lavalier), instead of letting the browser/OS pick whatever it currently calls the default input. Get a device's ID from the browser's own MediaDevices.enumerateDevices() on the same machine.

Publish your own camera, not just receive the agent's, with &enableCamera=1 — /session otherwise never turns your camera on at all (it only ever attaches the agent's video, if it publishes one). This is for a kiosk/embodied-agent setup — a dedicated webcam feeding the agent's own vision pipeline, not a video call — so it's an explicit opt-in, independent of audioOnly: you can combine audioOnly=1&enableCamera=1 to keep the minimal "talk into your microphone" view while still publishing your camera underneath it.

Once enableCamera=1 is set, the same kind of device/quality controls apply to video: &videoDeviceId=<id> (pick a specific camera — the dedicated overhead one, not a laptop's built-in webcam), and &videoWidth=/&videoHeight=/&videoFrameRate= for capture quality (e.g. request 1080p or 4K for better computer-vision input):

https://niuty.com/session?token=<token>&livekitUrl=<livekitUrl>&enableCamera=1&videoDeviceId=<id>&videoWidth=1920&videoHeight=1080&videoFrameRate=30

Save instance_id — every other call needs it. This call charges 560 credits atomically on success; if you don't have enough credit, it fails cleanly with {"status": "error", "message": "Insufficient credit."} and nothing is charged.

Step 3 — Check status (optional)

# One specific instance
awscurl --service execute-api --region ap-southeast-1 -X GET "https://api.niuty.com/status?instanceId=<instance_id>"

# Account-wide overview - credit balance + every instance you have
awscurl --service execute-api --region ap-southeast-1 -X GET "https://api.niuty.com/status"

The single-instance response includes the region it's actually running in (EU/SG, confirming whatever you picked, or the platform default if you didn't pick one):

{"instance_id": "<id>", "status": "active", "tier": 1, "region": "EU", "instance_ip": "1.2.3.4", "livekit_url": "wss://<id>.rt.niuty.com"}

If an instance looks unhealthy (status checks failing, unreachable, nothing responding) — reboot it yourself, no need to contact us:

awscurl --service execute-api --region ap-southeast-1 -X POST "https://api.niuty.com/reboot-instance?instanceId=<instance_id>"

Takes about a minute to come back. Rate-limited to one request per instance per 2 minutes — a second call before that window passes returns {"status": "error", ...} with a 429, not a second reboot.

Step 4 — Fetch your credential

awscurl --service execute-api --region ap-southeast-1 -X GET "https://api.niuty.com/fetch-credential?instanceId=<instance_id>" > ~/.niuty/credentials.enc

This is an encrypted envelope (key_id/nonce/ciphertext), not raw LiveKit credentials — only live-lib (below) can decrypt it. ~/.niuty/credentials.enc is the default path every live-lib API in this doc reads from.

Step 5 — Install live-lib

Run this inside the virtualenv you'll actually use for your agent (python3 -m venv .venv && source .venv/bin/activate first, if you haven't already):

mkdir -p ~/.niuty  # if it doesn't already exist

aws codeartifact login --tool pip --domain niuty --domain-owner 355681235310 --repository niuty-python --region ap-southeast-1

# aws codeartifact login always writes the index-url globally (affects every
# venv and your system Python, not just this one) - move it into this venv
# only, then remove the global copy:
pip config --site set global.index-url "$(pip config get global.index-url)"
pip config --user unset global.index-url

pip install "live-lib[livekit]"

Use the [livekit] extra — the base live-lib install is schema-only (for tools that just need to author/export flows) and deliberately doesn't pull in the LiveKit/OpenAI/Google SDK stack.

CodeArtifact's own upstream chain (niuty-python → pypi-store → public PyPI) means this same index also transparently serves ordinary public packages (boto3, awscurl, etc.) — you never need a second index configured for anything else this venv installs.

Troubleshooting:

• 401 Error, Credentials not correct on any pip install, even unrelated public packages: the CodeArtifact login token embedded in your pip config has expired — it's only valid 12 hours. Fix: re-run the aws codeartifact login + pip config --site set / pip config --user unset block above (inside the same venv) to get a fresh token.
• A different, unrelated venv (or your bare system Python) is also trying to hit CodeArtifact, or fails the same way: the index-url leaked into the global/user-level pip config instead of staying scoped to one venv — most often from running plain aws codeartifact login --tool pip ... without the --site/--user follow-up commands above. Fix: pip config --user unset global.index-url (removes it everywhere except whichever venvs already have their own --site-scoped copy), then redo the Step 5 block above inside whichever venv(s) actually need it.

If codeartifact login fails with AccessDeniedException on GetAuthorizationToken: your AWS account is already granted access at the resource-policy level (that happens automatically when your account registers), but if you're calling as a specific IAM user rather than the account root, that user also needs their own identity-based policy allowing it — a resource policy alone isn't sufficient for a non-root caller. Attach this to the IAM user you're using:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "codeartifact:GetAuthorizationToken",
        "codeartifact:GetRepositoryEndpoint",
        "codeartifact:ReadFromRepository"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": "sts:GetServiceBearerToken",
      "Resource": "*",
      "Condition": { "StringEquals": { "sts:AWSServiceName": "codeartifact.amazonaws.com" } }
    }
  ]
}

(Or simply call codeartifact login with your account's root credentials for this one step — root isn't subject to identity-based policy restrictions at all.)

3. live-lib API usage

This covers the APIs actually used in a real, production live-lib integration — not the full surface. Two independent halves: FlowGraph (design-time — build your agent's conversation logic in Python) and agent_runtime (run-time — connect that logic to a live LiveKit voice session using your niuty-provisioned instance).

FlowGraph — designing your agent's conversation

from live_lib import FlowGraph

graph = FlowGraph(name="support_agent", persona="Friendly, concise support rep.")

@graph.node("MAIN_ROUTING", objective="Figure out what the caller needs")
def main_routing(n):
    n.instructions("Greet the caller.", "Ask how you can help.")
    n.on(condition="caller wants billing help", to="BILLING")
    n.on(condition="caller wants technical support", to="SUPPORT")

@graph.node("BILLING", objective="Resolve a billing question")
def billing(n):
    n.instructions("Ask for the account or invoice number.", "Explain the charge or issue a fix.")
    n.on(condition="caller's question is resolved", to="MAIN_ROUTING")
    n.on(condition="caller needs a human", to="SUPPORT")

graph.entry("MAIN_ROUTING")
flow = graph.compile()

Each node is a live conversation state, not a text-generation step — transitions (n.on) are what your caller actually says next, not function calls. Flows are cyclic state machines (see BILLING looping back to MAIN_ROUTING above), not one-way scripts. See FlowGraph on the homepage for the full visual walkthrough.

LiveKitBridgeWorker — connecting a compiled flow to a live session

from live_lib.agent_runtime import LiveKitBridgeWorker

worker = LiveKitBridgeWorker(
    openai_api_key="<your OpenAI key>",
    google_api_key="",              # optional, only if using Google's models
    agent_name="my-agent-bridge",   # must match your dispatch/room-naming setup
    room_id_prefix="agent-",
)
worker.start()

livekit_url/api_key/api_secret are never constructor parameters — the worker reads them itself, internally, by decrypting ~/.niuty/credentials.enc (or a custom path via credential_path=). There's no other way to supply them; a missing or invalid credential file raises CredentialError and the worker never starts. This is deliberate: it means a real LiveKit connection is only ever possible using an instance you've actually provisioned and paid for through this system, not an ad-hoc URL/key pasted into code.

After construction, worker.livekit_url / worker.api_key are readable (useful for logging) even though they were never passed in.

Two more options (live-lib 1.9.3+) — see Data handling and privacy:

worker = LiveKitBridgeWorker(
    openai_api_key="<your OpenAI key>",
    log_conversation_content=False,   # log lengths only, never transcript/text content
)
worker.start()
...
worker.stop()   # graceful shutdown: ends active sessions, disconnects, returns True when done

Identifying the agent in the room (custom clients)

If you build your own client instead of using /session (for example, a receive-only player that subscribes manually), bind it to the agent's participant in the room, never to agent_name:

• agent_name is only the dispatch name. It decides which worker receives the job. It is not the agent's identity in the room.
• The agent's room identity is assigned per job by the server, in the form agent-<job id>. It is new for every dispatch, so it can't be known or configured in advance.
• What you can rely on is the participant kind: the agent always joins with kind AGENT. Clients joining with tokens from /test-token or generate_participant_token() never do.

Recommended binding, in your client (LiveKit client SDK, JavaScript shown):

import { RoomEvent, ParticipantKind } from "livekit-client";

let agent = null; // { identity, sid }

function bindAgent(room) {
  const agents = [...room.remoteParticipants.values()].filter(p => p.kind === ParticipantKind.AGENT);
  if (agents.length !== 1) return;          // fail closed: none yet, or unexpected extra agents
  agent = { identity: agents[0].identity, sid: agents[0].sid };
  // Tracks it published before we bound (TrackPublished won't fire again for these):
  for (const pub of agents[0].audioTrackPublications.values()) pub.setSubscribed(true);
}

// Connect with autoSubscribe off, then bind - the agent may already be in the room:
await room.connect(url, token, { autoSubscribe: false });
bindAgent(room);
room.on(RoomEvent.ParticipantConnected, () => { if (!agent) bindAgent(room); });
room.on(RoomEvent.ParticipantDisconnected, p => {
  if (agent && p.sid === agent.sid) agent = null;   // job ended: retire this binding permanently
});
// Only accept audio published by the bound participant:
room.on(RoomEvent.TrackPublished, (pub, p) => {
  if (agent && p.sid === agent.sid && pub.kind === "audio") pub.setSubscribed(true);
});

Behavior to expect:

• One dispatch per room means exactly one agent participant. If you ever see more than one, don't pick one; treat it as an error.
• A new dispatch creates a new participant with a new identity and SID. Treat it as a new binding.
• The agent can restart its voice session inside the same job (for example, after a model reconnect). The participant identity and SID stay the same, but its audio track is published again under a new track SID. Bind to the participant SID, not the track SID.

On your backend (live-lib 1.9.4+). Define either or both hooks on your agent class to get the same participant in Python, for example to issue or revoke an output grant against the exact SID:

class MyAgent(LiveKitAgentMixin):
    def on_job_started(self, info):          # or async def
        # info: JobParticipantInfo(room_name, identity, sid, kind="agent", job_id)
        grant_output(info.room_name, info.identity, info.sid)

    def on_job_ended(self, info, reason):    # or async def; exactly once per job
        revoke_output(info.sid)

• on_job_started runs once per job, after the agent has joined the room and before the voice session starts.
• on_job_ended runs exactly once per job. reason is one of:
  • "disconnected": fired as soon as the room disconnects, before the session finishes shutting down;
  • "shutdown": the job was stopped, for example by worker.stop();
  • "error";
  • "ended".
• Restarts inside a job don't count. A voice session restart (for example, update_livekit_instruction()) does not fire either hook; identity and SID stay the same.
• Keep both hooks quick. They run on the job's event loop. An async def hook is limited to 5 seconds, and a hook that raises is logged without affecting the session.
• get_livekit_job_info() (on LiveKitAgentMixin) returns the current JobParticipantInfo, or None when no job is running.

on_session_ready() is unchanged and still takes no arguments.

LiveKitAgentMixin — wiring a flow into your own agent class

from live_lib.agent_runtime import LiveKitAgentMixin

class MyAgent(MyBaseAgentClass, LiveKitAgentMixin):
    ...

Mix this into your existing agent class to pick up the LiveKit-side integration (room registration, dispatch pairing) without restructuring your own agent hierarchy.

GPT-Live backend delegation — OpenAI's full-duplex voice model

LLM_PROVIDER on your agent class picks which realtime voice model LiveKitBridgeWorker connects to LiveKit — "gemini", "openai" (the original single-model Realtime API), or now "gpt_live": OpenAI's GPT-Live (gpt-live-1), a full-duplex model built to listen and speak at the same time instead of taking turns, which hands off the actual reasoning and tool-calling to a separate backend model instead of doing that itself.

class MyAgent(MyBaseAgentClass, LiveKitAgentMixin):
    LLM_PROVIDER = "gpt_live"
    VOICE = "marin"                          # any GPT-Live voice
    GPT_LIVE_BACKEND_MODEL = "gpt-6.1-sol"   # or "gpt-6-astra" — any OpenAI Responses-API model
    SYSTEM_FLOW = flow                       # or SYSTEM_INSTRUCTION — same as every other provider

That's the entire setup — nothing else about your LiveKitBridgeWorker/LiveKitAgentMixin wiring changes. A few things worth knowing:

• Delegation, not one model doing everything. GPT-Live itself only handles the live conversation — listening, speaking, deciding when to interrupt or wait. GPT_LIVE_BACKEND_MODEL is a separate OpenAI Responses-API model that does the actual reasoning and calls your tools. Contrast with LLM_PROVIDER = "openai", where one model does both.
• Backend instructions are optional. Skip GPT_LIVE_BACKEND_INSTRUCTIONS and the backend model reasons from the exact same SYSTEM_FLOW/SYSTEM_INSTRUCTION content already driving the conversation — set it only if you want the backend's reasoning instructions to actually differ from your flow's persona/content.
• Tools need no extra wiring. Whatever get_livekit_tools(bridge) returns (see function_tool/find_function_tools below) is available to the backend model exactly the way it is for any other provider.
• One API key for both halves. GPT-Live and its backend model both authenticate with the same openai_api_key you already pass to LiveKitBridgeWorker — no second key to manage.

Backend tuning, all optional: GPT_LIVE_BACKEND_TOOL_CHOICE, GPT_LIVE_BACKEND_PARALLEL_TOOL_CALLS, GPT_LIVE_BACKEND_REASONING_EFFORT (a plain string — "low"/"medium"/"high"/etc.), and GPT_LIVE_BACKEND_SERVICE_TIER. GPT_LIVE_SERVICE_TIER is different — it's the whole connection's own processing tier, so (unlike the others) it applies regardless of which delegation mode you use.

Client delegation — route to your own backend instead of OpenAI's Responses API

GPT_LIVE_DELEGATION = "client" hands delegated work to your own code — a custom agent/service, a local vision-language model, Gemini Robotics ER, anything you want driving the reasoning, instead of OpenAI's Responses API:

class MyAgent(MyBaseAgentClass, LiveKitAgentMixin):
    LLM_PROVIDER = "gpt_live"
    VOICE = "marin"
    GPT_LIVE_DELEGATION = "client"
    SYSTEM_INSTRUCTION = "..."

    async def on_gpt_live_delegation(self, delegation):
        # delegation.id, delegation.pending_transcript — the caller's current turn
        answer = await call_my_own_backend(delegation.pending_transcript)
        if self.is_current_gpt_live_delegation(delegation.id):  # not superseded meanwhile
            self.respond_to_gpt_live_delegation(answer, delegation_id=delegation.id)

on_gpt_live_delegation can be def or async def — returning a coroutine lets your handler await a real call to your own backend without blocking the voice session. Function tools don't work under client delegation — the delegation handoff above is the only path; use "responses" delegation instead if you need tool-calling.

Two more optional hooks fire automatically once your handler finishes or raises, for orchestration/telemetry: on_gpt_live_delegation_completed(delegation_id), on_gpt_live_delegation_error(delegation_id, exc).

Testing with a synthetic participant instead of a real client

If you're publishing microphone audio from a Python rtc.Room.connect() client — e.g. to test locally without a real browser/mobile participant — the track can show up as published and even subscribed while transcripts, on_gpt_live_delegation, and any raw-audio hooks stay silent. This isn't specific to live-lib or to client delegation — it's livekit-agents' own AgentSession, which only treats a participant as "the user" if its ParticipantKind is STANDARD, SIP, or CONNECTOR. A real browser/mobile client connecting through LiveKit's official SDKs always gets STANDARD automatically, so this normally never comes up — it only affects synthetic/test participants, depending on how their connection code is written. Neither /create-instance-issued tokens nor /test-token set a kind, so a plain rtc.Room.connect() with either should default to STANDARD and just work — if transcripts/delegation go silent for a synthetic test client specifically, check what kind it's actually connecting as before assuming a delegation-mode or audio-pipeline bug.

Stopping in-flight work when a delegation is superseded

is_current_gpt_live_delegation (shown above) checks after your handler finishes, so you don't speak a stale answer. For realtime interaction — e.g. a child changing their answer while your backend is still reasoning about the previous one — you often want to stop that work the moment it's superseded, not pay for it to finish anyway. on_gpt_live_delegation_superseded(old_id, new_id) fires exactly then, and cancel_gpt_live_delegation(delegation_id) actually aborts it:

def on_gpt_live_delegation_superseded(self, old_id, new_id):
    self.cancel_gpt_live_delegation(old_id)

The two aren't linked automatically — you might want the old work to finish anyway (for logging), just not spoken, which is what is_current_gpt_live_delegation already handles. Cancelling raises inside your handler at whatever it's currently awaiting, so a real in-flight call to your backend gets a genuine chance to abort, not just have its result thrown away afterward — and it does not trigger on_gpt_live_delegation_completed/_error, since a cancelled delegation is its own distinct outcome, not a success or a failure.

Injecting context while the conversation is running

Independent of which delegation mode you use, three methods push live context into GPT-Live — not just for answering a delegation, also for proactively informing an ongoing conversation:

self.append_gpt_live_thinking("the caller just picked up the red mug")   # silent, informs replies
self.append_gpt_live_instructions("confirm the address before booking")  # steers behavior, not spoken
self.respond_to_gpt_live_delegation("delivery moves to Friday")          # spoken aloud

Each accepts an optional delegation_id= (default None = session-wide rather than tied to one delegation).

Camera snapshot — for a vision-capable client-delegation backend

GPT-Live's own realtime session has no image/video input channel — there's nothing to "send a picture into." For a vision use case, get_latest_camera_frame() returns the most recently captured camera frame (raw RGB) for your on_gpt_live_delegation handler to send on to your own vision backend:

ENABLE_VIDEO_STREAM is about video coming into your agent, not video your agent sends out. It watches for a video track published by someone else in the room and captures frames from it. If you're using /session as your client, &enableCamera=1 (see Embedding in a native Android app below for the equivalent on native) is what publishes that track in the first place — turn that on in the browser/app and ENABLE_VIDEO_STREAM = True on your agent picks it up automatically, no extra wiring on either side. This is completely separate from any video your agent itself sends back (e.g. a talking-avatar) — that's the other direction, shown in /session's own video stage, and has nothing to do with ENABLE_VIDEO_STREAM.

async def on_gpt_live_delegation(self, delegation):
    frame = self.get_latest_camera_frame()  # None if video isn't enabled / no frame yet
    answer = await call_my_vision_backend(delegation.pending_transcript, frame)
    self.respond_to_gpt_live_delegation(answer, delegation_id=delegation.id)

Updated automatically whenever your agent has video enabled (ENABLE_VIDEO_STREAM = True) — nothing extra to turn on.

For reasoning about motion rather than a single still — e.g. judging a physical action over a few seconds — get_recent_camera_frames(seconds=3.0, fps=None) returns the same kind of frames over a window instead of just the latest one, oldest to newest:

async def on_gpt_live_delegation(self, delegation):
    frames = self.get_recent_camera_frames(seconds=3.0, fps=5.0)
    answer = await call_my_vision_backend(delegation.pending_transcript, frames)
    self.respond_to_gpt_live_delegation(answer, delegation_id=delegation.id)

fps, if given, subsamples toward that rate — it can't invent frames faster than your agent actually captures them (set VIDEO_FRAME_INTERVAL_SECONDS = 0.2 for ~5 fps). Up to the last 10 seconds are kept regardless of fps; asking for longer just returns what's still buffered. No video/clip encoding is built in — most vision APIs (including the ones this was built for) take multiple still images per request rather than a video file, so a plain list of frames is the more directly useful shape anyway.

User transcript hooks — run your own ASR/processing in parallel

GPT-Live transcribes the caller's speech automatically, no setup needed. Two optional hooks expose it directly, regardless of which delegation mode you're using — useful for running your own processing alongside GPT-Live's own transcription, e.g. for speech it handles less reliably (young children):

class MyAgent(MyBaseAgentClass, LiveKitAgentMixin):
    def on_gpt_live_user_transcript_delta(self, transcript, item_id=None, timestamp=None):
        ...  # partial, not yet final

    async def on_gpt_live_user_transcript_final(self, transcript, item_id=None, timestamp=None):
        await my_own_asr_pipeline(transcript)

Either hook can be def or async def — an async one runs as its own task, so it can await real work without blocking the voice session.

Raw user audio — for your own ASR pipeline

Sometimes a transcript isn't enough — if GPT-Live misreads something (young-child speech is a common case), you need the original audio, not just the text it produced. Set ENABLE_USER_AUDIO_CAPTURE = True (off by default — it's a second, independent subscription to the caller's mic track) and either stream it or pull a recent clip:

class MyAgent(MyBaseAgentClass, LiveKitAgentMixin):
    LLM_PROVIDER = "gpt_live"
    ENABLE_USER_AUDIO_CAPTURE = True
    USER_AUDIO_SAMPLE_RATE = 16000  # default; a standard ASR-friendly rate

    async def on_gpt_live_user_audio_frame(self, pcm, sample_rate, timestamp):
        await my_own_streaming_asr(pcm, sample_rate)

async def on_gpt_live_delegation(self, delegation):
    clip = self.get_recent_user_audio(seconds=3.0)  # None if nothing captured yet
    if clip:
        text = await my_own_asr_pipeline(clip.pcm_bytes, clip.sample_rate)

Audio is always captured mono; on_gpt_live_user_audio_frame fires for every frame (no throttling, unlike camera snapshots — dropping audio would corrupt your ASR input), while get_recent_user_audio returns the window already concatenated into one clip, ready to hand to a pipeline. This capture works regardless of which LLM_PROVIDER you're using — it's just a second subscriber to the same mic track the framework already feeds to whichever model you're running; only the hook's name follows this GPT-Live-based use case.

Correlating streams — a shared clock

CameraFrame.captured_at, the transcript hooks' timestamp, and the audio hook/clip's timestamps all come from the same clock, stamped at the moment each one is captured — so you can directly compare them (e.g. a transcript at timestamp=12.4 lines up with whatever get_recent_camera_frames() shows around captured_at≈12.4) to correlate speech like "this one" with pointing or object movement in the matching frames. These are not wall-clock/calendar timestamps and aren't comparable across a restart — only within one running session.

function_tool / find_function_tools — exposing Python methods as LLM tools

from live_lib.agent_runtime.tools import function_tool, find_function_tools

class MyToolProvider:
    @function_tool(description="Forward a complex query to a backing service.")
    async def ask_backend(self, query: str) -> str:
        return await self._bridge.forward_to_agent(query)

# Later, wherever your worker/agent expects a tool list:
tools = find_function_tools(MyToolProvider())

@function_tool marks a method as LLM-callable with a description the model uses to decide when to call it; find_function_tools discovers every decorated method on an instance and returns them as a ready-to-use tool list.

4. Minimal working example (copy, paste, run)

Everything above as one complete, runnable script — a tiny FlowGraph wired into LiveKitBridgeWorker, using Gemini. This is the fastest way to have something real to talk to via Test Your Agent: save it, fill in your Gemini API key, run it, then connect from that page using the same room name.

import time

from live_lib import FlowGraph
from live_lib.agent_runtime import LiveKitBridgeWorker

# 1. Design the conversation. A single looping node is enough for a first test -
# see "FlowGraph" above for branching to multiple nodes.
graph = FlowGraph(name="minimal_greeter", persona="Friendly, upbeat voice assistant.")

@graph.node("GREETING", objective="Greet the caller and see how you can help")
def greeting(n):
    n.instructions("Greet the caller warmly.", "Ask what they'd like help with today.")
    n.on(condition="caller says anything", to="GREETING")

graph.entry("GREETING")
flow = graph.compile()


# 2. Wire the compiled flow into an agent class. LiveKitBridgeWorker reads these
# attributes via getattr - there's no import/registration step beyond this.
# get_livekit_tools is optional (no function tools needed for a minimal
# example) - defining it as a no-op just avoids a harmless but confusing
# "object has no attribute 'get_livekit_tools'" warning in your logs.
class MinimalAgent:
    LLM_PROVIDER = "gemini"
    VOICE = "Puck"
    SYSTEM_FLOW = flow

    def get_livekit_tools(self, bridge):
        return []


AGENT_ID = "demo"
ROOM_NAME = "demo-room"  # must match the roomName you pass to /test-token

# 3. livekit_url/api_key/api_secret come from ~/.niuty/credentials.enc
# (Step 4 above) - never pass them directly.
worker = LiveKitBridgeWorker(
    openai_api_key="",  # unused for this Gemini-only example
    google_api_key="<your Gemini API key>",
    agent_name="minimal-test-agent",
)

# 4. Register the agent, then map the room name directly with notify_join -
# _entrypoint's first (and fastest) pairing check is exactly this mapping,
# so this makes pairing instant. Without it, LiveKitBridgeWorker instead
# spends up to 2 real minutes retrying prefix-based matching (default
# room_id_prefix="agent-", which "demo-room" doesn't start with) before
# falling back to the single-registered-agent path - confirmed live: every
# first connection sat silent for ~120s before the agent ever responded.
worker.register(AGENT_ID, agent_ref=MinimalAgent())
worker.notify_join(AGENT_ID, ROOM_NAME)
worker.start()
time.sleep(3)  # give the worker a moment to register with LiveKit before dispatching

# 5. Jobs are only ever explicitly dispatched, never automatic on room join -
# this must be the SAME room name you pass as `roomName` to /test-token.
worker.create_agent_dispatch(ROOM_NAME)

print(f"Agent running, dispatched to room '{ROOM_NAME}'. Ctrl+C to stop.")
while True:
    time.sleep(3600)

Requires pip install "live-lib[livekit]" (Step 5 above) and a Gemini API key. Leave the process running, then go to Test Your Agent and request a token for roomName=demo-room — you should hear MinimalAgent greet you within a few seconds of connecting.

5. Embedding in a native Android app

Instead of integrating LiveKit's Android SDK directly, load /session into a WebView inside your own app — it already handles the WebRTC connection, microphone, and (with &audioOnly=1) skips video entirely. A LiveKit data channel carries structured messages between your Python agent and your native code, through the page.

Architecture: your own backend mints a token server-side (SigV4-signed /test-token call, same as Step 2 above) — never ship your AWS credentials inside the app. Hand only the resulting short-lived token/URL down to the device, then load:

https://niuty.com/session?token=<token>&livekitUrl=<livekitUrl>&audioOnly=1

For a kiosk/embodied-agent setup publishing its own camera too (see Camera snapshot above), add &enableCamera=1 to the same URL — everything below already handles both cases, since the permission grant only ever grants what the page actually asked for.

WebView setup (Kotlin):

webView.settings.javaScriptEnabled = true
webView.addJavascriptInterface(NiutyBridgeInterface(webView), "AndroidApp")

// Grant microphone (and, if the URL has &enableCamera=1, camera) access - LiveKit's WebRTC
// connection needs it, and a WebView never prompts for it on its own the way a normal browser
// tab does. The allow-list below covers both cases without needing two separate examples - if
// the page never requested video (no enableCamera=1), RESOURCE_VIDEO_CAPTURE is simply never in
// request.resources, so the filter is a no-op for it.
webView.webChromeClient = object : WebChromeClient() {
    override fun onPermissionRequest(request: PermissionRequest) {
        val allowed = setOf(PermissionRequest.RESOURCE_AUDIO_CAPTURE, PermissionRequest.RESOURCE_VIDEO_CAPTURE)
        request.grant(request.resources.filter { it in allowed }.toTypedArray())
    }
}

webView.loadUrl("https://niuty.com/session?token=$token&livekitUrl=$livekitUrl&audioOnly=1")

Also request RECORD_AUDIO at the Android permission level as usual (and CAMERA too, if using &enableCamera=1), and consider running the session inside a Foreground Service if it needs to survive your app being backgrounded — both are ordinary Android/WebRTC concerns, not specific to this integration:

<uses-permission android:name="android.permission.RECORD_AUDIO" />
<uses-permission android:name="android.permission.CAMERA" /> <!-- only if using &enableCamera=1 -->

Receiving messages from your Python agent — implement onDataReceived on the object you registered as "AndroidApp":

class NiutyBridgeInterface(private val webView: WebView) {
    @JavascriptInterface
    fun onDataReceived(jsonString: String) {
        val msg = JSONObject(jsonString)
        when (msg.getString("type")) {
            "MOVE" -> handleMove(msg.getString("direction"))
            // ...
        }
    }
}

Called once per inbound data-channel message (topic "logic") — the internal KEEP_ALIVE heartbeat is filtered out before it reaches you, so every call here is real application data.

Sending messages to your Python agent — call window.NiutyBridge.sendData(...) via evaluateJavascript:

val payload = JSONObject(mapOf("type" to "BUTTON_PRESSED", "button" to "call_staff"))
webView.post {
    webView.evaluateJavascript("window.NiutyBridge.sendData($payload)", null)
}

Server side (Python, live-lib) — implement on_rt_data_message to receive, call send_data_to_livekit (from LiveKitAgentMixin) to send:

class MyAgent(LiveKitAgentMixin):
    def on_rt_data_message(self, msg: dict) -> None:
        if msg.get("type") == "BUTTON_PRESSED":
            self.handle_button(msg.get("button"))

    def notify_status(self, status: str) -> None:
        self.send_data_to_livekit({"type": "STATUS", "status": status})

Envelope convention: {"type": "<YOUR_TYPE>", ...} — freeform beyond type, entirely yours to define on both ends. See LiveKitAgentMixin above for how this mixin fits into your own agent class.

6. Data handling and privacy

What each part of the system does with your users' audio, video, transcripts and other data, so you can assess it against your own requirements. This describes system behavior only — whether your use case meets a particular law or policy is for your own review. Your responsibilities, including for any use involving children, are set out in the Terms of Service.

live-lib (runs on your machine)

• No recording or upload. live-lib 1.9.3+ starts every session with recording explicitly disabled, so the LiveKit Agents SDK never uploads audio, transcripts, traces or logs anywhere. Don't set LIVEKIT_OBSERVABILITY_URL in your worker's environment.
• Nothing written to disk unless you opt in (debug_dump_path=, which writes only the system prompt). Camera frames and raw user audio exposed by the GPT-Live APIs are short in-memory buffers only.
• Your logs. By default the worker prints conversation content to stdout for debugging ([TRANSCRIPT IN]/[TRANSCRIPT OUT], spoken text, tool queries/results). Pass log_conversation_content=False to LiveKitBridgeWorker to log only lengths instead (<redacted len=42>). Takes effect at construction, before any session starts. Output from your own hooks is your own code and isn't affected.
• Model providers. Your worker calls OpenAI/Google directly with your own API key. niuty never receives or relays that traffic, so your provider account's own data controls apply as you configure them.
• Shutdown. worker.stop() ends every active session, disconnects from LiveKit and returns True once the worker has exited.

To check: confirm LIVEKIT_OBSERVABILITY_URL is unset, and that your worker only connects to your own *.rt.niuty.com instance and your model provider.

Your niuty instance (the LiveKit relay)

• Media and messages are relayed, not stored. Audio, video, data messages, chat and transcriptions pass between participants and are never recorded or written to disk — there is no recording service on the instance.
• Server logs record connection details only: room name, participant identity, IP addresses and client device info (SDK, OS, device model). Display names, metadata, attributes and message content are not logged. Don't put personal data (e.g. a person's name) in room names or participant identities — use opaque IDs.
• Retention. Logs stay on your instance only, and are deleted together with it when the instance is deprovisioned (the server, its IP address and its DNS record are all removed).