Use this SDK to add realtime video, audio and data features to your Python app. By connecting to LiveKit Cloud or a self-hosted server, you can quickly build applications such as multi-modal AI, live streaming, or video calls with just a few lines of code.
This repo contains two packages
- livekit: Real-time SDK for connecting to LiveKit as a participant
- livekit-api: Access token generation and server APIs
$ pip install livekit-apifromlivekitimportapiimportos# will automatically use the LIVEKIT_API_KEY and LIVEKIT_API_SECRET env varstoken=api.AccessToken() \
.with_identity("python-bot") \
.with_name("Python Bot") \
.with_grants(api.VideoGrants(
room_join=True,
room="my-room",
)).to_jwt()RoomService uses asyncio and aiohttp to make API calls. It needs to be used with an event loop.
fromlivekitimportapiimportasyncioasyncdefmain():
lkapi=api.LiveKitAPI("https://my-project.livekit.cloud")
room_info=awaitlkapi.room.create_room(
api.CreateRoomRequest(name="my-room"),
)
print(room_info)
results=awaitlkapi.room.list_rooms(api.ListRoomsRequest())
print(results)
awaitlkapi.aclose()
asyncio.run(main())Authenticate with an API key and secret (recommended for backend use), or with a
pre-signed token for client-side use, where the API secret must not be exposed.
Any omitted value falls back to LIVEKIT_URL, LIVEKIT_API_KEY,
LIVEKIT_API_SECRET, and LIVEKIT_TOKEN.
# API key & secret (backend)lkapi=api.LiveKitAPI("https://my-project.livekit.cloud", api_key="...", api_secret="...")
# pre-signed token (client-side); the token must already carry the grants for the callslkapi=api.LiveKitAPI.with_token(my_token, "https://my-project.livekit.cloud")Services can be accessed via the LiveKitAPI object.
lkapi=api.LiveKitAPI("https://my-project.livekit.cloud")
# Room Serviceroom_svc=lkapi.room# Egress Serviceegress_svc=lkapi.egress# Ingress Serviceingress_svc=lkapi.ingress# Sip Servicesip_svc=lkapi.sip# Agent Dispatchdispatch_svc=lkapi.agent_dispatch# Connector Serviceconnector_svc=lkapi.connectorA failed server API call raises api.ServerError, which exposes the error
code, message, and any server-provided metadata.
try:
awaitlkapi.room.create_room(api.CreateRoomRequest(name="my-room"))
exceptapi.ServerErrorase:
print(e.code, e.message)A failed SIP dial (e.g. the callee is busy or doesn't answer) raises
api.SipCallError, a ServerError subclass that also exposes the SIP response
status:
try:
awaitlkapi.sip.create_sip_participant(api.CreateSIPParticipantRequest(
sip_trunk_id="ST_...",
sip_call_to="+15105550100",
room_name="my-room",
wait_until_answered=True,
))
exceptapi.SipCallErrorase:
print(e) # e.g. "SIP call failed: 486 Busy Here (resource_exhausted)"ife.sip_status_code==486:
... # busyexceptapi.ServerErrorase:
print(e.code, e.message) # any other API error$ pip install livekitsee room_example for full example
fromlivekitimportrtcasyncdefmain():
room=rtc.Room()
@room.on("participant_connected")defon_participant_connected(participant: rtc.RemoteParticipant):
logging.info(
"participant connected: %s %s", participant.sid, participant.identity)
asyncdefreceive_frames(stream: rtc.VideoStream):
asyncforframeinstream:
# received a video frame from the track, process it herepass# track_subscribed is emitted whenever the local participant is subscribed to a new track@room.on("track_subscribed")defon_track_subscribed(track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant):
logging.info("track subscribed: %s", publication.sid)
iftrack.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 roomawaitroom.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 themforidentity, participantinroom.remote_participants.items():
print(f"identity: {identity}")
print(f"participant: {participant}")
fortid, publicationinparticipant.track_publications.items():
print(f"\ttrack id: {publication}")Perform your own predefined method calls from one participant to another.
This feature is especially powerful when used with Agents, for instance to forward LLM function calls to your client application.
The participant who implements the method and will receive its calls must first register support:
@room.local_participant.register_rpc_method("greet")asyncdefhandle_greet(data: RpcInvocationData):
print(f"Received greeting from {data.caller_identity}: {data.payload}")
returnf"Hello, {data.caller_identity}!"In addition to the payload, your handler will also receive response_timeout, which informs you the maximum time available to return a response. If you are unable to respond in time, the call will result in an error on the caller's side.
The caller may then initiate an RPC call like so:
try:
response=awaitroom.local_participant.perform_rpc(
destination_identity='recipient-identity',
method='greet',
payload='Hello from RPC!'
)
print(f"RPC response: {response}")
exceptExceptionase:
print(f"RPC call failed: {e}")You may find it useful to adjust the response_timeout parameter, which indicates the amount of time you will wait for a response. We recommend keeping this value as low as possible while still satisfying the constraints of your application.
The MediaDevices class provides a high-level interface for working with local audio input (microphone) and output (speakers) devices. It's built on top of the sounddevice library and integrates seamlessly with LiveKit's audio processing features. In order to use MediaDevices, you must have the sounddevice library installed in your local Python environment, if it's not available, MediaDevices will not work.
fromlivekitimportrtc# Create a MediaDevices instancedevices=rtc.MediaDevices()
# Open the default microphone with audio processing enabledmic=devices.open_input(
enable_aec=True, # Acoustic Echo Cancellationnoise_suppression=True, # Noise suppressionhigh_pass_filter=True, # High-pass filterauto_gain_control=True# Automatic gain control
)
# Use the audio source to create a track and publish ittrack=rtc.LocalAudioTrack.create_audio_track("microphone", mic.source)
awaitroom.local_participant.publish_track(track)
# Clean up when doneawaitmic.aclose()# Open the default output deviceplayer=devices.open_output()
# Add remote audio tracks to the player (typically in a track_subscribed handler)@room.on("track_subscribed")defon_track_subscribed(track: rtc.Track, publication, participant):
iftrack.kind==rtc.TrackKind.KIND_AUDIO:
player.add_track(track)
# Start playback (mixes all added tracks)awaitplayer.start()
# Clean up when doneawaitplayer.aclose()For full duplex audio with echo cancellation, open the input device first (with AEC enabled), then open the output device. The output player will automatically feed the APM's reverse stream for effective echo cancellation:
devices=rtc.MediaDevices()
# Open microphone with AECmic=devices.open_input(enable_aec=True)
# Open speakers - automatically uses the mic's APM for echo cancellationplayer=devices.open_output()
# Publish microphonetrack=rtc.LocalAudioTrack.create_audio_track("mic", mic.source)
awaitroom.local_participant.publish_track(track)
# Add remote tracks and start playbackplayer.add_track(remote_audio_track)
awaitplayer.start()devices=rtc.MediaDevices()
# List input devicesinput_devices=devices.list_input_devices()
fordeviceininput_devices:
print(f"{device['index']}: {device['name']}")
# List output devices output_devices=devices.list_output_devices()
fordeviceinoutput_devices:
print(f"{device['index']}: {device['name']}")
# Get default device indicesdefault_input=devices.default_input_device()
default_output=devices.default_output_device()See publish_mic.py and full_duplex.py for complete examples.
LiveKit is a dynamic realtime environment and calls can fail for various reasons.
You may throw errors of the type RpcError with a string message in an RPC method handler and they will be received on the caller's side with the message intact. Other errors will not be transmitted and will instead arrive to the caller as 1500 ("Application Error"). Other built-in errors are detailed in RpcError.
The underlying Rust SDK ships with platform-specific hardware-accelerated encoders/decoders, These are used automatically when available and compatible with the runtime environment (OS, drivers, GPU, and codec).
| Platform | Codec(s) | Encoder | Decoder | Backend |
|---|---|---|---|---|
| macOS | H264, H265 | ✓ | ✓ | VideoToolbox |
| Linux (AMD GPU) | H264 | ✓ | VAAPI | |
| Linux x64 (NVIDIA GPU) | H264, H265 | ✓ | ✓ | NVENC / NVDEC (NVIDIA Video Codec SDK) |
Software encoders (libvpx for VP8/VP9, libaom for AV1, OpenH264 for H264) are used as a fallback when hardware acceleration is not available.
Note: Availability depends on the specific machine configuration, including GPU model, driver support, and runtime environment.
- Facelandmark: Use mediapipe to detect face landmarks (eyes, nose ...)
- Basic room: Connect to a room
- Publish hue: Publish a rainbow video track
- Publish wave: Publish a sine wave
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 | |
|---|---|
| LiveKit SDKs | Browser · iOS/macOS/visionOS · Android · Flutter · React Native · Rust · Node.js · Python · Unity · Unity (WebGL) · ESP32 |
| Server APIs | Node.js · Golang · Ruby · Java/Kotlin · Python · Rust · PHP (community) · .NET (community) |
| UI Components | React · Android Compose · SwiftUI · Flutter |
| Agents Frameworks | Python · Node.js · Playground |
| Services | LiveKit server · Egress · Ingress · SIP |
| Resources | Docs · Example apps · Cloud · Self-hosting · CLI |