GitHub - cs50victor/livekit-python-sdks: LiveKit real-time SDK and server API for Python

The LiveKit icon, the name of the repository and some sample code in the background.

The LiveKit Python SDK provides a convenient interface for integrating LiveKit's real-time video and audio capabilities into your Python applications. With it, developers can easily leverage LiveKit's WebRTC functionalities, allowing them to focus on building their AI models or other application logic without worrying about the complexities of WebRTC.

This repo contains two packages

livekit: Real-time SDK for connecting to LiveKit as a participant
livekit-api: Access token generation and server APIs

Using Server API

$ pip install livekit-api

Generating an access token

from livekit import api
import os

# will automatically use the LIVEKIT_API_KEY and LIVEKIT_API_SECRET env vars
token = api.AccessToken() \
    .with_identity("python-bot") \
    .with_name("Python Bot") \
    .with_grants(api.VideoGrants(
        room_join=True,
        room="my-room",
    )).to_jwt()

Creating a room

RoomService uses asyncio and aiohttp to make API calls. It needs to be used with an event loop.

from livekit import api
import asyncio

async def main():
    lkapi = api.LiveKitAPI(
        'http://localhost:7880',
    )
    room_info = await lkapi.room.create_room(
        api.CreateRoomRequest(name="my-room"),
    )
    print(room_info)
    results = await lkapi.room.list_rooms(api.ListRoomsRequest())
    print(results)
    await lkapi.aclose()

asyncio.get_event_loop().run_until_complete(main())

Using Real-time SDK

Connecting to a room

from livekit import rtc

async def main():
    room = rtc.Room()

    @room.on("participant_connected")
    def on_participant_connected(participant: rtc.RemoteParticipant):
        logging.info(
            "participant connected: %s %s", participant.sid, participant.identity)

    async def receive_frames(stream: rtc.VideoStream):
        async for frame in video_stream:
            # received a video frame from the track, process it here
            pass

    # track_subscribed is emitted whenever the local participant is subscribed to a new track
    @room.on("track_subscribed")
    def on_track_subscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
        logging.info("track subscribed: %s", publication.sid)
        if track.kind == rtc.TrackKind.KIND_VIDEO:
            video_stream = rtc.VideoStream(track)
            asyncio.ensure_future(receive_frames(video_stream))

    # By default, autosubscribe is enabled. The participant will be subscribed to
    # all published tracks in the room
    await room.connect(URL, TOKEN)
    logging.info("connected to room %s", room.name)

    # participants and tracks that are already available in the room
    # participant_connected and track_published events will *not* be emitted for them
    for participant in room.participants.items():
        for publication in participant.tracks.items():
            print("track publication: %s", publication.sid)

Examples

Facelandmark: Use mediapipe to detect face landmarks (eyes, nose ...)
Whisper: Transcribe an audio track using OpenAI whisper
Basic room: Connect to a room
Publish hue: Publish a rainbow video track
Publish wave: Publish a sine wave

Getting help / Contributing

Please join us on Slack to get help from our devs / community members. We welcome your contributions(PRs) and details can be discussed there.

LiveKit Ecosystem
Client SDKs	Components · JavaScript · iOS/macOS · Android · Flutter · React Native · Rust · Python · Unity (web) · Unity (beta)
Server SDKs	Node.js · Golang · Ruby · Java/Kotlin · PHP (community) · Python (community)
Services	Livekit server · Egress · Ingress
Resources	Docs · Example apps · Cloud · Self-hosting · CLI