how can i get the url of recording uploaded in s3 from room egress?

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.

how can i get this in python code

You're asking for a Python example that receives LiveKit's

egress_ended

webhook and extracts the S3 recording URL. Below is a concise example using Flask that: 1. receives the webhook POST, 2. validates the webhook signature using a shared secret, 3. parses the JSON payload, 4. extracts the S3 URL from either

egressInfo.file.location

or any entries in

egressInfo.fileResults

. Note: Replace

YOUR_WEBHOOK_SECRET

with your webhook secret configured in LiveKit. If you do not use signature verification, remove the verification block — but verifying is recommended.

from flask import Flask, request, jsonify, abort
import hmac
import hashlib
import json

app = Flask(__name__)

# Set this to your LiveKit webhook secret
WEBHOOK_SECRET = b"YOUR_WEBHOOK_SECRET"

def verify_signature(raw_body: bytes, signature_header: str) -> bool:
    """
    Verify HMAC-SHA256 signature sent by LiveKit.
    LiveKit signs webhook bodies with HMAC-SHA256 using the webhook secret.
    The header usually contains the hex digest. Adjust if your header format differs.
    """
    if not signature_header:
        return False
    computed = hmac.new(WEBHOOK_SECRET, raw_body, hashlib.sha256).hexdigest()
    # Expect header to contain the hex digest directly. If LiveKit prefixes the header (e.g., "sha256="),
    # strip that prefix here before comparing.
    return hmac.compare_digest(computed, signature_header)

def extract_s3_urls(egress_info: dict) -> list:
    """
    Extract URLs from egressInfo object. Returns list of URLs (may be empty).
    """
    urls = []
    if not egress_info:
        return urls

    # Single-file responses
    file_obj = egress_info.get("file")
    if file_obj and file_obj.get("location"):
        urls.append(file_obj["location"])

    # Multi-file results (fileResults array)
    file_results = egress_info.get("fileResults") or egress_info.get("fileResults", [])
    if isinstance(file_results, list):
        for fr in file_results:
            loc = fr.get("location")
            if loc:
                urls.append(loc)

    return urls

@app.route("/webhook/egress", methods=["POST"])
def egress_webhook():
    raw_body = request.get_data()
    sig_header = request.headers.get("X-Livekit-Signature") or request.headers.get("X-Signature")

    # Verify signature if you have a webhook secret
    if not verify_signature(raw_body, sig_header):
        # If you don't want to verify, comment out the above block.
        abort(401, "invalid signature")

    try:
        payload = request.get_json(force=True)
    except Exception:
        abort(400, "invalid json")

    event = payload.get("event")
    # Only process egress_ended (you can handle other events as needed)
    if event != "egress_ended":
        return jsonify({"ok": True, "message": "event ignored"}), 200

    egress_info = payload.get("egressInfo", {})
    urls = extract_s3_urls(egress_info)

    if urls:
        # Do whatever you want with the URL(s): store in DB, notify user, etc.
        # For demo, return them in the response.
        return jsonify({"ok": True, "s3_urls": urls}), 200
    else:
        return jsonify({"ok": True, "s3_urls": [], "message": "no file locations found"}), 200

if __name__ == "__main__":
    # Run in debug for local testing. In production, use a proper WSGI server.
    app.run(host="0.0.0.0", port=5000, debug=True)

Extra notes and tips • Header name: Confirm the header LiveKit sends for the signature; common names used are

X-Livekit-Signature

or

X-Signature

. Adjust the code accordingly. • Signature format: This snippet assumes the header contains the HMAC hex digest. If LiveKit prefixes the header (for example,

sha256=...

), strip that prefix before verifying. • Egress shapes: For room composite egress you will typically find

egressInfo.file.location

. For multi-output egress you may need to iterate

egressInfo.fileResults

. • You can also view the URL in the LiveKit Cloud dashboard if you prefer a UI inspection. If you tell me: • which header your LiveKit instance sends for signatures, or • whether you want an example using FastAPI or Django instead of Flask, I’ll provide an updated snippet.