Can I create a global call ended function?

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

python So to make sure I'm looking for function that will on call end no matter what...if ai is programmed to hang up or if caller hangs up

You're right — to catch every call end (AI hangup or caller hangup) implement both a webhook for the

room_finished

event and an

on_session_end

callback in your Python session code. Webhook catches the room lifecycle globally;

on_session_end

runs within your agent/session code. Use both for full coverage. Below are minimal Python examples. 1) Webhook handler (FastAPI) — global, fires when the last participant leaves or when the room is deleted:

from fastapi import FastAPI, Request
import uvicorn

app = FastAPI()

@app.post("/webhook")
async def webhook(req: Request):
    payload = await req.json()
    if payload.get("event") == "room_finished":
        room = payload.get("room", {}).get("name")
        # global call-ended logic here
        print(f"room_finished: {room}")
    return {"status": "ok"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

2)

on_session_end

callback — runs when an agent/session ends (AI hangup or session termination):

from some_agent_framework import rtc_session  # replace with your decorator/import

async def handle_call_ended(session):
    # per-session call-ended logic here
    print("Session ended for room:", session.room.name)

@rtc_session(on_session_end=handle_call_ended)
async def agent_session(ctx):
    # agent logic
    pass

3) Force end (optional but ensures webhook triggers): delete the room from server-side when you want to force hangup:

# example using an API client available in your environment
await api.room.delete_room(api.DeleteRoomRequest(room=room_name))

Recommended pattern: • Use the webhook as the authoritative global signal for call termination and to run shared cleanup/notifications/logging. • Use

on_session_end

for immediate per-session teardown or state updates inside the agent. • Optionally call

delete_room

when you need to forcibly end a call and ensure

room_finished

fires. Do you want a complete runnable example wired to your LiveKit webhook signing/verification or to see this integrated with your existing agent code?

wtv... so if i create a room how do i track when everyone disconnects from the room,

You can reliably detect when everyone disconnects by using the

room_finished

webhook — it fires when the last participant leaves the room. Use that as your global, authoritative signal to run cleanup, update state, send notifications, etc. Quick steps (Python): 1. Expose a webhook endpoint (e.g., FastAPI) and inspect incoming events. 2. When

event == "room_finished"

, run your "call ended" logic for that room. 3. Optionally also handle participant disconnect events if you want per-user updates before the final end. Minimal FastAPI example:

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/webhook")
async def webhook(req: Request):
    payload = await req.json()
    if payload.get("event") == "room_finished":
        room = payload.get("room", {}).get("name")
        # global call-ended logic here
        print("call ended for room:", room)
    return {"status": "ok"}

If you need per-session cleanup inside your agent code, add an

on_session_end

callback too — but for global detection of "everyone left," rely on

room_finished

. Do you want the webhook signing/verification example next?

what room tho? this code is no where near logical.

You're right — I should be explicit about what "room" means and show the full flow. Quick plan: tell me how you create rooms (server SDK or client?), then I’ll give a concrete, logical Python example that shows creating a room, where the room name comes from, and how the webhook receives that same room name when everyone leaves. A few quick questions to tailor the example: • Do you create rooms server-side with the LiveKit Python SDK, or let clients create them? • Do you already have a webhook endpoint URL you can receive events on? • Do you need webhook signing/verification included? If you prefer, I can assume a common setup (server creates room with a name, FastAPI webhook receives

room_finished

) and give a single, runnable Python example tying creation -> clients join -> webhook handles

room_finished

. Which do you want?

ok my last question when some disconnects/hangs up. I'm using twilio sip, the live kit server is still running for that session like if i have remaining_p = 1: print("Only agent") will that work after the caller has hung up?

You're asking if checking remaining participant count (e.g.,

remaining_p == 1

) will reliably show "only agent left" after a caller hanging up via Twilio SIP. Short answer: yes — but use LiveKit events (webhook or server callbacks) rather than polling to make it reliable. Key points: • When the Twilio SIP leg disconnects, LiveKit will emit a participant disconnect event for that SIP participant. That lets you update your participant count immediately. • If your logic runs when you detect a participant left and you compute remaining participants (e.g., from room state or your tracked set),

remaining_p == 1

means only one participant remains (the agent). • To be authoritative about the whole call ending, listen for

room_finished

(fires when the last participant leaves). That covers the final hangup. Recommended pattern (concise): • Maintain a server-side set of active participants for each room. • Update it on participant connect / participant_disconnect events (webhook or server SDK). • On disconnect: remove the participant, compute remaining count, then: ◦ If remaining == 1 → run your "only agent" logic. ◦ If remaining == 0 → run final cleanup (or rely on

room_finished

webhook). If you want, I can give a minimal Python webhook example showing: 1) handling

participant_disconnected

to update your participant set and check remaining count, and 2) handling

room_finished

for final cleanup. Which would be more helpful: a FastAPI webhook example or a server-SDK callback example?

Show me a real life full code example, not a snippet

Yes — checking remaining participant count (e.g.,

remaining_p == 1

) can work after a caller hangs up, but make it event-driven for reliability. Here's a complete, runnable Python example that ties room creation, Twilio SIP outbound, webhook handling for participant disconnects, and final room cleanup together. Save as main.py and run with

uvicorn main:app --reload

. Fill env vars (

LIVEKIT_URL

,

LIVEKIT_API_KEY

,

LIVEKIT_API_SECRET

) and configure your LiveKit webhook to point to

/webhook

.

# main.py
import os
import asyncio
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
from typing import Dict, Optional
from dotenv import load_dotenv

# Import your LiveKit server SDK classes - adjust import to your installed package
from livekit.api import LiveKitAPI, CreateSIPParticipantRequest, DeleteRoomRequest
from livekit import AccessToken, VideoGrant
from livekit.api import WebhookReceiver  # adjust if package differs

load_dotenv()

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

if not (LIVEKIT_URL and API_KEY and API_SECRET):
    raise RuntimeError("Set LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET in env")

lk_api = LiveKitAPI(LIVEKIT_URL, API_KEY, API_SECRET)
webhook_receiver = WebhookReceiver(API_KEY, API_SECRET)  # verifies signature

app = FastAPI()

# Simple in-memory state for demo; use Redis/DB in production
# active_calls: room_name -> { "agent_identity": str|None, "caller_identity": str|None }
active_calls: Dict[str, Dict[str, Optional[str]]] = {}

class OutboundRequest(BaseModel):
    phone_number: str
    agent_identity: Optional[str] = "agent-1"

@app.post("/start-outbound")
async def start_outbound(req: OutboundRequest):
    """
    Create a room and place an outbound SIP call (Twilio) into that room.
    Returns a join token for the agent and the room name.
    """
    room_name = f"outcall-{int(asyncio.get_event_loop().time()*1000)}"
    caller_identity = f"caller-{req.phone_number.replace('+','')}"
    agent_identity = req.agent_identity

    # create token for agent to join
    token = (
        AccessToken(API_KEY, API_SECRET)
        .with_identity(agent_identity)
        .with_grants(VideoGrant(room_join=True, room=room_name))
        .to_jwt()
    )

    # Initialize room state
    active_calls[room_name] = {"agent_identity": agent_identity, "caller_identity": caller_identity}

    try:
        # Place outbound SIP call into the room. Replace sip_trunk_id with your trunk.
        await lk_api.sip.create_sip_participant(
            CreateSIPParticipantRequest(
                room_name=room_name,
                participant_identity=caller_identity,
                sip_trunk_id="ST_YOUR_TRUNK_ID",
                sip_call_to=req.phone_number,
                wait_until_answered=True,
            )
        )
    except Exception as e:
        # cleanup on failure
        active_calls.pop(room_name, None)
        raise HTTPException(status_code=500, detail=f"Failed to place call: {e}")

    return {"room_name": room_name, "agent_token": token}

@app.post("/webhook")
async def webhook(request: Request):
    """
    Receives LiveKit webhook events.
    Handles participant_joined / participant_left and room_finished.
    """
    body = await request.body()
    auth_header = request.headers.get("Authorization")

    try:
        event = webhook_receiver.receive(body, auth_header)
    except Exception as e:
        raise HTTPException(status_code=401, detail=f"Webhook verification failed: {e}")

    evt_name = event.event
    room_name = getattr(event.room, "name", None)
    identity = getattr(getattr(event, "participant", None), "identity", None)

    # participant joined: ensure we know identities
    if evt_name == "participant_joined":
        if room_name not in active_calls:
            active_calls[room_name] = {"agent_identity": None, "caller_identity": None}
        kind = getattr(event.participant, "kind", None)
        # classify participant (adjust to your identity/kind scheme)
        if kind == "SIP" or (identity and identity.startswith("caller-")):
            active_calls[room_name]["caller_identity"] = identity
        else:
            active_calls[room_name]["agent_identity"] = identity

    # participant left: update tracking and check remaining count
    elif evt_name in ("participant_left", "participant_disconnected"):
        if room_name in active_calls:
            current = active_calls[room_name]
            # remove the identity if known
            if current.get("caller_identity") == identity:
                current["caller_identity"] = None
                print(f"Caller {identity} left room {room_name}")
            if current.get("agent_identity") == identity:
                current["agent_identity"] = None
                print(f"Agent {identity} left room {room_name}")

            # compute remaining participants heuristically
            remaining = sum(1 for v in (current.get("caller_identity"), current.get("agent_identity")) if v)
            if remaining == 1:
                print(f"Only one participant remains in {room_name} — likely the agent.")
                # run your "only agent" logic here (e.g., notify agent, play goodbye)
            elif remaining == 0:
                print(f"No participants remain in {room_name} — final cleanup can run.")
                # clean up local state; room_finished should also arrive
                active_calls.pop(room_name, None)

    # final room finished event: authoritative signal that everyone left
    elif evt_name == "room_finished":
        if room_name in active_calls:
            print(f"room_finished for {room_name} — final cleanup.")
            active_calls.pop(room_name, None)
        else:
            print(f"room_finished received for unknown room {room_name}")

    return {"status": "ok"}

@app.post("/end-call/{room_name}")
async def end_call(room_name: str):
    """Force-end a call by deleting the room."""
    try:
        await lk_api.room.delete_room(DeleteRoomRequest(room=room_name))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to delete room: {e}")
    active_calls.pop(room_name, None)
    return {"status": "ended"}

@app.get("/status")
async def status():
    return {"active_calls": active_calls}

How this answers your question: • When Twilio SIP caller hangs up, LiveKit emits a participant_left/participant_disconnected webhook — remove that identity and recalc remaining participants. •

remaining == 1

will reliably indicate only one tracked participant remains (usually the agent) if you maintain server-side state. • Use

room_finished

as the authoritative final signal that everyone left. • Don't rely solely on polling room state on the LiveKit server; webhooks (or server SDK event callbacks) give immediate, reliable updates. If you want, I can: • Add webhook signing verification details for your exact SDK, • Replace in-memory tracking with a Redis example, • Show how to play an audio message to the agent or trigger agent cleanup when only the agent remains. Which would you like? Sources: Webhooks | LiveKit docs | Managing participants | LiveKit docs | server-sdk-js | livekit/livekit_composite | DeepWiki | Agents telephony integration | LiveKit docs | Server-side | livekit/livekit_composite | DeepWiki

can calls be streamed live?

Yes — you can stream calls live. Quick options and when to use each: • Real-time in-room (WebRTC): any participant (web/mobile) can subscribe to the SIP caller’s audio in the LiveKit room for ultra-low latency listening — use this when you want live monitoring or a web client to hear the call. • Agent/server-side processing: run a server-side agent that subscribes to the call audio (e.g.,

rtc.AudioStream

) to do real-time transcription, AI processing, or monitoring. • Track egress to WebSocket: forward a live audio track as raw PCM to a WebSocket for real-time processing by external services (good for low-latency STT or ML pipelines). • Room composite egress (RTMP): broadcast the room to external platforms (YouTube, Twitch, custom RTMP) if you need public livestreaming. Recommended pattern for telephony (Twilio SIP) + LiveKit: 1. Connect the SIP participant into a LiveKit room. 2. Subscribe or attach a server-side consumer (agent or egress) to the caller’s audio track. 3. Use Webhook events to coordinate start/stop of egress when the call starts/ends. Do you want a short runnable example for: • subscribing with a server-side agent to process audio, or • creating an RTMP egress to broadcast the call? Sources: LiveKit Use-cases: Livestreaming | Capturing the First Few Seconds of a Call for Processing | Universal Egress | Track egress | LiveKit docs | Inbound calls with Twilio Voice | LiveKit docs