LiveKit Real-time SDK for Python
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
- gotcha The LiveKit Python SDK is built asynchronously using `asyncio`. Forgetting to `await` async functions or not running the main application within an `asyncio` event loop is a common mistake that will prevent the SDK from functioning correctly.
- gotcha Incorrectly generating `AccessToken` with invalid `identity`, `name`, `grants`, or an expired time can lead to connection failures or unauthorized access to LiveKit rooms. Ensure participant identity is unique and grants match required permissions.
- breaking Versions 1.1.1 and 1.1.2 introduced changes to how `memoryview` and `bytes` are handled in `AudioFrame` and `VideoFrame` (e.g., `materialize sliced memoryviews`, `normalize memoryview format`). This primarily affects applications directly manipulating raw audio/video data.
- gotcha Failing to explicitly call `await room.disconnect()` will leave the client connected to the LiveKit server, consuming resources and potentially delaying cleanup or leading to unexpected behavior on subsequent connection attempts.
Install
-
pip install livekit
Imports
- Room
from livekit import rtc
- AccessToken
from livekit import access_token as at
- VideoGrants
from livekit import access_token as at
Quickstart
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.")