LiveKit Python SDKs

PlatformAudio uses WebRTC's Audio Device Module (ADM) for microphone capture. It is the recommended mode for most applications, especially voice and video calls.

Key Features:

• Built-in voice processing: Echo Cancellation (AEC), Noise Suppression (NS), and Auto Gain Control (AGC).
• Hardware acceleration on supported platforms.
• Automatic speaker playout for received audio.
• Device enumeration and selection.

Limitations:

• No direct access to raw audio frames (ADM sends directly to WebRTC).
• Cannot apply custom audio processing before publishing.

Usage Example:

platform_audio = rtc.PlatformAudio()
source = platform_audio.create_audio_source(
    rtc.PlatformAudioOptions(
        echo_cancellation=True,
        noise_suppression=True,
        auto_gain_control=True,
    )
)
track = rtc.LocalAudioTrack.create_audio_track("microphone", source)

RPC allows one participant to call a predefined method on another participant. This is useful for agent-to-client communication.

1. Registering a method: The receiver uses @room.local_participant.register_rpc_method(name) to define a handler. The handler receives RpcInvocationData and can return a response.
2. Performing a request: The caller uses await room.local_participant.perform_rpc(...) specifying the destination_identity, method, and payload.

Note on Errors: If a handler raises an RpcError with a string message, that message is sent to the caller. Other unhandled errors arrive at the caller as code 1500 ("Application Error").

# Registering a method (Receiver side)
@room.local_participant.register_rpc_method("greet")
async def handle_greet(data: RpcInvocationData):
    print(f"Received greeting from {data.caller_identity}: {data.payload}")
    return f"Hello, {data.caller_identity}!"

# Performing a request (Caller side)
try:
  response = await room.local_participant.perform_rpc(
    destination_identity='recipient-identity',
    method='greet',
    payload='Hello from RPC!'
  )
  print(f"RPC response: {response}")
except Exception as e:
  print(f"RPC call failed: {e}")

The LiveKitAPI class supports two authentication modes depending on your environment:

1. API key & secret (Backend): Recommended for server-side applications. The SDK uses your LIVEKIT_API_KEY and LIVEKIT_API_SECRET to sign a short-lived token for every request. Never expose your API secret to a client.
2. Access token (Client-side): Used when you cannot expose the API secret. You must provide a pre-signed access token that already contains the necessary grants for the operations you intend to perform. The SDK sends this token verbatim.

Configuration follows a hierarchy: explicitly passed arguments take precedence over environment variables. The SDK falls back to the following environment variables if arguments are omitted:

• LIVEKIT_URL
• LIVEKIT_API_KEY
• LIVEKIT_API_SECRET
• LIVEKIT_TOKEN

from livekit import api

# Mode 1: API key & secret (Backend)
# Uses LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET env vars
lkapi = api.LiveKitAPI()

# Or override env vars explicitly:
lkapi = api.LiveKitAPI("https://my.livekit.host", api_key="api-key", api_secret="api-secret")

# Mode 2: Pre-signed access token (Client-side)
# Requires LIVEKIT_URL to be set in env vars, then pass the token
lkapi = api.LiveKitAPI.with_token("a-pre-signed-token")

The rtc.MediaDevices class provides a high-level interface for audio I/O. It requires the sounddevice library to be installed in your environment.

Capturing Microphone Input

Use devices.open_input() to open a microphone. You can enable Acoustic Echo Cancellation (enable_aec), noise suppression, high-pass filters, and auto-gain control. The resulting source can be used to create a rtc.LocalAudioTrack for publishing.

Playing Audio to Speakers

Use devices.open_output() to create a player. You can add remote audio tracks to this player using player.add_track(track) and then start playback with await player.start().

Full Duplex Audio

For effective echo cancellation in full duplex mode, open the input device with enable_aec=True first, then open the output device. The output player will automatically feed the APM's reverse stream back to the microphone input.

from livekit import rtc

# Capturing Microphone
devices = rtc.MediaDevices()
mic = devices.open_input(enable_aec=True, noise_suppression=True)
track = rtc.LocalAudioTrack.create_audio_track("microphone", mic.source)
await room.local_participant.publish_track(track)

# Playing Audio
player = devices.open_output()
player.add_track(remote_audio_track)
await player.start()

To participate in a live session, use rtc.Room. You can listen to events like participant_connected and track_subscribed.

When a video track is subscribed, you can create an rtc.VideoStream to process incoming frames. By default, autosubscribe is enabled, meaning the participant will automatically subscribe to all published tracks in the room.

from livekit import rtc
import asyncio
import logging

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 stream:
            # process video frame here
            pass

    @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))

    await room.connect(URL, TOKEN)
    logging.info("connected to room %s", room.name)

    # Access existing participants and tracks
    for identity, participant in room.remote_participants.items():
        print(f"identity: {identity}")
        for tid, publication in participant.track_publications.items():
            print(f"\ttrack id: {publication}")