filepath

filepath

egress_id

# Example (Python) — requires LiveKit server SDK client with egress methods
from livekit.api import LiveKitAPI, RoomCompositeEgressRequest
import asyncio
import time

async def start_and_get_output_file(lkapi: LiveKitAPI, room_name: str):
    # 1) Start room composite egress and capture egress_id
    req = RoomCompositeEgressRequest(
        room_name=room_name,
        layout="speaker-dark",
        # Specify the desired output filepath. This determines the file name.
        # Example for local or cloud path depending on your egress sink:
        filepath="recordings/my-room-{room_name}-{time}.mp4"
    )
    info = await lkapi.egress.start_room_composite(req)
    egress_id = info.egress_id
    print("Started egress_id:", egress_id)

    # 2) Poll for completion (or you can use webhook/events instead)
    while True:
        # list_egress or get_egress depending on SDK; using list_egress filtered by room
        egress_list = await lkapi.egress.list_egress(room_name=room_name)
        target = next((e for e in egress_list if e.egress_id == egress_id), None)
        if not target:
            print("Egress not found yet; retrying...")
            time.sleep(2)
            continue

        print("Egress status:", target.status)  # e.g., STARTED, COMPLETED, FAILED
        if target.status == "COMPLETED":
            # The API response includes metadata about outputs. Check fields like `output` or `filepath`
            # Print whichever field holds the final file path/name
            print("Final output filepath:", getattr(target, "filepath", None) or getattr(target, "output", None))
            return getattr(target, "filepath", None) or getattr(target, "output", None)
        if target.status == "FAILED":
            raise RuntimeError(f"Egress failed: {target.error}")
        time.sleep(2)

# Usage example
async def main():
    lkapi = LiveKitAPI()  # initialize with your config/auth
    filepath = await start_and_get_output_file(lkapi, "my-room")
    print("Resolved output file:", filepath)
    await lkapi.aclose()

asyncio.run(main())

filepath

list_egress

get_egress

COMPLETED

stop_egress

stop_egress

stop_egress

egress_ended

info = await lkapi.egress.stop_egress(api.StopEgressRequest(egress_id="EGRESS_ID"))
if info.file_info:
    print("File path:", info.file_info.location)
    print("Download URL:", info.file_info.download_url)

info, err := egressClient.StopEgress(ctx, &livekit.StopEgressRequest{EgressId: "EGRESS_ID"})
if err != nil { return err }
if info.FileInfo != nil {
    fmt.Println("File path:", info.FileInfo.Location)
    fmt.Println("Download URL:", info.FileInfo.DownloadUrl)
}

filepath

EgressInfo

egress_ended

stop_egress

stop_egress

egress_ended

EgressInfo

file_info.location

file_info.download_url

file_info

get_egress

list_egress

egress_id

COMPLETED

file_info.location

download_url

# On your webhook route receiving egress_ended payload
payload = request.json
egress_info = payload.get("egress_info") or payload.get("data")  # depends on webhook format
file_info = egress_info.get("file_info")
if file_info:
    print("Final file path:", file_info.get("location"))
    print("Download URL:", file_info.get("download_url"))

egress_ended

file_info.location

file_info.download_url

egress_info

# webhook_egress.py
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
import hmac
import hashlib
import os
import json

app = FastAPI()
# Set LIVEKIT_WEBHOOK_SECRET in env if you want to verify signatures
WEBHOOK_SECRET = os.getenv("LIVEKIT_WEBHOOK_SECRET", None)

class EgressEndedPayload(BaseModel):
    event: str
    egress_info: dict | None = None

def verify_signature(body: bytes, signature: str, secret: str) -> bool:
    # LiveKit webhook signature is HMAC-SHA256 of the raw body using the webhook secret.
    expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.post("/webhook")
async def handle_webhook(request: Request):
    body = await request.body()
    # Optional signature header name; replace if your setup differs
    signature = request.headers.get("x-livekit-signature", "")

    if WEBHOOK_SECRET:
        if not signature:
            raise HTTPException(status_code=400, detail="Missing signature")
        if not verify_signature(body, signature, WEBHOOK_SECRET):
            raise HTTPException(status_code=401, detail="Invalid signature")

    try:
        payload = json.loads(body)
    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="Invalid JSON")

    event_type = payload.get("event") or payload.get("type")  # depends on your LiveKit setup
    if event_type not in ("egress_ended", "EGRESS_ENDED"):
        return {"status": "ignored", "event": event_type}

    # Egress info may be under `egress_info` or `data.egress_info` depending on your webhook format
    egress_info = payload.get("egress_info") or payload.get("data", {}).get("egress_info")
    if not egress_info:
        return {"status": "no_egress_info"}

    # Extract file info
    file_info = egress_info.get("file_info") or egress_info.get("FileInfo")
    if not file_info:
        return {"status": "no_file_info"}

    location = file_info.get("location") or file_info.get("Location")
    download_url = file_info.get("download_url") or file_info.get("DownloadUrl")

    # Do something with the filename/path (store in DB, notify user, start post-processing, etc.)
    print("Egress completed. File location:", location)
    print("Download URL:", download_url)

    return {"status": "ok", "location": location}

# Run with: uvicorn webhook_egress:app --reload --port 8000

filepath

file_info.location

event

egress_info

data