Python SDK

Make tokens and manage rooms, chat and streams from your Python backend. Sync and async.

livqeno is for your Python backend (Django, Flask, FastAPI, or a plain script). It holds your secret API key and talks to the Livqeno REST API for you.

Its main job is making tokens. Your app (web, React, React Native, or Flutter) can't join a call or chat without one, and only a backend holding the API key can make one. In a typical app, this SDK sits behind an endpoint like POST /join-room: your app calls it, your code checks who the user is, and this SDK returns a token for that user.

It works the same way as the TypeScript SDK, with snake_case names. Pick whichever matches your backend's language. Both work with every client SDK.

You need: a project and an API key. See API credentials.

Install

Terminal

pip install livqeno

The package is called livqeno, and that is also the module you import.

Set up the client

On your backend, create one Livqeno instance and reuse it for every request:

import os
from livqeno import Livqeno
 
livqeno = Livqeno(
    api_key=os.environ["LIVQENO_API_KEY"],
    base_url=os.environ["LIVQENO_API_URL"],  # https://api.livqeno.com
)

Pass the API key yourself. The SDK never reads it from the environment on its own. This is on purpose: a silently picked-up variable is how a production key ends up in a development process.

In a short script or a one-off job, use with to close network connections cleanly when you're done:

with Livqeno(
    api_key=os.environ["LIVQENO_API_KEY"],
    base_url=os.environ["LIVQENO_API_URL"],  # https://api.livqeno.com
) as livqeno:
    ...

Async

Use AsyncLivqeno if your backend is async, like FastAPI or an asyncio worker. Using the sync client there would block your event loop on every call.

from livqeno import AsyncLivqeno, CreateTokenParams
 
async def main():
    async with AsyncLivqeno(
        api_key=os.environ["LIVQENO_API_KEY"],
        base_url=os.environ["LIVQENO_API_URL"],  # https://api.livqeno.com
    ) as livqeno:
        token = await livqeno.tokens.create(CreateTokenParams(room=room_id, identity="user-42"))

AsyncLivqeno has the same methods as Livqeno. You just await them. Both use httpx, the package's only dependency.

RTC tokens

Call this when a user is about to join a call, once per user, per join. Take identity from your own login session (for example request.user.id), never from the request body, or any user could join as anyone.

from livqeno import CreateTokenParams, TokenPermissions
 
token = livqeno.tokens.create(
    CreateTokenParams(
        room=room_id,
        identity="user-42",
        permissions=TokenPermissions(join=True, subscribe=True, publish=True),
        expires_in=3600,
    )
)
# {"token": ..., "endpoint": ..., "iceServers": [...], "expiresAt": ...}

Return this whole object to your app, unchanged. The iceServers part is what lets calls work on mobile networks; dropping it is the most common reason calls work on Wi-Fi but fail on cellular.

CreateTokenParams also takes the participant's starting metadata and attributes, and TokenPermissions(update_own_metadata=True) lets the app change its own (Metadata and attributes).

Rooms

room = livqeno.rooms.create("standup", get_or_create=True)  # 2nd caller gets the same room, not a 409
rooms = livqeno.rooms.list()
room = livqeno.rooms.get(room_id)                            # id or name
livqeno.rooms.delete(room_id)
 
participants = livqeno.rooms.participants.list(room_id)      # who's connected right now
participant = livqeno.rooms.participants.get(room_id, "alice")

Pass get_or_create=True so every participant can call rooms.create() when they join. The first call creates the room; every call after that just returns the same room instead of failing. Pass e2ee=True if you want the room to require end-to-end encryption (E2EE).

Moderating a live call

These act on someone who is connected right now (otherwise RAVEN_PARTICIPANT_NOT_CONNECTED, 404). room_id takes the room's id or its name. Every method is on AsyncLivqeno too.

livqeno.rooms.mute_participant(room_id, "alice", muted=True, kind="audio")  # or track_id=...
livqeno.rooms.update_participant_permissions(room_id, "alice", publish=False)  # returns the result
livqeno.rooms.remove_participant(room_id, "alice")  # doesn't revoke their token
 
# Metadata and attributes: None deletes an attribute
livqeno.rooms.update_participant(room_id, "alice", metadata='{"role":"host"}', attributes={"hand": None})
 
# A data message from the server, never retried by the SDK
livqeno.rooms.send_data(room_id, "meeting ends in 5 min", topic="system")

Webhooks

from livqeno import LivqenoError, verify_webhook
 
try:
    event = verify_webhook(raw_body, request.headers.get("swalvo-signature"), os.environ["LIVQENO_WEBHOOK_SECRET"])
except LivqenoError as err:
    ...  # err.code: RAVEN_WEBHOOK_INVALID_SIGNATURE / _TIMESTAMP_OUT_OF_TOLERANCE / _INVALID_PAYLOAD

Pass the raw body (bytes or str, before any JSON parsing). It returns the parsed event as a dict. tolerance_seconds defaults to 300, and secret can be a list while you rotate. See Webhooks.

Chat

For chat, your backend creates conversations, decides who is a member, and makes a chat token for each user. The browser never creates conversations or adds members itself. Example: a support app where alice (an agent) and bob (a customer) share a conversation.

The most important method is create_token():

Your backend checks who the user is (your own login).
It asks Livqeno for a token for that one user.
Only that token goes to the browser.

from livqeno import CreateChatTokenParams, CreateConversationParams, ConversationMember, SendChatMessageParams
 
conversation = livqeno.chat.create_conversation(
    CreateConversationParams(
        name="support-room-42",
        members=[ConversationMember(user_id="alice", role="ADMIN"), ConversationMember(user_id="bob")],
    )
)
 
token = livqeno.chat.create_token(
    CreateChatTokenParams(user_id=request.user.id, conversations=[conversation["publicId"]], expires_in=3600)
)
# Hand token["token"] to the browser. Nothing else.

Members, messages, and moderation. You'd use these for things like adding a supervisor to a support chat, posting an automatic notice, or removing a bad message:

from livqeno import ListChatMessagesParams
 
livqeno.chat.add_member(room, "carol", role="MODERATOR")
livqeno.chat.remove_member(room, "carol")
 
livqeno.chat.send_message(room, SendChatMessageParams(
    sender_id="system", text="Maintenance in 5 minutes", type="system", # server-only
))
 
page = livqeno.chat.list_messages(room, ListChatMessagesParams(limit=50))
livqeno.chat.delete_message("msg_abc")  # soft delete, keeps an audit trail

Live streaming

A live stream is one main host, optional co-hosts, and many viewers who only watch, like a webinar. On your backend: create the stream, start it, then give each person the right credential.

from livqeno import AddHostParams, CreateLiveStreamParams
 
stream = livqeno.live_streams.create(
    CreateLiveStreamParams(title="Friday Q&A", host_identity="user-123")
)
livqeno.live_streams.start(stream["id"])
 
# Registering a co-host mints full-publish RTC + moderator chat
# credentials in one call. Hand the result to the client SDK unchanged.
host_credential = livqeno.live_streams.add_host(stream["id"], AddHostParams(identity="user-456"))
 
# A viewer token is always subscribe-only — there is no field here that
# can request publish access.
viewer_credential = livqeno.live_streams.create_viewer_token(stream["id"], "user-789")
 
livqeno.live_streams.remove_host(stream["id"], "user-456")
livqeno.live_streams.end(stream["id"])  # LIVE -> ENDED, terminal

livqeno.live_streams is the same as livqeno.liveStreams in TypeScript, with snake_case names. It also works on AsyncLivqeno.live_streams.
The method you call decides the role: add_host() makes a host, create_viewer_token() makes a viewer. Nothing in the request can change that.

More: Live Streaming Overview and SDK Support Matrix.

Live transcription & AI voice agents

transcription = livqeno.transcriptions.start(
    room_id, language="en", provider="deepgram", persist=True
)
segment = livqeno.transcriptions.transcript(transcription["id"])
livqeno.transcriptions.stop(room_id)
 
agent = livqeno.agents.create(room_id, {
    "name": "Support bot",
    "instructions": "You are a friendly support agent for Acme. Keep answers short.",
    "tts": {"provider": "cartesia"},
})
livqeno.agents.say(agent["id"], "One moment while I look that up.")
livqeno.agents.interrupt(agent["id"])
livqeno.agents.delete(agent["id"])

livqeno.transcriptions also has get() to check how a transcription is doing right now, export() to download it as "vtt" or "txt", and erase_participant(identity) to delete everything one participant said (for a GDPR request).
livqeno.agents also has list() and get() to look up agents, and update() to change a running agent's instructions, listen_to, or interruptible setting.
If you pass provider_keys, Livqeno uses them for that one request only and never stores or logs them.

More: Live Transcription & Captions and AI Voice Agents.

Account and observability

project = livqeno.projects.get()
connections = livqeno.connections.list()       # active RTC/chat sockets
connection = livqeno.connections.get(connection_id)
errors = livqeno.errors.list()                 # logged API/SFU errors
metrics = livqeno.metrics.get()
diagnostics = livqeno.diagnostics.get()        # dependency health

These calls aren't about any single room or stream — they look at your whole project: its live connections, past errors, usage metrics, and the health of Livqeno's own dependencies. You can see the same data in the dashboard, or through the Observability API. Every method here also works on AsyncLivqeno, with the same names.

Errors

Every failed call raises a LivqenoError. Log its request_id: support can use it to find the exact request.

from livqeno import LivqenoError
 
try:
    livqeno.rooms.get("2f1b6c3e-8d4a-4b7e-9c21-5a0e7d3f9b10")  # room ids are UUIDs
except LivqenoError as error:
    print(error.code)         # e.g. "RAVEN_ROOM_NOT_FOUND"
    print(error.status_code)  # e.g. 404
    print(error.request_id)   # matches the API's x-request-id

Errors never contain server stack traces, database errors, or RTC/TURN credentials. All codes: Error Codes.

Retries, timeouts, and lists

Setting	Behavior
Retries	Only temporary failures: network errors, timeouts, 429, 502, 503, 504. Uses exponential backoff. `max_retries` defaults to 2. Other 4xx errors (400, 401, 403, 404, …) are never retried, because the same bad request would just fail again.
Timeouts	`timeout` in seconds, default 10, for every request.
Lists	Chat history uses cursors (`nextCursor` / `previousCursor`, no `offset`). Every other list returns a plain list. You can cap it with `limit`.

Security

The API key is stored in a hidden attribute. It doesn't show in vars() or repr(), and it's never logged or put in an error. So printing the client while debugging won't leak it. See Authentication.

Common mistakes

Mistake	What you'll see	Fix
Taking the user id from the request body	Any user can join as anyone else.	Use your login session, like `request.user.id`.
Creating a room without `get_or_create=True`	`RAVEN_CONFLICT` (409) for the second person who joins.	Pass `get_or_create=True`.
Using `Livqeno` inside FastAPI handlers	Slow responses: each call blocks the event loop.	Use `AsyncLivqeno`.
Removing `iceServers` from the token response	Calls work on Wi-Fi but fail on mobile data.	Return the whole object as-is.
Forgetting `import os` or `base_url`	`NameError`, or requests go to `localhost:4100`.	Set `LIVQENO_API_URL` and pass it as `base_url`.

Next steps

Build a token endpoint and a working call: Quickstart (choose the Python tab).
See everything a token can allow: Access tokens.
Set up the app side: Next.js, React, React Native, or Flutter.