How many agent concurrently i can run in free version of livekit cloud?

Thanks for reaching out, Blue. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

I built two different agent and running it via self hosted in livekit cloud they are not deployed in livekit cloud. one is for sip outbound and one is for web call but when i call for outbound call using sip then web call agent also speaking in same call. what can be a issue?

Thanks — I’m sorry you’re seeing the web agent join and speak on SIP outbound calls. That usually means both agents are being dispatched to the same room. I’ll walk you through the most likely causes and the simplest fixes, and then ask a couple quick questions to pinpoint the problem. What I’m responding to: your two agents (SIP outbound and web call) are self-hosted and not deployed to LiveKit Cloud, but when you place a SIP outbound call the web agent also joins/speaks on that call. Likely cause • Both agents are being dispatched into the same room (auto-dispatch or overlapping dispatch rules), so the web agent is joining SIP call rooms it shouldn’t. Quick checks and fixes 1. Give each agent a unique name • Ensure each agent uses a unique

agent_name

(for example

sip-agent

and

web-agent

). When an agent has an

agent_name

, it will not be auto-dispatched to every room — it will only join when explicitly dispatched. 1. Make SIP dispatch rules explicit • Create a dispatch rule for SIP calls that explicitly lists only the SIP agent in the room config. That prevents the web agent being selected for SIP rooms. • Example concept: rooms created for SIP should use a prefix like

sip-call-

and the dispatch rule should reference

sip-agent

only. 1. Avoid auto-dispatch for the web agent • If the web agent should only serve browser sessions, make sure it is not in any SIP-related dispatch rules. Name it and dispatch it only for web rooms. 1. Use distinct room naming patterns • Ensure SIP-generated rooms and web session rooms follow different naming patterns (e.g.,

sip-call-*

vs

web-session-*

) so dispatch rules can target them cleanly. 1. Check agent logs to confirm what’s happening • Inspect each agent’s logs to see which rooms they join and when they’re dispatched. Look for entries showing the web agent joining the SIP room and the dispatch source. • Example commands (if you have a logging/CLI): check

sip-agent

logs and

web-agent

logs for room join events. What I need from you to narrow this down • Do your agents currently have

agent_name

values? If so, what are they? • How are SIP rooms being created/named (room name or prefix)? • Do you have any dispatch rules configured (SIP dispatch rules or general agent dispatch configuration)? If yes, please paste the relevant rule(s) (redact secrets). • Any log lines showing the web agent joining the SIP room would help (timestamp + join event). If you share those details I’ll: • Point to the exact configuration change needed, or • Provide a sample dispatch rule you can apply to isolate the SIP agent, and • Suggest the exact log entries to look for to verify the fix You’re close — this is almost always a dispatch/room-naming overlap and is straightforward to fix once we see the agent names and dispatch rules. Sources: Agents telephony integration | LiveKit Documentation | Agent dispatch | LiveKit Documentation

def prewarm(proc): proc.userdata["vad"] = silero.VAD.load( min_speech_duration=0.08, #0.1 min_silence_duration=0.65, activation_threshold=0.3, max_buffered_speech=90 ) class Assistant(Agent): base_instructions = """""" super().__init__(instructions=base_instructions) async def on_enter(self): # time.wait(2) await self.session.say(text="hello. am i audiable?", allow_interruptions=False) server = AgentServer() def prewarm(proc: JobProcess): proc.userdata["vad"] = silero.VAD.load() server.setup_fnc = prewarm async def entrypoint(ctx: JobContext) -> None: session = AgentSession( llm=openai.LLM(model="gpt-4o-mini"), stt=deepgram.STT(model="nova-3"), tts=deepgram.TTS(model="aura-2-luna-en"), vad=ctx.proc.userdata["vad"], turn_detection="vad" ) agent = Assistant( ) await session.start( ) if name == "__main__": cli.run_app(server) this is my sample code for web calling agent where i can define the name of agent and after running the agent how should i connect to it? what is best practice

Thanks — I’ll help you define the agent name in your sample code, show how to connect to it after running, and list the best practices tailored to your snippet. What I’m responding to: your web-calling agent code and the question “where do I define the agent name and how do I connect to it after running?” 1) Where to set the agent name • Set the

agent_name

on the

@server.rtc_session(...)

decorator. That gives the agent a unique identity and disables automatic dispatch so it will only join rooms when explicitly dispatched. • Example (showing the minimal, corrected structure using your components):

from livekit.agents import AgentServer, JobContext, AgentSession, Agent, cli
from livekit.plugins import silero, openai, deepgram

server = AgentServer()

def prewarm(proc):
    proc.userdata["vad"] = silero.VAD.load(
        min_speech_duration=0.08,
        min_silence_duration=0.65,
        activation_threshold=0.3,
        max_buffered_speech=90
    )

server.setup_fnc = prewarm

class Assistant(Agent):
    def __init__(self) -> None:
        base_instructions = "You are a helpful voice AI assistant."
        super().__init__(instructions=base_instructions)

    async def on_enter(self):
        await self.session.say(text="hello. am i audible?", allow_interruptions=False)

# Give the agent a name here:
@server.rtc_session(agent_name="web-assistant-agent")
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        llm=openai.LLM(model="gpt-4o-mini"),
        stt=deepgram.STT(model="nova-3"),
        tts=deepgram.TTS(model="aura-2-luna-en"),
        vad=ctx.proc.userdata["vad"],
        turn_detection="vad"
    )

    agent = Assistant()
    # start the session and connect the agent to the room
    await session.start(agent=agent, room=ctx.room)
    await ctx.connect()

if __name__ == "__main__":
    cli.run_app(server)

2) How to connect to the agent after running it • With

agent_name

set, your agent will wait to be dispatched. You can connect/dispatch it in two common ways: ◦ Option A — Dispatch via room token (recommended for web clients): ▪︎ Create an access token / room configuration that includes the agent in

RoomConfiguration.agents

so the agent is dispatched automatically when a user joins with that token. ▪︎ This is useful when you want a user’s join to automatically spawn a specific agent for that room. ◦ Option B — Dispatch via API/CLI or explicit server-side dispatch: ▪︎ Use the agent dispatch API / control plane (or CLI tooling) to create a job for a specific

agent_name

and room when you want the agent to join (for example, when a SIP call is created). ▪︎ Use clear room naming conventions so the dispatch rule targets only SIP rooms or web rooms (e.g.,

sip-call-*

vs

web-session-*

). ◦ For development/testing: ▪︎ Run your agent process (e.g.,

python agent.py dev

or the equivalent command used in your environment). ▪︎ Connect a web client (LiveKit JS) to the same room with a token that either includes the agent dispatch or triggers dispatch server-side. 3) Best practices to avoid agent overlap (your original issue where web agent joined a SIP call) • Always assign unique

agent_name

values (e.g.,

sip-agent

,

web-agent

). • Use explicit dispatch rules that list which agent(s) should join SIP rooms vs web rooms. Don’t rely on auto-dispatch for production. • Use distinct room naming patterns (e.g., prefix SIP rooms with

sip-call-

and web rooms with

web-session-

) so dispatch rules can be precise. • Reuse heavy models (VAD) via

server.setup_fnc

to avoid reloading for every job — you already have this pattern, keep one

prewarm

definition. • Inspect agent logs to confirm which room each agent joined if behavior looks wrong. • Consider

preemptive_generation=True

and

turn_detection="vad"

for responsiveness and stable turn-taking if your agent framework supports those options. 4) Quick diagnostics if you still see both agents in the same call • Verify both agents do not share the same

agent_name

. • Show me your dispatch rule(s) or room creation code (redact secrets). If a dispatch rule doesn’t explicitly target

sip-agent

, the web agent may be auto-selected. • Provide one sample log line showing the web agent joining the SIP room (timestamp + join event) and I’ll point to the exact dispatch misconfiguration. If you want, paste your current dispatch rule or the token generation code you use for the caller and I’ll give the exact snippet you need to isolate the SIP agent from the web agent. Sources: Simple Content Filter | LiveKit Documentation | Agent dispatch | LiveKit Documentation | Introduction | LiveKit Documentation | LLM Metrics | LiveKit Documentation | Simple Call Answering Agent | LiveKit Documentation

can you give me code snippet for dispatch request of web call agent?

Thanks — I’ll give you concise examples for creating a dispatch request that assigns your named web call agent to a room. I’ll include a Python example (using a typical LiveKit API client / token creation flow) and a curl example that calls the dispatch API directly. Replace placeholders (API URL, API key/secret, room name, agent name, and tokens) with your values. What this does: the request creates a job that dispatches the specific agent (by

agent_name

) into the target room so the web agent will join only when you want it to. 1) Python example — generate a dispatch job via the REST API • This sends a POST to the LiveKit control/dispatch endpoint to assign

web-assistant-agent

to

web-session-123

.

import requests
import os
import time
import jwt  # PyJWT

LIVEKIT_API_URL = "https://<YOUR_CONTROL_PLANE_HOST>/v1/projects/<PROJECT_ID>"
API_KEY = os.environ["LIVEKIT_API_KEY"]
API_SECRET = os.environ["LIVEKIT_API_SECRET"]

def make_jwt(path: str, method: str = "POST", exp_seconds: int = 60):
    now = int(time.time())
    payload = {
        "iss": API_KEY,
        "exp": now + exp_seconds,
        "room": "*",      # adjust if you want to scope token
        "kind": "server"  # server token for control operations
    }
    token = jwt.encode(payload, API_SECRET, algorithm="HS256")
    return token

def dispatch_agent(room_name: str, agent_name: str):
    path = f"/dispatch"
    url = LIVEKIT_API_URL + path
    token = make_jwt(path)

    body = {
        "room": room_name,
        "agents": [
            {"agent_name": agent_name}
        ],
        # optional: metadata, priority, or other dispatch fields
        "metadata": {"reason": "user_requested_web_agent"}
    }

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    resp = <http://requests.post|requests.post>(url, json=body, headers=headers)
    resp.raise_for_status()
    return resp.json()

if __name__ == "__main__":
    room = "web-session-123"
    agent = "web-assistant-agent"
    result = dispatch_agent(room, agent)
    print("Dispatch result:", result)

Notes: • Use a server-scoped token (signed with your API secret). The exact claims needed may vary with your LiveKit control plane; the example shows a simple server token pattern. • The

/dispatch

path may be under a project-specific base URL. Confirm your control-plane base path and adjust. 2) curl example — direct dispatch request • Equivalent raw HTTP request to create a dispatch job.

# Create a short-lived JWT (server token) and set it to LIVEKIT_JWT
# Then call the dispatch endpoint. Replace placeholders.

curl -X POST "https://<YOUR_CONTROL_PLANE_HOST>/v1/projects/<PROJECT_ID>/dispatch" \
  -H "Authorization: Bearer <SERVER_JWT>" \
  -H "Content-Type: application/json" \
  -d '{
    "room": "web-session-123",
    "agents": [
      { "agent_name": "web-assistant-agent" }
    ],
    "metadata": { "reason": "user-initiated" }
  }'

3) Token-based dispatch (alternative) — include agent in RoomConfiguration when issuing a join token • If you prefer the agent to be dispatched automatically when a user joins, include the agent in the room configuration inside the access token you give the client. High-level idea (pseudo): • When generating a room join token for a user, embed a room config that requests

web-assistant-agent

be dispatched for that room. The LiveKit server will then create the job automatically when that user connects. 4) Best practices and tips • Use distinct room name patterns to target dispatches (e.g.,

web-session-*

). • Only dispatch the agent by

agent_name

to avoid auto-dispatch collisions. • Verify agent logs after dispatch to confirm the agent joined the intended room. • If you also have SIP dispatch rules, ensure SIP dispatch rules don’t include the web agent. If you want, tell me: • which method you prefer (REST dispatch vs token-based), and • the control-plane host or SDK you are using, and I’ll provide a tailored snippet (e.g., exact LiveKit control-plane endpoint and JWT claims format for your setup).

I can not join web call agent without token, right?

Short answer: both clients and agents need a valid room token to join a LiveKit room. You cannot join a room (as a participant or as an agent) without a token. Why: • LiveKit requires an access token for authentication/authorization when connecting to a room. Tokens ensure only permitted identities can join and allow you to embed room configuration (including agent dispatch) or permissions for that connection. What that means for your web-call agent and web clients • Web client (browser): must connect with a room join token (access token) generated by your server. • Agent process: when dispatched into a room, the agent is still created as a LiveKit participant and the control plane will allocate it a job and the agent process will use a server-scoped token or job context to join the room. ◦ If you dispatch via the control/dispatch API or include the agent in the RoomConfiguration in a user token, the control plane issues the job and the agent joins using credentials provided by the agent runtime (handled by the server/agent framework). You don’t manually share a browser token with the agent. How to handle tokens in practice (recommended) • Always generate tokens server-side (never in client-side code). • For web users: generate a join token scoped to the room + any permissions, then hand it to the browser client to call

room.connect(LIVEKIT_URL, token)

. • For dispatching an agent: ◦ Option A (recommended): Include the agent in the

RoomConfiguration.agents

when generating the user’s room token. When the user connects with that token, the control plane will create the dispatch job and the agent runtime will join the room automatically. ◦ Option B: Call the control-plane dispatch endpoint to create a dispatch job that references

agent_name

. The control plane will start the job and the agent runtime will join the room using server credentials (this is done by your backend/agent framework, not by the browser). Security tips • Use short-lived tokens for clients. • Sign tokens server-side using your API key/secret. • Scope tokens to a specific room and identity where possible. If you want, I can: • Provide a short code snippet for generating a room join token (server-side) for a web client, and • Show how to include the agent in the

RoomConfiguration

so the agent is dispatched automatically when the user joins. Which language/SDK do you prefer (Python, Node, Go)?

token = AccessToken().with_identity(call_id).with_name("call_id").with_grants(VideoGrants( room=room_name, room_join=True, can_subscribe=True, can_publish=True, agent="web-interview-agent-dev" )) this how i am generating token.. is it right? I need joiner(user) to access chat, camera, mic all things. User have to turn on the camera.

Thanks — you're very close. I'll confirm what's wrong in your snippet, show the corrected token code you should use, and highlight the exact grants needed so the user can publish camera/mic and use chat. I'll also show how to auto-dispatch your web agent. What I’m responding to: your token creation snippet and the requirement that the joiner (user) can turn on camera, mic, and use chat — plus automatically dispatch the agent. 1) What’s incorrect in your snippet • Putting

agent="web-interview-agent-dev"

inside

VideoGrants

is not valid.

VideoGrants

controls participant permissions only; it does not dispatch agents. The

agent

field will be ignored and will not dispatch the agent. 2) Correct approach — token that both allows a user to publish media/data and requests agent dispatch • Use

VideoGrants

for room/permission grants. • Use

with_room_config(...)

and a

RoomAgentDispatch

entry to request the agent be dispatched for that room. 3) Example (Python) — corrected token generation • This token: ◦ Allows the user to join the room and publish camera/mic ◦ Enables subscribing to others ◦ Enables data messages (chat) ◦ Requests that LiveKit dispatch

web-interview-agent-dev

when the room is created/joined

from livekit.api import AccessToken, VideoGrants, RoomConfiguration, RoomAgentDispatch

# replace call_id, room_name appropriately
token = (
    AccessToken()
    .with_identity(call_id)
    .with_name("caller_name")  # optional display name
    .with_grants(
        VideoGrants(
            room_join=True,
            room=room_name,
            can_publish=True,        # allows camera/mic publishing
            can_subscribe=True,      # allows receiving others' tracks
            can_publish_data=True    # allows sending/receiving data (chat)
        )
    )
    .with_room_config(
        RoomConfiguration(
            agents=[
                RoomAgentDispatch(
                    agent_name="web-interview-agent-dev",
                    metadata='{"reason":"auto-dispatch-for-web"}'  # optional
                )
            ]
        )
    )
    .to_jwt()
)

4) Notes and best practices • can_publish enables the user to publish both audio and video; the user still must enable camera/mic in the browser (via getUserMedia). • can_publish_data=True enables data packets used for chat messages. • Generate tokens server-side and use short lifetimes for user tokens. • Distinctly name agents (e.g.,

web-interview-agent-dev

) and ensure your agent runtime is started with that

agent_name

(via

@server.rtc_session(agent_name=...)

). • If you prefer not to auto-dispatch via token, you can dispatch the agent later using the control-plane dispatch API referencing

agent_name

and room. 5) Quick checklist to verify end-to-end • Server generates the token above and hands it to the browser client. • Browser uses that token to call

room.connect(LIVEKIT_URL, token)

and then calls getUserMedia to enable camera/mic. • LiveKit control plane sees the RoomConfiguration in the token and creates a dispatch job for

web-interview-agent-dev

. • Your agent runtime (running and registered with that agent_name) receives the job and joins the room as a participant. If you want, I can: • Provide a small frontend snippet showing how the browser should request and use the token and enable camera/mic, or • Show the control-plane dispatch API example if you prefer server-initiated dispatch instead of token-based dispatch — which would you prefer? Sources: Agent dispatch | LiveKit Documentation

now i want to give metadata with it.

Great — you can attach metadata to the agent dispatch inside the room configuration when you generate the join token. The agent will receive that metadata in its JobContext (so your agent code can read it from

ctx.job.metadata

or similar). You can also include custom metadata in the token for the participant if you need the server or other services to see it. Here’s a concise Python example that: • Grants the user full publish/subscribe/data rights • Adds an agent dispatch with custom metadata • Also adds optional participant-level metadata

from livekit.api import AccessToken, VideoGrants, RoomConfiguration, RoomAgentDispatch

# replace values with your variables
call_id = "caller-123"
room_name = "web-session-123"

token = (
    AccessToken()
    .with_identity(call_id)
    .with_name("Caller Name")
    .with_grants(
        VideoGrants(
            room_join=True,
            room=room_name,
            can_publish=True,
            can_subscribe=True,
            can_publish_data=True,
        )
    )
    .with_metadata('{"caller_role":"interviewer","topic":"engineering"}')  # optional participant metadata
    .with_room_config(
        RoomConfiguration(
            agents=[
                RoomAgentDispatch(
                    agent_name="web-interview-agent-dev",
                    metadata='{"user_id":"%s","session_type":"web","priority":"normal"}' % call_id
                )
            ]
        )
    )
    .to_jwt()
)

Notes: • Agent dispatch metadata is the string in

RoomAgentDispatch.metadata

. Your agent receives it with the job context and can parse JSON there. • Participant metadata (optional) is set via

with_metadata(...)

on the token; it’s separate from agent metadata. • Keep metadata JSON small and safe — avoid embedding secrets. • If your agent expects structured fields, parse the JSON in your agent’s

JobContext

(e.g.,

ctx.job.metadata

). • Confirm your agent runtime reads the job metadata in its entrypoint; for example: ◦ ctx.proc.userdata or ctx.job.metadata (depending on agent framework) — inspect the job context in your agent logs to find the exact field. Would you like a matching agent-side snippet showing how to read and parse that metadata in your entrypoint?

how can i end the meeting?

Good question — there are two common ways to "end a meeting" depending on what you want to achieve. I’ll summarize both and give short examples so you can pick the right one. What I’m responding to: you asked how to end the meeting (disconnect everyone / stop the room). 1) Delete the room (recommended — disconnects everyone and stops the meeting) • Deleting the room will immediately disconnect all participants and stop related resources. • Use the LiveKit server/control API or SDK from your backend (server-side only). Python (async example using LiveKit API client):

from livekit import api
import os

livekit_url = os.getenv("LIVEKIT_URL")
api_key = os.getenv("LIVEKIT_API_KEY")
api_secret = os.getenv("LIVEKIT_API_SECRET")

client = api.LiveKitAPI(livekit_url, api_key, api_secret)
await client.room.delete_room(api.DeleteRoomRequest(room="room_name"))

Node (example using server SDK):

const { RoomServiceClient } = require("livekit-server-sdk");

const roomService = new RoomServiceClient(
  process.env.LIVEKIT_URL,
  process.env.LIVEKIT_API_KEY,
  process.env.LIVEKIT_API_SECRET
);

await roomService.deleteRoom("room_name");

curl (HTTP):

curl -X POST "https://<LIVEKIT_CONTROL_HOST>/livekit-api/rooms.delete" \
  -H "Authorization: Bearer <SERVER_JWT_OR_API_TOKEN>" \
  -H "Content-Type: application/json" \
  -d '{"room": "room_name"}'

• Use the exact endpoint and auth required by your LiveKit deployment (example endpoints vary by setup). 2) Have participants leave / end locally (client-side) • If you want just the current user to leave the meeting, call the client’s disconnect method (web SDK): ◦ JS:

room.disconnect()

— this only removes that participant. • Useful when you want the meeting to continue for others but a specific participant to exit. 3) End agent participation only (agent shutdown) • If you only want to remove an agent from a room (not delete the room), shut down the agent session:

await session.shutdown(drain=True)  # Python agent runtime

• This closes the agent gracefully but does not end the meeting for other participants. Which should you use? • To fully end the meeting for everyone: delete the room via the server-side API. • To remove a single participant (user or agent): use client disconnect or agent shutdown respectively. If you want, tell me: • Do you want to end the meeting for everyone or just remove the agent / a single participant? • Which environment/SDK are you using (Python backend, Node backend, or plain HTTP curl)? I’ll give the exact endpoint and a focused snippet for your environment. Sources: Why is there extra audio at the end of my recordings of an agent session? |... | Job lifecycle | LiveKit Documentation

I am using python sdk. Now i want to check if connected user has turned on their camera or not?

Good question — you can check whether a connected user has their camera on by inspecting their published video track(s) and the track mute state. Below are concise Python examples (polling and event-driven) using the LiveKit Python SDK patterns so you can pick what fits your agent or backend code. What I’m responding to: how to detect if a connected user has their camera turned on (publish camera) using the Python SDK. 1) Basic check (snapshot) • A participant has their camera "on" when they have a published video track that is not muted. Example (poll current participants):

from livekit import rtc

def is_camera_on_for_participant(participant: rtc.RemoteParticipant) -> bool:
    for pub in participant.track_publications.values():
        if pub.kind == rtc.TrackKind.KIND_VIDEO:
            # If a video track is published and not muted -> camera is on
            return not pub.is_muted
    return False

# Usage: iterate remote participants in the room
for p in room.remote_participants.values():
    print(f"{p.identity}: camera {'on' if is_camera_on_for_participant(p) else 'off'}")

2) Event-driven approach (recommended for realtime) • Listen for

track_subscribed

,

track_unsubscribed

,

track_muted

, and

track_unmuted

events to update UI or state whenever camera status changes. Example event handlers:

# Called when a remote track is subscribed (received)
@room.on("track_subscribed")
def on_track_subscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
    if publication.kind == rtc.TrackKind.KIND_VIDEO:
        status = "on" if not publication.is_muted else "off"
        print(f"{participant.identity} camera subscribed, status: {status}")

# Called when a remote track is unsubscribed (track removed)
@room.on("track_unsubscribed")
def on_track_unsubscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
    if publication.kind == rtc.TrackKind.KIND_VIDEO:
        print(f"{participant.identity} camera unsubscribed (off)")

# Called when a published track is muted (user turned camera off)
@room.on("track_muted")
def on_track_muted(publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
    if publication.kind == rtc.TrackKind.KIND_VIDEO:
        print(f"{participant.identity} turned camera off")

# Called when a published track is unmuted (user turned camera on)
@room.on("track_unmuted")
def on_track_unmuted(publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
    if publication.kind == rtc.TrackKind.KIND_VIDEO:
        print(f"{participant.identity} turned camera on")

3) Notes and edge cases • If no video track is published for a participant, treat camera as off. •

is_muted

indicates the participant disabled the camera even if the track is still present (common with "stop video" actions). • Screen-share vs camera: a user may publish a video track for screen sharing. Check

publication.source

if you need to distinguish

camera

vs

screen_share

. • Network or permission issues may temporarily stop publishing — handle transient disconnects accordingly. 4) Recommended pattern • Use the event-driven handlers to keep UI/state up-to-date. • Use the snapshot function when you need an immediate check (e.g., on page load or agent join). • Log track subscription/mute events during debugging to verify both client and server behavior. If you want, tell me: • Are you running this check inside an agent job (JobContext) or in a backend service? I can provide the exact import paths and a small snippet tailored to that environment. Sources: client-sdk-android | livekit/livekit_composite | DeepWiki | livekit/livekit_composite | DeepWiki | Vision | LiveKit Documentation

I want my code to continuously checking for camera if camera is off then it should tell that please turn on the camera.

Great — I’ll give you a clear, practical approach and code you can drop into your Python agent or backend so it continuously monitors participants’ camera state and prompts them to turn it on when it’s off. What I’m responding to: you want continuous checking for camera state and to notify the participant with a message when their camera is off. Two reliable approaches (recommended combination) • Event-driven: listen to track events (

track_subscribed

,

track_unsubscribed

,

track_muted

,

track_unmuted

) to update state instantly. • Periodic polling: run a short background loop that checks participants’ current camera state in case events are missed or participants never published a camera at all. Below is a concise example that combines both. It assumes you’re running inside an agent job or a server process with access to the LiveKit room object (adjust imports/names to your environment). 1) Helper functions and state

import asyncio
from typing import Dict
from livekit import rtc

# keep last-known camera status per participant
camera_status: Dict[str, bool] = {}  # identity -> True if camera on, False if off

def participant_has_camera_on(participant: rtc.RemoteParticipant) -> bool:
    for pub in participant.track_publications.values():
        if pub.kind == rtc.TrackKind.KIND_VIDEO:
            return not pub.is_muted
    return False

2) Event handlers (update state immediately)

def setup_event_listeners(room: rtc.Room):
    @room.on("track_subscribed")
    def on_track_subscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
        if publication.kind == rtc.TrackKind.KIND_VIDEO:
            camera_status[participant.identity] = not publication.is_muted
            # Optionally notify here if off
            if not camera_status[participant.identity]:
                asyncio.create_task(prompt_turn_on_camera(participant))

    @room.on("track_unsubscribed")
    def on_track_unsubscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
        if publication.kind == rtc.TrackKind.KIND_VIDEO:
            camera_status[participant.identity] = False
            asyncio.create_task(prompt_turn_on_camera(participant))

    @room.on("track_muted")
    def on_track_muted(publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
        if publication.kind == rtc.TrackKind.KIND_VIDEO:
            camera_status[participant.identity] = False
            asyncio.create_task(prompt_turn_on_camera(participant))

    @room.on("track_unmuted")
    def on_track_unmuted(publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
        if publication.kind == rtc.TrackKind.KIND_VIDEO:
            camera_status[participant.identity] = True
            # Optionally send a confirmation message or clear prompts

3) Periodic polling task (covers missed events / never-published cameras)

async def polling_camera_check(room: rtc.Room, interval_seconds: int = 10):
    while True:
        for participant in list(room.remote_participants.values()):
            current = participant_has_camera_on(participant)
            prev = camera_status.get(participant.identity)
            if prev is None:
                # initialize state
                camera_status[participant.identity] = current
                if not current:
                    asyncio.create_task(prompt_turn_on_camera(participant))
            elif prev != current:
                camera_status[participant.identity] = current
                if not current:
                    asyncio.create_task(prompt_turn_on_camera(participant))
                else:
                    # optionally clear previous prompt or notify camera is on
                    asyncio.create_task(notify_camera_on(participant))
        await asyncio.sleep(interval_seconds)

4) Prompting the user • How you prompt depends on your architecture: ◦ If you can publish a data message (chat) to the participant, send a data packet saying “Please turn on your camera.” ◦ If you have an agent voice channel, you can use

session.say(...)

or agent TTS to speak to that participant. ◦ If you control the frontend, send a data message that the client reads to show a UI prompt. Example: send a data message to the participant (pseudo-code):

async def prompt_turn_on_camera(participant: rtc.RemoteParticipant):
    # room is the connected Room instance
    try:
        message = "Please turn on your camera so we can continue."
        # publish a direct data message (reliable/unreliable depending on API)
        # Replace with your SDK method for sending data messages to a specific participant
        await room.local_participant.publish_data(message.encode("utf-8"), kind=rtc.DataPacketKind.RELIABLE, destination=[participant.sid])
    except Exception as e:
        # fallback: log or use agent TTS to speak to the room
        print(f"failed to prompt {participant.identity}: {e}")

async def notify_camera_on(participant: rtc.RemoteParticipant):
    try:
        message = "Thanks — camera is on."
        await room.local_participant.publish_data(message.encode("utf-8"), kind=rtc.DataPacketKind.RELIABLE, destination=[participant.sid])
    except Exception:
        pass

5) Bringing it together inside your agent/entrypoint

async def entrypoint(ctx: JobContext):
    room = ctx.room  # adjust to your context
    setup_event_listeners(room)
    # start polling in background
    asyncio.create_task(polling_camera_check(room, interval_seconds=10))
    # continue your agent/session lifecycle...

6) UX considerations and best practices • Don’t spam: rate-limit prompts for the same participant (e.g., wait 30–60s between messages). Use timestamps per participant to enforce cooldowns. • Respect privacy: ensure prompts are friendly and not intrusive. • Use data messages for UI prompts; use TTS only if voice prompts are acceptable. • Distinguish camera vs screen-share using

publication.source

if you need to treat them differently. • Test with unstable network conditions to ensure prompts are not sent erroneously on transient drops. Would you like: • A version with built-in cooldown logic (so each participant is only prompted once per minute)? • A ready-to-run agent-side snippet that includes imports and exact LiveKit SDK method names for your environment?

{"pid": 1873460, "job_id": "AJ_DHtVeQAv9FJ5", "room_id": "RM_JEwW7fsPPuVU"} Traceback (most recent call last): File "/home/bdcbeta/Desktop/t/lib/python3.10/site-packages/livekit/agents/utils/log.py", line 16, in async_fn_logs return await fn(*args, **kwargs) File "/home/bdcbeta/Desktop/t/lib/python3.10/site-packages/livekit/agents/voice/agent_activity.py", line 466, in _traceable_on_enter await self._agent.on_enter() File "/home/bdcbeta/Desktop/AI_/interview-scheduler-prod/web_call_agent.py", line 99, in on_enter video_tracks = [publication.track for publication in list(remote_participant.track_publications.values()) if publication.track.kind == rtc.TrackKind.KIND_VIDEO] File "/home/bdcbeta/interview-scheduler-prod/web_call_agent.py", line 99, in listcomp video_tracks = [publication.track for publication in list(remote_participant.track_publications.values()) if publication.track.kind == rtc.TrackKind.KIND_VIDEO] AttributeError: 'NoneType' object has no attribute 'kind'

Thanks — that traceback makes the issue clear. I’ll explain what’s happening, why you see the AttributeError, and give a small, safe code fix plus a couple of robust alternatives you can use in your agent. What I’m responding to: the AttributeError thrown when your code tries to access

publication.track.kind

while

publication.track

is None. Why this happens • A TrackPublication can exist even when its actual

track

object is not yet attached (for example: the remote participant published a track but it hasn’t been subscribed or the SDK hasn’t created the track object yet). • In that case

publication.track

is None, so doing

publication.track.kind

raises AttributeError. Quick safe fix • Check that

publication.track

is not None before accessing

kind

. Replace your list comprehension line with a safe version:

video_tracks = [
    publication.track
    for publication in remote_participant.track_publications.values()
    if publication.track is not None and publication.track.kind == rtc.TrackKind.KIND_VIDEO
]

Alternative: inspect publication metadata instead of

publication.track

• Some SDKs expose

publication.kind

(or similar) on the publication itself so you don’t need the

track

object to know the track type. If available, use that:

video_tracks = [
    publication.track
    for publication in remote_participant.track_publications.values()
    if publication.kind == rtc.TrackKind.KIND_VIDEO and publication.track is not None
]

(Use

publication.kind

only if the Python SDK you use provides it — check via

dir(publication)

.) Better pattern: handle both current state and future events • The safest approach combines: ◦ A snapshot check with the None guard (above), and ◦ Event handlers for

track_subscribed

, `track_muted`/`track_unmuted`, and

track_unsubscribed

to catch when a track appears or its state changes. Example pattern (concise):

def get_video_tracks(remote_participant):
    tracks = []
    for pub in remote_participant.track_publications.values():
        # guard against None track
        t = pub.track
        if t is not None and t.kind == rtc.TrackKind.KIND_VIDEO:
            tracks.append(t)
    return tracks

# in your on_enter or startup:
video_tracks = get_video_tracks(remote_participant)
if not video_tracks:
    # no active camera track right now — prompt or wait for subscription event
    pass

# set up event listeners so you react when a video track is subscribed later
@room.on("track_subscribed")
def on_track_subscribed(track, publication, participant):
    if participant.sid == remote_participant.sid and publication.kind == rtc.TrackKind.KIND_VIDEO:
        # now you have the track
        ...

Extra tips • Add logging to see when publications exist but track is None; that helps diagnose timing issues. • If you prompt users immediately when no video tracks are present, consider a short delay (e.g., 1–2s) to avoid false positives while the client is still starting publishing. • Use events to avoid polling: when track appears (subscribed) you can clear any prompts. If you want, I can: • Update your exact on_enter snippet with the safe checks and event handlers integrated, using the same variable names as your project. • Show a small debounce/cooldown example so prompts don’t spam users. Which would you prefer?

154435.983 DEBUG… livekit.agents start reading stream {"participant": "web-call-7471434507", "source": "SOURCE_MICROPHONE", "pid": 1879613, "job_id": "AJ_JbvWLiXgGtov", "room_id": "RM_hCzj9jBFRWoc"} 154435.987 DEBUG… livekit.agents start reading stream {"participant": "web-call-7471434507", "source": "SOURCE_CAMERA", "pid": 1879613, "job_id": "AJ_JbvWLiXgGtov", "room_id": "RM_hCzj9jBFRWoc"} these are some logs, but when i ask if my camera feed is visible or not then agent denying it.

Thanks — I can see why that’s confusing: your logs show the agent started reading the camera stream, but the agent verbally denies seeing the camera. I’ll walk through the likely gaps and give concrete checks and a small code checklist so you can pinpoint and fix the problem. What I’m responding to: logs show camera stream started (SOURCE_CAMERA) but the agent still says it cannot see the camera. Quick summary of likely causes • The LiveKit session is receiving the video track, but the agent runtime is not actually sampling/processing video frames (so the agent has no visual input to reason about). • The agent’s code may not be checking for video frames or may be checking the wrong property (track exists ≠ frames processed). • Vision/LLM configuration may not be enabled so even processed frames aren’t used in agent responses. Concrete checks to run now 1. Confirm video input is enabled in the AgentSession / RoomOptions • Ensure you created the session with video enabled (some SDKs require an explicit flag). ◦ Example: session/room options or

session.input.set_video_enabled(True)

(or similar) — if video sampling is disabled the agent will receive the track but not sample frames. 1. Verify frames are actually being received and processed • Add explicit logging where your code handles incoming frames (or where the agent stores latest frames). ◦ Log when a new frame is received (timestamp + participant id). ◦ Log frame resolution or a short checksum to prove content arrived. • Example debug line: "received video frame from {participant.identity} at {timestamp} — size: {w}x{h}" 1. Check agent-side logic that decides “I can see the camera” • Ensure the agent checks processed frames (not only publication existence). ◦ publication.track != None or publication.track.kind == VIDEO indicates track presence — but you must sample frames to produce vision-based responses. • If your agent uses a

self._latest_frame

or similar, verify it gets set when frames arrive and is used when forming responses. 1. Confirm your LLM / multimodal settings support vision input • The model/LLM invocation must accept vision inputs. If the LLM is text-only or vision is disabled, frames won’t influence responses. • If your stack requires passing image bytes/frames explicitly into the LLM context, verify that step is implemented and logged. 1. Test with a deterministic visual prompt • Quick test: ask the participant to show a distinct object (e.g., a red card) and check: ◦ Do frame logs show the frames during that time? ◦ Does the agent detect the object? • If frames are logged but the agent still misses it, the problem is likely in the frame → LLM input path. 1. Ensure you aren’t accidentally ignoring camera vs screen-share • Check publication.source (camera vs screen) and ensure your code treats

SOURCE_CAMERA

as a camera feed to analyze. Small code checklist / safe guards to add • Guard when reading publications:

for pub in remote_participant.track_publications.values():
    if pub.track is None:
        continue
    if pub.track.kind == rtc.TrackKind.KIND_VIDEO:
        # log and process frames here

• Log on frame callback:

def on_video_frame(frame, participant):
    logger.debug("frame received", extra={"participant": participant.identity, "w": frame.width, "h": frame.height})
    self._latest_frame = frame

• Before agent replies about visibility, check you have a recent frame:

if not self._latest_frame or time.time() - self._latest_frame_ts > 5:
    # no recent frame -> prompt user or say "I can't see your camera"
else:
    # include frame in LLM context and answer

What to send me if you want a tailored fix • The AgentSession creation code (how you start the session / RoomOptions). • The snippet that handles video frames or the on_enter handler where you check

video_tracks

. • Recent logs showing both the "start reading stream" lines and any frame-received logs, and the agent’s denial response. If you share those, I’ll give a minimal patched snippet you can paste in (including the frame callback and an example of passing a frame into the agent’s reasoning path). Sources: Vision | LiveKit Documentation

how to check if video stream is attached or not and going to llm or not?

Good — here’s a focused, step-by-step checklist and code you can use right now to confirm (1) a video stream is actually attached to your agent session, and (2) frames from that stream are being passed into the LLM context. What I’m responding to: how to check whether the video stream is attached and whether frames are reaching the LLM. 1) Confirm the track is subscribed (stream attached) • Ensure you have an event handler for

track_subscribed

and that it runs for video tracks. Log when this fires. Example:

from livekit import rtc
import asyncio
import logging

logger = logging.getLogger("video-check")

@room.on("track_subscribed")
def on_track_subscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
    if publication.kind == rtc.TrackKind.KIND_VIDEO or (track and track.kind == rtc.TrackKind.KIND_VIDEO):
        <http://logger.info|logger.info>("track_subscribed: participant=%s source=%s pub_has_track=%s",
                    participant.identity,
                    getattr(publication, "source", "unknown"),
                    publication.track is not None)
        # Attach VideoStream to process frames
        asyncio.create_task(handle_video_stream(track, participant))

2) Verify frames are being received (log frames) • Wrap the video track in a VideoStream (or use the SDK frame iterator) and log each frame you receive so you can see actual frames coming through. Example:

async def handle_video_stream(track: rtc.Track, participant: rtc.RemoteParticipant):
    if track is None:
        logger.warning("No track object for participant %s", participant.identity)
        return

    # create a VideoStream if your SDK provides it
    vs = rtc.VideoStream(track)
    async for frame in vs:
        # Log basic frame info to prove data is flowing
        logger.debug("video_frame: participant=%s ts=%s size=%dx%d",
                     participant.identity, getattr(frame, "timestamp", "na"),
                     getattr(frame, "width", "na"), getattr(frame, "height", "na"))
        # store latest frame for LLM usage
        # self._latest_frame = frame

3) Ensure agent session is configured to accept video input • When creating or starting your AgentSession / RoomOptions, confirm

video_input=True

(or the equivalent option in your SDK). If this is not enabled, frames may be ignored for LLM input. Example sketch:

from livekit.agents import room_io

room_opts = room_io.RoomOptions(video_input=True)
await session.start(room=ctx.room, agent=my_agent, room_options=room_opts)

4) Confirm you actually pass frames to the LLM (ImageContent) • Once you have a frame, encode it and add it to the message/context you send to the LLM (many LLMs require an

ImageContent

or base64-encoded image in the chat context). Example pattern:

import base64
from livekit.agents import ImageContent

def encode_frame_to_base64(frame) -> str:
    # convert frame to JPEG bytes (use your encoder)
    jpeg_bytes = encode_frame_as_jpeg(frame)  # implement with your image lib
    return base64.b64encode(jpeg_bytes).decode("utf-8")

# when preparing LLM prompt
if self._latest_frame:
    b64 = encode_frame_to_base64(self._latest_frame)
    img_content = ImageContent(image=f"data:image/jpeg;base64,{b64}")
    chat_ctx.add_message(role="user", content=[ "Please analyze this image", img_content ])

5) Add a confirmation log right before LLM call • Before you call the LLM, log what you are sending (indicate image included). This proves your code actually hands an image to the model. Example:

<http://logger.info|logger.info>("Invoking LLM: include_image=%s", self._latest_frame is not None)
# then call the LLM with the prepared chat_ctx

6) Test flow with a deterministic visual prompt • Ask the participant to show a distinctive object (e.g., “hold up a red card”), then: ◦ Watch the frame logs to see frames with that time range. ◦ Confirm the pre-LLM log shows image inclusion. ◦ Confirm LLM response acknowledges the visual cue. 7) Common failure points and checks • Track exists but

publication.track

is None — wait for

track_subscribed

event and guard access with

if publication.track is None: continue

. •

video_input

disabled — enable it in RoomOptions. • Frames received but not encoded/passed to LLM — ensure you encode and attach as ImageContent. • LLM model not multimodal — confirm the LLM supports vision input. • Timing: sample a recent frame (use a latest-frame variable) rather than assuming immediate availability. 8) Quick debugging checklist to paste into your logs • On track_subscribed: participant id, publication.source, publication.track is None? • On first frame: participant id, width x height, timestamp • Before LLM call: include_image True/False • After LLM call: LLM returned text and any references to image If you’d like, I can: • Provide a tailored snippet that plugs into your existing agent code (paste your

on_enter

or session start snippet), or • Add a short debounce/cooldown for prompts so the agent doesn’t repeatedly tell users to turn on their camera. Which would you prefer? Sources: An open source stack for real-time multimodal AI | Vision | LiveKit Documentation | Agent session | LiveKit Documentation | Large language models (LLM) overview | LiveKit Documentation

I am using gpt-4o-mini model?

Thanks — that’s an important detail. GPT-4o-mini is a text-only model and does not accept image/vision inputs, so you cannot send video frames to it for visual understanding. That explains why your agent may receive frames but still “deny” seeing the camera: even if frames are attached, GPT-4o-mini won’t process them. What I’m responding to: you asked whether using the gpt-4o-mini model supports video/frame input. Short answers and next steps • GPT-4o-mini does NOT support vision inputs. Do not expect image frames passed as ImageContent to affect its outputs. • To get an agent that can reason about camera frames you must use a vision-capable model (for example, models advertised as multimodal or “vision” capable such as Gemini Live or other LiveKit-compatible vision models). Recommended actions 1. If you need visual understanding (object detection, “do you see my face?”, etc.): ◦ Switch to a vision-capable LLM. Use a model that explicitly supports image inputs in your integration. ◦ Ensure your AgentSession is configured with

video_input=True

and you pass frames as ImageContent to the model. 1. If you keep using GPT-4o-mini: ◦ Use video frames only for local logic in your agent code (e.g., run a separate vision pipeline to detect camera presence or faces, then pass a text summary into GPT-4o-mini). ◦ Example flow: sample frame → run quick CV check (face detection or histogram) → produce text like "camera_on": true → feed that text to GPT-4o-mini. 1. Quick hybrid example (high level) ◦ Receive frames in agent, run a lightweight CV check (OpenCV, face detection). ◦ If face/frame detected, call: ▪︎ either a vision-capable LLM directly with ImageContent, or ▪︎ send GPT-4o-mini a text prompt like "I can see the user's camera feed (face detected). Proceed with…" If you want, I can: • Suggest specific vision-capable models you can use with LiveKit (based on your platform) and show an example of passing ImageContent into the model, or • Provide a small code snippet that does local CV checks (e.g., OpenCV face detection) and then sends a short text summary to GPT-4o-mini so your agent can respond appropriately. Which option would you like: switch to a vision model example, or a hybrid local-CV + GPT-4o-mini snippet?

here in this screenshots gpt4o-mini is accepting image as input

Thanks — I see the screenshot you shared and your point: the UI shows image/camera icons next to “GPT‑4o mini,” which makes it look like the model accepts images. I’ll clarify what’s happening and give clear next steps so you can get visual processing working. What I’m responding to: you showed a UI screenshot implying GPT‑4o‑mini accepts images and asked about using it for camera/video input. Short answer • GPT‑4o‑mini is a text-only model and does not process image/video frames. Seeing camera/image icons in a UI does not guarantee that specific model supports vision input. • To get real visual understanding inside your agent you must either: ◦ Use a multimodal / vision-capable model, or ◦ Run local CV (e.g., OpenCV/face detection) on the frames and pass summarized text to GPT‑4o‑mini. Why the UI can be misleading • UI pages often show capability icons generically for a family of models or to advertise the platform’s overall features. • A model card or playground visual doesn’t always reflect the exact constraints of every variant. The presence of an image icon does not confirm image support for that specific model variant. Recommended options (pick one) 1) Switch to a vision-capable model (recommended if you need real image understanding) • Use an LLM that explicitly supports image inputs (multimodal). Examples used in LiveKit integrations include Gemini Live / other vision-enabled models. • Then: ◦ Ensure your AgentSession is started with video enabled (

video_input=True

). ◦ Sample and encode frames as ImageContent, include them in the LLM prompt, and the LLM will reason about the image. • I can provide a concrete snippet showing how to encode a frame and attach it as ImageContent for a vision model if you want. 2) Hybrid approach: local CV + GPT‑4o‑mini (recommended if you must keep gpt‑4o‑mini) • Process frames locally (e.g., OpenCV, face detection, color/histogram checks). • Produce a concise text summary (e.g., “face detected, lighting OK, resolution 640x480”) and send that text to GPT‑4o‑mini. • This gives you reliable visual signals while keeping your existing model usage. 3) Confirm model capabilities before switching • Check model docs or the API/model list for explicit “vision/multimodal” support. If unsure, I can help verify a specific model name for you. If you want, I can: • Provide a quick local CV snippet (OpenCV) to detect whether a camera frame contains a face, and show how to send the result to GPT‑4o‑mini. • Provide a vision-model integration snippet showing how to encode a frame and include it as ImageContent for a multimodal model (including the exact fields to add to the chat context). • Help you choose a model available to your platform that supports vision and show end-to-end sample code. Which would you like: the hybrid OpenCV + gpt-4o‑mini snippet, or a vision‑model integration example?

how to get recording of web call agent with transcription?

Good — I’ll walk you through how to record a web call agent session and get a transcription. I’ll include the recommended approach (use LiveKit Egress to record the room) plus a Python example that starts a recording and requests transcription, where to store the file, and how to retrieve the transcript. If you prefer a different storage (S3, local, etc.) or want agent-side recording only, tell me and I’ll tailor it. What I’m responding to: how to get recordings of web call agent sessions together with transcriptions. Overview — recommended approach • Use LiveKit Egress (server-side) to record a room (audio and/or video) for the web call. Egress creates a media file (MP4/MKV/MP3) or live stream and can optionally produce a separate audio file. • Once you have the audio recording, generate a transcription. Two options: ◦ Request transcription as part of your egress workflow if your LiveKit deployment or egress tooling supports automatic transcription (check your deployment). ◦ Or export the audio file (from egress output) and send it to a transcription service (Deepgram, OpenAI Whisper/Audio API, or another STT provider). • Store recordings and transcripts in your chosen storage (S3, cloud bucket, or your server) and surface links in your app. Steps and example code (Python) 1) Start an egress (recording) for the specific room • Use the LiveKit server SDK or control-plane API to request a recording/egress job for the room where the web call agent runs. • Typical outputs: MP4 (audio+video), MKV, or audio-only (MP3/WAV). Example (conceptual Python using LiveKit control API client — adjust to your SDK):

from livekit import api
import os

LIVEKIT_URL = os.getenv("LIVEKIT_URL")
API_KEY = os.getenv("LIVEKIT_API_KEY")
API_SECRET = os.getenv("LIVEKIT_API_SECRET")

client = api.LiveKitAPI(LIVEKIT_URL, API_KEY, API_SECRET)

# Start a recording job
req = api.StartEgressRequest(
    room="your_room_name",
    # Example: record all participants' audio+video into an mp4
    output=api.EgressOutput(
        # choose your sink: object storage or direct URL; details vary by deployment
        file=api.EgressFile(file_path="recordings/room_name-{{.StartedAt}}.mp4")
    ),
    # layout or stream options optional
)
resp = client.egress.start_egress(req)
print("egress started:", resp)

Notes: • The exact Python class names and request fields depend on your LiveKit SDK version. Use your SDK’s

egress.start_egress

or control-plane

POST /egress

equivalent. • You can also record audio-only (mp3/wav) by specifying an audio-only output, which is often simpler for transcription. 2) Wait for egress to finish and get the output file location • The egress job returns a job id. Poll job status or subscribe to webhook callbacks your LiveKit control plane can send when egress completes. • The completed job response includes the file URL/path where the recording was saved. 3) Transcribe the recorded audio • If your LiveKit deployment doesn’t run transcription automatically, fetch the audio file and call your STT provider. • Example: send audio to Deepgram or OpenAI’s transcription endpoint, or run Whisper locally. Example: sending audio file to a transcription API (pseudo)

import requests

audio_url = "<URL_FROM_EGRESS>"  # egress output audio file
audio_bytes = requests.get(audio_url).content

# Example: Deepgram (replace with your provider's API)
dg_resp = <http://requests.post|requests.post>(
    "<https://api.deepgram.com/v1/listen?model=general&language=en>",
    headers={"Authorization": f"Token {DEEPGRAM_API_KEY}", "Content-Type": "audio/mpeg"},
    data=audio_bytes
)
transcript = dg_resp.json()
print("transcript text:", transcript["results"]["channels"][0]["alternatives"][0]["transcript"])

4) Store transcript alongside recording and surface in your app • Save the transcript text file or JSON in your storage (S3, DB), link it to the recording entry and room/job id so you can show playback + captions. Optional: Real-time transcription (live captions) • If you need live captions during the call, run a real-time transcription pipeline: ◦ Option A: Ingest the agent’s audio stream to a real-time STT (Deepgram Realtime, Whisper Realtime, or WebSocket-based STT) and publish transcription results back to the room as data messages or overlays. ◦ Option B: Use LiveKit egress with streaming outputs if supported and stream audio to an STT service for near-live captions. Practical considerations and tips • Audio quality: For accurate transcripts, record audio-only or extract mixed audio from the room to avoid overlapping voices. Use single-channel or per-participant audio when possible. • Speaker diarization: If you need speaker labels, use an STT provider that supports diarization or run a pipeline that splits per-participant tracks (LiveKit can support per-participant recording options). • Webhooks: Configure egress-complete webhooks so your backend receives a notification with the file URL and can immediately start transcription. • File formats: MP3/WAV or FLAC are great for STT. Use audio-only when you don’t need video playback. • Cost & limits: Recording and STT services have costs — choose audio-only for cheaper transcription runs. Need exact code for your setup? • I can produce a full working example tailored to your environment if you tell me: ◦ Which LiveKit SDK/version you use (python package name/version). ◦ Do you want audio-only or audio+video recording? ◦ Which transcription provider you prefer (Deepgram, OpenAI, Whisper locally, etc.)? ◦ Where you want to store recordings (S3, local disk, GCS)? Which options would you like me to expand into a ready-to-run example?

with audio and video both.

Great — since you want both audio and video recorded plus a transcription, here’s a concrete, end-to-end plan with runnable Python examples and recommended best practices. I’ll show: • How to start a room composite egress (MP4 recording of audio+video) using the LiveKit control API. • How to run a transcription agent (real-time or post-recording) and store the transcript. • How to wire them together (webhook or poll) so you get the recording file and transcription automatically. Tell me if you want everything targeted to a specific LiveKit SDK version or hosting (self-hosted LiveKit vs LiveKit Cloud) or a preferred STT provider (Deepgram, OpenAI transcription, Whisper). I’ll assume general control-plane API usage and a Deepgram transcription example below. 1) Start room composite egress (record audio + video MP4) • Use the control-plane egress API to start a RoomComposite egress that records the room as an MP4. The egress job produces a file you can store (S3 or local) and later transcribe. Example: Start an MP4 egress via HTTP POST (replace placeholders)

# Example curl to start composite egress (audio+video MP4)
curl -X POST "https://<LIVEKIT_CONTROL_HOST>/v1/projects/<PROJECT_ID>/egress/room_composite" \
  -H "Authorization: Bearer <SERVER_JWT>" \
  -H "Content-Type: application/json" \
  -d '{
    "room": "web-session-123",
    "layout": "speaker-dark",               # or a custom layout
    "output": {
      "file": {
        "file_path": "<s3://my-bucket/recordings/web-session-123-{{.StartedAt}}.mp4>"
      }
    }
  }'

Notes: • Use your control-plane base URL and project path. The exact route may vary per deployment—adjust to your control API. •

file_path

can be an S3 URL if you've configured credentials, or a local path depending on your Egress config. • The egress response returns a job id. Save this to track status. 2) Poll or receive webhook when egress finishes • Configure egress-complete webhooks (or poll job status) so your backend is notified when the MP4 is ready and you have the final file URL. Webhook handling (Python Flask example skeleton):

from flask import Flask, request, jsonify
app = Flask(__name__)

@app.route("/webhook/egress", methods=["POST"])
def egress_webhook():
    event = request.json
    # Example event fields will include egress_id, status, outputs
    if event.get("type") == "egress_completed":
        file_url = event["egress"]["output"]["file"]["url"]  # adjust per event shape
        # enqueue transcription job or call STT directly
        enqueue_transcription(file_url, room=event["egress"]["room"])
    return jsonify({"ok": True})

3) Transcribe the recorded audio (post-recording) • Option A (recommended for accuracy): extract the audio track from MP4 (or request audio-only egress) and send to transcription API (Deepgram, OpenAI, Whisper). • Option B: use a LiveKit agent with STT plugin for real-time transcription during the call (useful for live captions). Example: Download MP4, extract audio, send to Deepgram (Python)

import requests
import subprocess
import os

# 1) download file
mp4_url = "<EGRESS_FILE_URL>"
r = requests.get(mp4_url, stream=True)
with open("recording.mp4", "wb") as f:
    for chunk in r.iter_content(chunk_size=8192):
        f.write(chunk)

# 2) extract audio as WAV using ffmpeg
audio_file = "recording.wav"
subprocess.run([
    "ffmpeg", "-y", "-i", "recording.mp4",
    "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1",
    audio_file
], check=True)

# 3) send WAV to Deepgram (transcription)
DEEPGRAM_API_KEY = os.environ["DEEPGRAM_API_KEY"]
with open(audio_file, "rb") as af:
    dg_resp = <http://requests.post|requests.post>(
        "<https://api.deepgram.com/v1/listen?punctuate=true&diarize=true>",
        headers={"Authorization": f"Token {DEEPGRAM_API_KEY}"},
        data=af
    )
transcript_json = dg_resp.json()
# Example path to combined transcript text depends on provider
transcript_text = transcript_json.get("results", {}).get("channels", [])[0].get("alternatives", [])[0].get("transcript", "")
print("Transcript:", transcript_text)

4) Real-time (or near real-time) transcription during the call If you want live captions or streaming transcription: • Run an agent-side STT (e.g., Deepgram real-time or agent plugin) attached to the room that listens and emits transcribed lines as data messages or text-stream events. • This agent can also save final transcripts to storage when the call ends. Example agent snippet outline (LiveKit Agents Python + Deepgram STT):

from livekit.agents import AgentSession, JobContext, cli
from livekit.plugins import deepgram

@server.rtc_session(agent_name="transcriber-agent")
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        stt=deepgram.STT(api_key=os.environ["DEEPGRAM_API_KEY"])
    )

    @session.on("transcript")
    def on_transcript(transcript):
        # called with partial/final transcripts
        if transcript.is_final:
            save_transcript(ctx.room.name, transcript.text)

    await session.start(room=ctx.room, agent=None)
    await ctx.connect()

• This provides live captions and you can also write the final transcript to storage. 5) Recording layout and per-participant audio (optional) • If you need per-participant tracks for cleaner transcription and speaker diarization, you can: ◦ Use per-track recording options (if your Egress supports per-participant outputs). ◦ Or record composite MP4 and use an STT provider with diarization (Deepgram offers diarize=true), or use per-participant audio files if available. 6) Best practices and tips • Audio quality matters: prefer sampling rate 16k or higher and single-channel audio for STT. • Use

diarize

option if you need speaker labels; otherwise, run speaker labeling post-process. • Store both MP4 and extracted audio in your storage (S3) and keep a transcript JSON for search. • Use short-lived egress jobs or stop egress when the meeting ends to avoid excess costs. • Protect access to recorded files and transcripts (signed URLs, access control). 7) Putting it all together — orchestration flow • User joins room → Agent joins → Start Egress (audio+video) for that room (server-side). • Agent (or a separate transcriber agent) produces live transcripts and/or waits for egress file. • On egress completion webhook, download file → extract audio → call transcription API → save transcript. • Notify caller/UI with file URL and transcript link when done. If you want, I can: • Produce a complete ready-to-run Python example that: 1) Starts egress for a given room (using your LiveKit control endpoint). 2) Handles webhook when egress completes. 3) Extracts audio and calls Deepgram to produce a transcript saved to S3. • Or provide the alternative using a real-time agent-only approach (no post-recording step) that writes transcripts as the call runs. Which would you prefer: a post-recording pipeline example (egress → download → transcribe) or the real-time agent transcription pipeline? Sources: Universal Egress | Transcriber | LiveKit Documentation | Pi Zero Transcriber | LiveKit Documentation | Text and transcriptions | LiveKit Documentation | server-sdk-js | livekit/livekit_composite | DeepWiki

s3_upload = api.S3Upload( bucket=os.environ.get("AWS_BUCKET_NAME", ""), region=os.environ.get("AWS_REGION", ""), access_key=os.environ.get("AWS_ACCESS_KEY", ""), secret=os.environ.get("AWS_SECRET_KEY", ""), force_path_style=False, ) recording_req = api.WebEgressRequest( url="http://localhost:5174/", audio_only=False, preset=api.EncodingOptionsPreset.PORTRAIT_H264_1080P_30, # video_only=True, file_outputs=[api.EncodedFileOutput( file_type=api.EncodedFileType.MP4, filepath=f"recordings/{ctx.room.name}_web_call_recording.mp4", # segment_duration=2, s3=s3_upload )], ) # egress recording start lkapi = api.LiveKitAPI() recording_res = await lkapi.egress.start_web_egress(recording_req) this is my code to save recording in s3 bucket.. below are the logs 24 Feb, 105329.304 am livekit.agents input stream attached 24 Feb, 105329.304 am livekit.agents input stream attached 24 Feb, 105329.305 am livekit.agents input stream detached 24 Feb, 105329.305 am livekit.agents input stream attached 24 Feb, 105329.308 am livekit.agents start reading stream 24 Feb, 105329.308 am livekit.agents http_session(): creating a new httpclient ctx 24 Feb, 105329.310 am livekit.agents using audio io:

RoomIO

->

RecorderIO

->

AgentSession

->

RecorderIO

->

TranscriptSynchronizer

->

RoomIO

24 Feb, 105329.310 am livekit.agents using transcript io:

AgentSession

->

TranscriptSynchronizer

->

RoomIO

24 Feb, 105329.310 am livekit.agents using video io:

RoomIO

>

AgentSession

> (none) 24 Feb, 105338.196 am asyncio Unclosed client session client_session: <aiohttp.client.ClientSession object at 0x7f9ca6565c90> 24 Feb, 105338.197 am asyncio Unclosed connector connections: ['deque([(<aiohttp.client_proto.ResponseHandler object at 0x7f9ca65e4c40>, 673998.128451912)])'] connector: <aiohttp.connector.TCPConnector object at 0x7f9ca6565e70> 24 Feb, 105341.800 am web-call-agent-dev Camera turned ON by web-call-4190098108 24 Feb, 105341.802 am web-call-agent-dev ================Video track subscribed by web-call-4190098108============== 24 Feb, 105341.805 am livekit.agents start reading stream 24 Feb, 105347.683 am livekit.agents received user transcript 24 Feb, 105349.337 am livekit.agents received user transcript 24 Feb, 105352.759 am livekit.agents resumed false interrupted speech 24 Feb, 105400.737 am livekit.agents received user transcript 24 Feb, 105401.637 am livekit.agents received user transcript 24 Feb, 105405.549 am livekit.agents executing tool 24 Feb, 105405.550 am web-call-agent-dev Ending call web-call-4190098108 24 Feb, 105412.608 am livekit.agents resumed false interrupted speech 24 Feb, 105417.934 am livekit.agents Input is shorter by 9736 samples; silence has been prepended to align the input channel. The resulting recording may not accurately reflect the original audio. 24 Feb, 105418.288 am livekit.agents stream closed 24 Feb, 105418.291 am web-call-agent-dev Camera turned OFF by web-call-4190098108 24 Feb, 105418.292 am web-call-agent-dev ================Video track unsubscribed by web-call-4190098108============== 24 Feb, 105418.298 am livekit.agents closing agent session due to participant disconnect (disable via

RoomInputOptions.close_on_disconnect=False

) 24 Feb, 105419.303 am web-call-agent-dev Call web-call-4190098108 ended successfully 24 Feb, 105419.304 am livekit.agents tools execution completed 24 Feb, 105422.155 am livekit.agents input stream detached 24 Feb, 105422.155 am livekit.agents input stream detached 24 Feb, 105422.160 am livekit.agents Input is shorter by 117982 samples; silence has been prepended to align the input channel. The resulting recording may not accurately reflect the original audio. 24 Feb, 105424.163 am livekit.agents deleting room on agent session close (disable via

RoomInputOptions.delete_room_on_close=False

) 24 Feb, 105424.171 am livekit.agents session closed 24 Feb, 105424.182 am livekit.agents uploading session report to LiveKit Cloud 24 Feb, 105424.753 am livekit.agents finished uploading 24 Feb, 105424.754 am livekit.agents shutting down job task 24 Feb, 105424.754 am livekit.agents shutting down telemetry tracer provider Its not uploading recording on s3 bucket.

Thanks — the logs are helpful. They show the egress/recording was started and the agent processed audio/video, but the file never appeared in S3. I’ll walk through the most likely causes and give targeted checks and fixes so you can get your MP4 uploaded to S3. What I’m responding to: your WebEgressRequest code and logs showing the recording lifecycle, but no upload to S3. Top causes to check (ordered by likelihood) 1. S3 credentials / permissions 2. Egress S3 configuration mismatch (bucket/region/path) 3. Egress job failed or didn’t finish successfully 4. Unclosed HTTP client/connection errors observed in logs 5. Egress service not allowed to write to S3 because of network/VPC restrictions Step-by-step checks and fixes 1) Confirm S3 credentials and permissions • Verify the AWS values your code is using: ◦ AWS_BUCKET_NAME, AWS_REGION, AWS_ACCESS_KEY, AWS_SECRET_KEY • Ensure the IAM user/role has these permissions at minimum: ◦ s3:PutObject, s3:PutObjectAcl, s3:ListBucket (and if using multipart or copy: s3:AbortMultipartUpload) • Check the bucket policy for any restrictions (deny by source IP, require encryption, require specific ACLs). • Quick test: from the same host/container where LiveKit runs, try a simple S3 upload using boto3 with the same creds and path:

import boto3, os
s3 = boto3.client("s3",
    aws_access_key_id=os.environ["AWS_ACCESS_KEY"],
    aws_secret_access_key=os.environ["AWS_SECRET_KEY"],
    region_name=os.environ["AWS_REGION"],
)
s3.put_object(Bucket=os.environ["AWS_BUCKET_NAME"], Key="test.txt", Body=b"ok")
print("upload success")

If this fails, fix IAM/creds first. 2) Confirm the egress returned a success output and file URL • When you call

start_web_egress

, save the resulting job id and poll job status or watch for egress completion events. The egress may have errored after starting. • Poll example (pseudo):

job = await lkapi.egress.start_web_egress(recording_req)
job_id = job.egress_id
# poll job status endpoint or use SDK method until status == completed or failed

• If the job status shows failure, check the job error message to identify S3 upload errors. 3) Inspect webhooks / egress-completed event payload • Configure an egress-complete webhook (or check existing webhook logs). The webhook often contains either the final S3 URL or an error message explaining why upload failed. • If you already receive a webhook, examine the `error`/`output` fields for S3-specific errors. 4) Check for unclosed HTTP client / connection errors in logs • Your logs show: "Unclosed client session client_session" and an unclosed connector. These indicate some HTTP client in your process didn’t close. They can sometimes cause network resource exhaustion and make subsequent requests (like S3 upload) fail. • Fix: Ensure any aiohttp or other HTTP client you create is properly closed (use

async with

context or call

.close()

). • Also check whether the LiveKit egress process spawns its own http client — ensure your runtime can create outbound HTTP connections. 5) Confirm S3 path and encodingOptionsPreset compatibility • Confirm the egress filepath is valid for S3:

recordings/{ctx.room.name}_web_call_recording.mp4

becomes the S3 key. If ctx.room.name includes characters disallowed in your bucket policy, upload may fail. • Consider adding a timestamp or UUID to the filename to avoid overwrite or completion race conditions. 6) Network/VPC or private bucket access • If LiveKit Egress runs in a VPC without internet access, ensure it has route to S3 (VPC endpoint or NAT gateway). • If your bucket enforces a VPC endpoint, verify the endpoint and policy allow the egress host. 7) Check Egress service logs for S3 upload errors • Look for lines in the egress job logs mentioning S3, upload, or permission denied. Those messages are the most direct indicator. • If the LiveKit control plane returns an egress job error, paste the error body and I’ll interpret it. 8) Test with local file output first • Temporarily change

file_outputs

to write to local filesystem (if supported) to confirm egress is producing the MP4. • If local output succeeds but S3 fails, the issue is definitely S3 config/network/permissions. Common fixes based on your logs • The “Input is shorter… silence has been prepended” lines are warnings about audio alignment and not directly related to S3. • The unclosed aiohttp client suggests a resource leak. Ensure your code and any custom HTTP usage are clean. This can cause intermittent failures to upload. • The egress may have completed but failed to upload. Check the egress job status and webhook for an explicit upload error. Minimal checklist to run now 1. From the same host/container, run the simple boto3 upload test above. Confirm success. 2. Poll the egress job status after `start_web_egress`; share the job response or status if it shows failed. 3. Check webhook payload for the egress job; paste the error fields if present. 4. Search the egress/agent logs for “s3”, “upload”, “403”, “AccessDenied”, or “permission” messages. 5. If you use a VPC, confirm S3 access (NAT or endpoint). If you want, I can: • Review the egress job status/webhook payload if you paste it (redact secrets). • Provide a patched snippet to ensure aiohttp sessions are closed and show a safe pattern for starting egress and waiting for completion. • Provide a temporary local-file egress example so you can confirm MP4 creation before troubleshooting S3. Which would you like me to help with next — check the egress job status (paste job response) or the boto3 upload test results?

in url is it right that that it can capture localhost:5174 ? or should i use different things like track composite request?

Thanks — this is a helpful question. Short answer: no, using a localhost URL in a WebEgressRequest will not work when the egress runner (LiveKit control plane or egress service) needs to load a page. For recording audio+video you should generally use RoomComposite (or TrackComposite) egress that renders the room and writes an MP4 to S3. Below are the reasons, recommended choices, and exact next steps you can take. Why localhost:5174 is not appropriate • The WebEgressRequest

url

is used by the egress service to load a web page (headless browser) that renders the layout used for the recorded MP4. The egress service runs remotely (LiveKit Cloud or your egress worker), so a

<http://localhost:5174/>

URL points to the egress machine itself — not your local dev machine — and therefore the page won’t be reachable unless you expose it publicly. • If you are running everything locally (egress + app) and the egress process is on the same host, localhost could work — but in most setups (LiveKit Cloud, separate servers, or managed egress), the egress service cannot access your machine’s localhost. Recommended approaches • Use RoomComposite or TrackComposite egress (preferred for audio+video MP4): ◦ RoomCompositeEgress records the rendered room (all participant tracks) into MP4 using a layout. ◦ TrackComposite (if available) can be used to compose tracks in a specific way. ◦ Both support writing directly to S3 via the egress

file

output configuration. • If you want to use a custom web page for layout: ◦ Host that page on a publicly reachable URL (HTTPS recommended) so the egress headless browser can load it. ◦ Provide that public URL in the

WebEgressRequest

. ◦ Ensure the page is static or stable and does not require local-only resources. Concrete options depending on your setup 1) LiveKit Cloud or remote egress (common) • Use RoomComposite egress and S3 file output. No localhost required. • Example (pseudo Python control API): ◦ Start room composite egress for room "my-room" and set

file_outputs

with an S3

EncodedFileOutput

. • Host any custom web layout on a public URL (S3 static site, CloudFront, or your public server). 2) Self-hosted egress on the same machine as your app (local dev) • You can use localhost if the egress runner is running on the same host and can access the page at localhost:5174. • For testing from a separate machine (or Cloud), use ngrok/tunnel or host the page publicly. 3) Simpler: no custom web page, use built-in layouts • Use prebuilt

layout

options like

speaker

,

grid

, or

speaker-dark

in RoomComposite; then you don’t need a custom URL at all. This is simplest and recommended for most recordings. Practical sample: start RoomComposite egress (Python-like pseudo) • Use RoomComposite + file output to S3 (no localhost required). • Example pseudo-code to start MP4 composite and upload to S3:

from livekit import api
# build S3 upload object (your existing s3_upload)
# build a RoomComposite request with layout and S3 file output

req = api.RoomCompositeEgressRequest(
    room="web-session-123",
    layout="speaker-dark",  # built-in layout
    output=api.EncodedFileOutput(
        file_type=api.EncodedFileType_MP4,
        filepath=f"recordings/web-session-123-{timestamp}.mp4",
        s3=s3_upload
    )
)
resp = await lkapi.egress.start_room_composite_egress(req)

(Adjust to exact SDK method names / signature in your environment.) Checklist & troubleshooting steps to get S3 upload working 1. Use RoomComposite/TrackComposite (avoid WebEgress with localhost unless you host the page publicly). 2. Confirm S3 credentials and permissions (PutObject, ListBucket, etc.). Test a direct boto3 upload from the same host. 3. Poll the egress job status or watch egress webhooks for completion/failure and upload errors — the webhook will show S3 errors if any. 4. If you must use WebEgress with a custom page: ◦ Host the page at a public HTTPS URL reachable by the egress runner. ◦ Ensure any assets referenced by that page are also publicly reachable. 1. Address unclosed HTTP client warnings in your logs — resource leaks can cause intermittent egress/upload failures. Use

async with

or explicitly close

aiohttp

sessions. 2. Try local-file output first: configure egress to write to a local filepath (if supported) to confirm MP4 generation before troubleshooting S3. 3. If you plan to test from local dev but egress is remote, use a public tunnel (ngrok) only for temporary testing — production should use a stable public host. If you want, I can: • Provide the exact RoomComposite egress Python snippet matching your SDK version and S3 configuration. • Review an egress job status or webhook payload if you paste it (redact secrets) to interpret any S3 error. • Provide a sample public-hosted layout example and the exact WebEgressRequest fields if you truly need a custom page. Which would you like next: the RoomComposite Python example (with your s3_upload object plugged in), or help inspecting an egress job/webhook that failed?

Can I get the audio call recording if I'm using self-hosted livekit? I can't use anything that is reliant on livekit cloud.

Yes — you can record audio calls while using a self‑hosted LiveKit deployment without relying on LiveKit Cloud. You have two main, production-ready options depending on how you run LiveKit and how you want to store/process recordings: Summary of options • Run the LiveKit Egress service (self‑hosted) to record rooms (recommended). Egress produces MP4 / audio files (WAV/MP3) and can write to local disk or to S3/GCS. • Record locally in your application/agent (agent-side recording). Have the agent capture and save audio (and optionally video) per participant or mixed, then upload/transcribe as needed. I’ll list what to do for each option and provide minimal code/config examples and practical tips. Option A — Use self‑hosted LiveKit Egress (recommended) What it is • Egress is the component that renders a room (composite or track-based) and writes media output. Self‑hosting LiveKit gives you Egress you control, so no Cloud required. Why use it • Reliable, scalable recording of full room (mixed audio + video). • Produces single MP4 or separate audio files suitable for transcription. • Supports writing to local filesystem or cloud object storage you manage (S3/GCS). How to set it up (high level) 1. Ensure your self-hosted LiveKit deployment includes and runs the egress service (egress-worker / chromium renderer). 2. Call the Egress API to start a RoomComposite or TrackComposite egress for the room you want recorded. 3. Configure the egress output to file (local path) or S3 by setting an EncodedFileOutput with an S3 upload config. 4. Poll job status or use webhooks to detect completion; download/upload the file as needed; send audio to STT. Minimal example (Python-like pseudo) • Start a RoomComposite egress that writes MP4 locally:

from livekit import api  # adjust to your installed SDK

s3_upload = api.S3Upload(...)  # only if you want direct upload to S3

req = api.RoomCompositeEgressRequest(
    room="my-room",
    layout="speaker-dark",  # built-in layout
    output=api.EncodedFileOutput(
        file_type=api.EncodedFileType.MP4,
        filepath=f"/tmp/recordings/my-room-{timestamp}.mp4",
        s3=s3_upload  # omit or provide for S3 upload
    )
)

lkapi = api.LiveKitAPI(url=LIVEKIT_URL, api_key=API_KEY, api_secret=API_SECRET)
resp = await lkapi.egress.start_room_composite_egress(req)
# Save resp.egress_id and poll for status, or configure webhook listener for completion

Notes and practical points • If you write locally, ensure the egress worker has write permission and enough disk space. • If you write to S3, configure credentials and bucket access on the egress worker or include S3Upload in the request (your worker must be able to reach S3). • Use RoomComposite for full room recordings; use TrackComposite/Track outputs if you need per-participant files. • For transcription, either: a) run an agent or background worker to extract audio from the MP4 and call your STT provider, or b) request audio-only egress output if your egress options allow it. Option B — Agent‑side / application-level recording (alternate) What it is • Your agent or server captures audio streams directly from the room (e.g., via RoomIO → RecorderIO) and writes files. Works when you want fine-grained control (per-participant files, chunking, or immediate streaming to STT). Why use it • Greater control, can stream audio to STT in real time (low latency captions). • Useful if you do not want to run the full egress stack or prefer to process and store media via your own component. Minimal pattern (conceptual) 1. In your AgentSession, attach RecorderIO / capture incoming mixed audio buffers or per-participant tracks. 2. Write to file or stream to your STT provider as chunks. 3. Finalize file when call ends and store/upload to your storage. Example sketch (pseudo):

# inside your agent/session
with open("/tmp/room-audio.raw", "wb") as out:
    async for audio_chunk in room_audio_stream:
        out.write(audio_chunk)
# post-process with ffmpeg to convert to WAV/MP3

Tradeoffs • You must handle mixing, format conversion, synchronization, and file fragmentation. • More implementation work but more control (and good for live transcription). Transcription step (common to both options) • After you have an audio file (WAV/MP3/FLAC): ◦ Send to your STT provider (Deepgram, OpenAI transcription endpoint, Whisper, etc.). ◦ If you want real-time captions, stream the audio to a real-time STT service while recording. • Use diarization if you need speaker labels. Or record per-participant tracks to map speaker IDs easily. Troubleshooting checklist (if you run into problems) • Egress created but no file: check egress job status and egress worker logs for errors (S3 permissions, disk no space, network errors). • S3 uploads failing: test S3 creds from egress worker host with boto3; check VPC/S3 endpoint restrictions. • No audio in file: verify the layout chosen captures audio; verify participants published audio and egress options include audio. • Headless page not loading: if using WebEgress with a custom URL, the URL must be public/reachable by the egress worker (avoid localhost unless worker runs there). Security & operational notes • Keep API secrets server-side; generate server JWTs for Egress control calls. • Rotate S3 credentials or use IAM roles where possible. • Clean up old recordings automatically to save space. Next step I can do for you • Provide an exact runnable snippet targeted to your environment: ◦ self-hosted LiveKit version + Python SDK version, and ◦ whether you want local disk or S3 output, and ◦ whether you want post-call transcription (provider) or real-time captioning. • Or help you inspect your egress job status/webhook payloads if you have failures. Tell me: • Do you have Egress already deployed with your self-hosted LiveKit? (Yes / No) • Do you want local files or S3 for storage? • Which STT provider (if any) you plan to use for transcription? I’ll give the precise code and configuration next.