LiveKit Real-time SDK for Python

1.1.5 · active · verified Fri Apr 10

The LiveKit Python SDK provides client-side functionality for building real-time audio, video, and data applications using the LiveKit platform. It supports connecting to LiveKit rooms, managing participants, publishing and subscribing to media tracks, and handling various room events. The current version is 1.1.5, with frequent releases addressing bug fixes, performance improvements, and new features.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to connect to a LiveKit room, generate an access token, listen for basic room events (participant connected/disconnected, room disconnected), and properly disconnect. Ensure you have `LIVEKIT_URL`, `LIVEKIT_API_KEY`, and `LIVEKIT_API_SECRET` environment variables set to connect to your LiveKit server.

import asyncio
import os
from livekit import rtc, access_token as at

async def main():
    # Environment variables for LiveKit server URL and API credentials
    livekit_url = os.environ.get('LIVEKIT_URL', 'ws://localhost:7880')
    api_key = os.environ.get('LIVEKIT_API_KEY', 'devkey')
    api_secret = os.environ.get('LIVEKIT_API_SECRET', 'secret')

    # Participant identity and room name
    identity = "my-python-participant"
    room_name = "my-test-room"

    # Create an access token with necessary grants
    grants = at.VideoGrants(room_join=True, room=room_name)
    token = at.AccessToken(api_key, api_secret).with_identity(identity).with_grants(grants).to_jwt()

    # Initialize a Room
    room = rtc.Room()

    # Define event handlers
    @room.on("participant_connected")
    def on_participant_connected(participant: rtc.RemoteParticipant):
        print(f"Participant connected: {participant.identity} (sid: {participant.sid})")

    @room.on("participant_disconnected")
    def on_participant_disconnected(participant: rtc.RemoteParticipant):
        print(f"Participant disconnected: {participant.identity}")

    @room.on("room_disconnected")
    def on_room_disconnected():
        print("Room disconnected")

    try:
        print(f"Connecting to LiveKit room: {room_name} at {livekit_url}")
        await room.connect(livekit_url, token, rtc.RoomOptions())
        print(f"Connected to room {room.name} as {room.local_participant.identity}")

        # Keep the room alive for a short period or until manually disconnected
        print("Staying in room for 10 seconds...")
        await asyncio.sleep(10)

    except Exception as e:
        print(f"Error connecting to room: {e}")
    finally:
        print("Disconnecting from room...")
        await room.disconnect()
        print("Disconnected.")

if __name__ == "__main__":
    # LiveKit requires Python 3.9+
    if os.environ.get('LIVEKIT_URL') and os.environ.get('LIVEKIT_API_KEY') and os.environ.get('LIVEKIT_API_SECRET'):
      asyncio.run(main())
    else:
      print("Please set LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET environment variables.")

view raw JSON →