Python server SDK

The applooma package wraps the AppLooma RTC REST API for Python backends. It runs on your server, holds your API secret, and issues the join tokens your client apps need.

Install#

pip install applooma

Requires Python 3.9 or newer. The package builds on urllib from the standard library and pulls in no third-party dependencies.

Configure credentials#

Read your key pair from the environment. Never write the secret into source control.

APPLOOMA_API_KEY=ak_your_key
APPLOOMA_API_SECRET=as_your_secret

import os

from applooma import AppServerClient

applooma = AppServerClient(
    api_key=os.environ["APPLOOMA_API_KEY"],
    api_secret=os.environ["APPLOOMA_API_SECRET"],
)

Using os.environ[...] rather than os.getenv means a missing variable raises at import time instead of producing a confusing failure on the first user request.

Requests go to https://api.applooma.dev/v1 with the headers x-applooma-key and x-applooma-secret. Pass base_url= to the constructor for a different region or a self-hosted deployment. An empty key or secret raises ValueError immediately.

A token endpoint#

This is the only endpoint most integrations need. Authenticate your own user first, then decide the identity and role you are willing to grant.

import os

from fastapi import FastAPI, Depends, HTTPException
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel

from applooma import AppServerClient, AppApiError

app = FastAPI()
applooma = AppServerClient(
    api_key=os.environ["APPLOOMA_API_KEY"],
    api_secret=os.environ["APPLOOMA_API_SECRET"],
)


class TokenRequest(BaseModel):
    channel: str


@app.post("/api/rtc-token")
async def rtc_token(body: TokenRequest, user=Depends(current_user)):
    # Decide the role on the server. Never trust a role from the client.
    role = "host" if await is_host(user.id, body.channel) else "audience"

    try:
        # The SDK is synchronous, so keep it off the event loop.
        result = await run_in_threadpool(
            applooma.create_token,
            room_name=body.channel,       # [a-zA-Z0-9_-], 1-64 characters
            identity=user.id,             # unique per participant
            role=role,
            metadata={"name": user.display_name},
        )
    except AppApiError as err:
        if err.code == "QUOTA_EXCEEDED":
            raise HTTPException(503, "Calling is temporarily unavailable")
        if err.status == 400:
            raise HTTPException(400, err.args[0])
        raise HTTPException(502, "Could not start the call")

    return {"token": result["token"], "wsUrl": result["wsUrl"]}

Method reference#

Every method is synchronous and raises AppApiErrorwhen the API answers with a status of 400 or above. Responses are returned as plain dictionaries with the API's own camelCase keys — the SDK does not rename them.

create_token

POST /v1/token — issues a join token for one participant.

create_token(room_name: str, identity: str, role: str = "host", metadata: dict | None = None) -> dict

Returns a dict:

list_rooms

GET /v1/rooms — lists the channels of your project that currently have participants.

list_rooms() -> list

Takes no arguments. Returns a list of dicts; the method unwraps the rooms field and returns [] if it is absent.

list_participants

GET /v1/rooms/{roomName}/participants — lists who is in one channel right now.

list_participants(room_name: str) -> list

The room name is URL-encoded for you. An invalid name returns 400, and a channel that has already closed returns 500.

for room in applooma.list_rooms():
    print(room["roomName"], room["participantCount"])
    for p in applooma.list_participants(room["roomName"]):
        print("  ", p["identity"], p.get("name"), p["joinedAt"])

Error handling#

Failed requests raise AppApiError, a subclass of Exception. It carries the HTTP status and, where the API supplies one, a machine-readable code.

class AppApiError(Exception):
    status: int          # HTTP status
    code: str | None     # machine-readable code, when present
    # str(err) gives the human-readable message from the API

from applooma import AppApiError

try:
    result = applooma.create_token(room_name="lobby", identity=user.id)
except AppApiError as err:
    if err.code == "QUOTA_EXCEEDED":
        # Out of minutes for the month. Alert your team and upgrade the plan.
        alert_ops("AppLooma quota exhausted")
        raise ServiceUnavailable()
    if err.code == "ACCOUNT_SUSPENDED":
        # Billing or policy issue. Only the account owner can resolve it.
        raise ServiceUnavailable()
    if err.status == 400:
        raise BadRequest(str(err))
    if err.status == 401:
        raise RuntimeError("AppLooma credentials are wrong")
    raise

What is not wrapped#

The Python SDK covers tokens and rooms only. Virtual gifting is wrapped in the Node SDK but not in Python, and live streaming ingest, RTMP push and cloud recording are not wrapped in any server SDK yet. Call those endpoints directly with the same two headers.

import os

import requests

BASE = "https://api.applooma.dev/v1"
HEADERS = {
    "Content-Type": "application/json",
    "x-applooma-key": os.environ["APPLOOMA_API_KEY"],
    "x-applooma-secret": os.environ["APPLOOMA_API_SECRET"],
}


def gift_catalog() -> list:
    res = requests.get(f"{BASE}/gifts", headers=HEADERS, timeout=15)
    res.raise_for_status()
    return res.json()["gifts"]


def send_gift(room_name: str, gift_id: str, sender: str, receiver: str, quantity: int = 1) -> dict:
    # Debit the sender in YOUR coin system before calling this.
    res = requests.post(
        f"{BASE}/gifts/send",
        headers=HEADERS,
        json={
            "roomName": room_name,
            "giftId": gift_id,
            "sender": sender,
            "receiver": receiver,
            "quantity": quantity,
        },
        timeout=15,
    )
    if res.status_code >= 400:
        body = res.json()
        raise RuntimeError(f"{res.status_code} {body.get('code')}: {body.get('error')}")
    return res.json()          # {"txId", "coinTotal", "sentAt"}


def create_ingest(room_name: str) -> dict:
    res = requests.post(
        f"{BASE}/streams/ingest",
        headers=HEADERS,
        json={"roomName": room_name, "name": "obs-ingest", "type": "rtmp"},
        timeout=15,
    )
    res.raise_for_status()
    return res.json()          # {"ingestId", "url", "streamKey", "roomName", "status"}

If you prefer no third-party dependency, the same calls work with urllib.request; requests is used here only for brevity.

Security#

• The API secret lives on your server only. It never belongs in a browser bundle, a mobile binary, a mobile app config file, or a repository.
• Anyone holding the secret can issue unlimited tokens against your project and spend your quota.
• Your token endpoint must authenticate the caller. An open token endpoint is the same as publishing your secret.
• Decide the role on the server. Do not copy role straight from the request body, or any audience member can promote themselves to host.
• If a secret leaks, rotate the key pair in the console immediately. Deploy the new pair first, then deactivate the old one.
• Do not log the secret or full tokens. Log err.status and err.code instead.

Next#