Python Server SDK for Flask, FastAPI and Django

voltkit-rtc-server (import voltkit_rtc.server) wraps the REST API for a Python backend. Your backend does two things: decide who may do what, and hand the client a short-lived token. It is never in the media path. A Python token server for the video call API is one endpoint: Handling a failed call shows it in Flask, FastAPI and Django.

Install

pip install voltkit-rtc-server

Python 3.9 or newer, standard library only: no dependencies, nothing to pin. Everything is imported from voltkit_rtc.server; voltkit_rtc on its own is a namespace shared by VoltKit's Python packages and holds nothing:

from voltkit_rtc.server import StreamClient, StreamError, parse_webhook, verify_webhook

Replies are the service's JSON as plain dicts and lists, camelCase keys included, typed for your editor with TypedDicts (RoomWithHostToken, RoomToken, ChatMessage, UsageSummary, ProjectBalance, RtcRates, StreamEvent and the rest in voltkit_rtc.server.types; the package ships py.typed). This page describes 0.1.1, at parity with the Node.js SDK 0.4.8; see Versions.

Create a client

import os

from voltkit_rtc.server import StreamClient

rtc = StreamClient(
    os.environ["VOLTKIT_RTC_URL"],  # your API base URL
    os.environ["VOLTKIT_RTC_SECRET"],  # your API secret: server-side only
)

Both values are in the credentials file your browser downloaded when you created the project or the key. The API base URL is also on your project's Credentials tab in the dashboard; the secret is shown only once. See Quickstart.

Create one client when your server starts and share it: it holds no state outside a call and is safe to use from any thread. The constructor checks every option, so a wrong value fails at startup instead of on a user's first call (see Mistakes caught before a request). StreamClient.from_env() reads the same two variables, plus the optional VOLTKIT_API_URL for rates, and nothing else; the plain constructor reads nothing from the environment:

rtc = StreamClient.from_env()  # VOLTKIT_RTC_URL, VOLTKIT_RTC_SECRET and, optionally, VOLTKIT_API_URL

Rooms and tokens

# Create the room, or get it back with a fresh host token if it already exists.
room = rtc.ensure_room("show-42", host_identity="user-7")
# -> {"roomId", "originNode", "hostToken", "wsUrl", "expiresAt", "existed", ...}

# Let someone in, with the role YOU decide.
grant = rtc.create_token("show-42", identity="user-99", role="viewer")
# -> {"token", "wsUrl", "role", "identity", "roomId", "expiresAt"}

view = rtc.get_room("show-42")  # the authoritative view: {"room", "participants", "producers"}
rooms = rtc.list_rooms()["rooms"]  # every live room; list_rooms returns the object, not a list
people = rtc.list_participants("show-42")  # [{"identity", "role", "handRaised", "muted", "joinedAt", "metadata"}, ...]
rtc.delete_room("show-42")  # ends it for everyone

• ensure_room creates the room, or returns the live one with a fresh host token. existed says which, so you can tell a new "go live" from a host reconnecting. create_room is the strict version: it fails with room-exists (409) if the id is taken, and makes up an id if you leave room_id out.
• metadata (any JSON) on either call is room state that every participant receives when they join. It is applied only when the room is created. Left None, it is left out of the body.
• role is host, cohost, moderator or viewer. It is signed into the token, so the client cannot change it. A token lasts ten minutes; a connected client is sent fresh ones, so a call can run for hours.
• Poll get_room for a live count rather than adding up webhook joins and leaves: delivery is at-least-once and unordered, so a counter built from events drifts.
• Room ids are [A-Za-z0-9._-]{1,128} and scoped to your project: another project's room is a 404, never a 403.

Roles are covered in Rooms, and a complete token endpoint is below under Handling a failed call and on Token server.

Chat

# Admit someone to a channel's chat with no media rights (role defaults to viewer).
chat = rtc.create_chat_token("show-42", identity="user-99")
# -> {"token", "wsUrl", "role", "identity", "channel", "expiresAt"}      wsUrl ends in /ws/chat

# Take part without holding a connection.
rtc.send_message("show-42", text="The show starts in 5 minutes", type="system")
rtc.send_message("show-42", type="gift", from_="user-7", metadata={"gift": "rose", "count": 5})
rtc.send_message("show-42", type="reaction", metadata={"emoji": "heart"}, ephemeral=True)  # not stored
rtc.send_message("show-42", to="user-13", text="Please keep it friendly")  # a direct message

messages = rtc.list_messages("show-42", limit=100)  # oldest first
older = rtc.list_messages("show-42", before=messages[0]["id"])  # the page before that
rtc.delete_message("show-42", message_id)
rtc.mute_chat("show-42", "user-13")  # survives reconnecting; they can still read
rtc.unmute_chat("show-42", "user-13")
rtc.list_chat_mutes("show-42")  # -> ["user-13"]
rtc.get_chat_info("show-42")  # -> {"channel", "members"}

A channel shares its name with the room of the same name; a channel with no room behind it is chat on its own.

A message sent from your backend can be trusted in a way a client-sent one cannot, which makes it the right place for gifts, purchases and system notices. from_ (the wire's from, which is a Python keyword) defaults to system, and you may also speak as one of your own users. Backend messages skip the per-connection rate limit and the mute list, which exist to restrain clients. A message needs text or metadata; text is capped at 4096 bytes.

list_messages returns the newest limit messages (50 by default, at most 100), oldest first, direct messages included: your backend owns the conversation. Pass the oldest id you hold as before to page back. send_message is never retried. See Chat.

Usage

usage = rtc.get_usage()
# -> {"minutesRemaining", "today", "month", "lifetime", "allowance", ...}

days = rtc.get_usage_series(30)  # one entry per UTC day, oldest first, ending today; at most 90

Each period carries participantSeconds, chatMessages, broadcastSeconds, broadcastViewerSeconds and more. minutesRemaining is -1 on an unlimited plan. allowance breaks it down: free minutes left today and this month, the paid balance, the daily and monthly limits, and blocked with the reason when the project cannot be used. A day with no usage is present and zero, so a chart drawn from the series has no gaps. See Reading your usage.

Balance and rates

The balance is the VoltKit Wallet a project spends from, after its free and bought minutes.

Balance

balance = rtc.get_balance()
# -> {"tenantId", "billingMode": "wallet", "canJoin": True, "blocked": None,
#     "legacy": {"unlimited", "freeSeconds", "ownSeconds", "sharedSeconds", "totalSeconds"},
#     "wallet": {"spendOrder", "scopes", "paidCu", "promoCu", "availableCu": "186000000", "availableCredits": "31"},
#     "burn": {"liveParticipants", "projectCuPerSecond", "walletCuPerSecond"},
#     "remainingSeconds", "walletLow": {"active", "graceSeconds"}, "pricingVersion", "updatedAt"}

GET /v1/billing/balance: what this project can spend right now, to show a balance, warn before a call runs out, or decide whether to start one. It is read only (recharges happen in the dashboard) and authenticated with your API secret, so call it from your server and hand a browser only what it needs to show.

canJoin says whether a newcomer would be admitted now. When it is False, blocked is the code a join or a token would be refused with, so you can say why before anyone tries:

from typing import Optional


def admission_notice() -> Optional[tuple[int, str]]:
    """What to tell a viewer before they try, or None when they would be admitted."""
    balance = rtc.get_balance()
    if balance["canJoin"]:
        return None
    if balance["blocked"] == "tenant-suspended":
        return 403, "This account is suspended"
    if balance["blocked"] == "billing-unavailable":
        return 503, "Billing is being set up"
    return 402, "Recharge to go live"  # quota-exhausted, daily- or monthly-limit-reached

• legacy are the minutes spent first, in participant-seconds. On an unlimited plan unlimited is True, and ownSeconds and totalSeconds are -1.
• wallet is None unless the project is billed from the wallet (billingMode "wallet"). It lists only the shares this project may spend, in the order it spends them: its own share (project:<id>), the RTC share (product:rtc), then unallocated. Never another project's.
• burn is what this project's rooms cost per second now, and what every project drawing the same wallet costs. remainingSeconds estimates how long the funds last at that rate (-1 when nothing is being spent or nothing limits it).
• walletLow["active"] is True while the project's rooms are being warned. pricingVersion is the price list new sessions start on: None for a legacy project, and while no price list is active.

Amounts are strings. Money is in credit units: 1 credit = 6,000,000 units, and every amount in units (paidCu, availableCu, projectCuPerSecond and the rest) is a decimal string, because a JSON number is a float and rounds a large balance without telling you. Add them up with int, which has no limit; show availableCredits, which is already formatted (at most four decimals, rounded down):

CU_PER_CREDIT = 6_000_000
cu = int(balance["wallet"]["availableCu"]) if balance["wallet"] else 0  # "186000000" -> 186000000
whole_credits, remainder_cu = divmod(cu, CU_PER_CREDIT)  # 31 credits, 0 units over

Rates

rtc = StreamClient(
    os.environ["VOLTKIT_RTC_URL"],
    os.environ["VOLTKIT_RTC_SECRET"],
    rates_base_url="https://api.voltkit.in",  # VoltKit's website API: the origin, without /api/v1
)
rates = rtc.get_rates()
# -> {"version", "kind": "standard", "effectiveAt", "unit",
#     "tiers": [{"id": "video_30", "credits": "99.00", "cuPerParticipantSecond": "9900", ...}, ...],
#     "surcharges", "idleTier", "modes", "conversion", "notes"}

The public price list: each rate in credits per 1,000 participant-minutes, the media modes that are switched on, and the conversion for new recharges. Prices are set by VoltKit on the website, so get_rates() reads GET /api/v1/billing/rtc/rates/current from the website API named by rates_base_url, and sends it no credential: no API key and no Authorization header. The secret only ever goes to base_url. The website caches the list for 30 seconds; cache it on your side too if you read it often.

• Without rates_base_url, get_rates() fails at once with rates-unconfigured (status 0) and sends nothing.
• A rates_base_url that names something other than the website API gets a reply with no rates in it. The call fails with http-<status> and is not retried.
• The website API's own error codes are passed through, such as RATE_LIMITED (429, retried) or NOT_FOUND.
• get_rates() sends no secret, so it may follow redirects (up to five with the default transport); a request with the key never does.

A rate's cuPerParticipantSecond is exact, so a cost is integer arithmetic:

video = next(tier for tier in rates["tiers"] if tier["id"] == "video_30")
cost = int(video["cuPerParticipantSecond"]) * 4 * 30 * 60  # 4 people for 30 minutes: 71280000 units, 11.88 credits
affordable = int(balance["wallet"]["availableCu"]) >= cost if balance["wallet"] else False  # rough: free and bought minutes are spent first

Every method

"Retried" means the SDK tries the call again after a temporary failure (see Timeouts and retries). Keyword-only arguments (after *) are optional; None leaves the field out of the body.

parse_webhook and verify_webhook are functions of their own: see Webhooks. ERROR_CODES, REST_ERROR_CODES, error_info and is_terminal are the shared registry. Ids go into the path percent-encoded, as one segment (show 42 is sent as show%2042); an id of "", "." or ".." is refused before anything is sent. Every request and reply is in the API reference; the stream methods are not offered on VoltKit cloud today.

Timeouts and retries

rtc = StreamClient(
    os.environ["VOLTKIT_RTC_URL"],
    os.environ["VOLTKIT_RTC_SECRET"],
    timeout=5,  # each attempt: headers and body (default 10 seconds)
    retries=1,  # at most one more attempt (default 2; 0 turns retries off)
)

• Timeout. timeout (default 10 seconds) is the deadline for each attempt, from sending the request to reading the last byte of the reply. A service that does not answer, or a reply that stalls after its headers, fails with network and "timed out after 5000ms". The default transport checks it before the connect, before the send and before each read of the body.
• What is retried. A call that is safe to repeat is tried again after a network failure (no answer, a refused or dropped connection, the timeout), a 429 or a 5xx. The waits are about 200 ms, then 400 ms, doubling each time, plus up to 100 ms at random, so a fleet of servers does not retry in step; they are held at 2^31−1 ms however large retries is. Retry-After is not read.
• What is not. Any other 4xx (the service made a decision and will make it again), billing-unavailable (a 503 that lasts until VoltKit acts), redirected, and a 2xx reply that is not JSON.
• How long a call can take. Up to (retries + 1) × timeout, plus the waits between attempts: about 31 seconds with the defaults. In a request a user is waiting on, lower both: see A slow request in a web handler.
• Cancelling. The SDK is synchronous. A KeyboardInterrupt, or any exception raised by a sleep hook of your own, stops the retries at once and reaches you as itself, never as a StreamError.

When an error reaches you with retryable True, the SDK has already spent its retries. Try again later, from a job queue or the user's next action, rather than at once in a loop.

Calls that are never repeated

create_room, create_stream and send_message are sent once, however they fail. If a reply is lost, repeating them would turn a success into a spurious room-exists or stream-exists, or post the message twice. Every other call is safe to repeat: tokens are independent, reads change nothing, ensure_room returns the room that exists, and a mute or a delete done twice ends the same way.

When one of the three fails with retryable True, you decide:

• A 429, whatever its code, is a refusal: nothing was done. Send it again after a pause.
• A network error or a 5xx may have done the work before the reply was lost. Check first:

import time

from voltkit_rtc.server import StreamError


def announce_gift(channel: str, gift_id: str, sender: str, name: str) -> dict:
    """Announce a gift once, even when a reply is lost."""
    args = dict(type="gift", from_=sender, metadata={"giftId": gift_id, "gift": name})
    try:
        return rtc.send_message(channel, **args)
    except StreamError as err:
        if not err.retryable:
            raise
        if err.status != 429:
            # It may have been stored before the reply was lost: look before sending again.
            for message in rtc.list_messages(channel, limit=100):
                if (message.get("metadata") or {}).get("giftId") == gift_id:
                    return message
        time.sleep(1)
        return rtc.send_message(channel, **args)

Deletes that lost their reply

delete_room, delete_message, unmute_chat and delete_stream hold across a lost reply. When an attempt fails in a way that may have reached the service (a network error or a 5xx) and the retry answers 404, the first attempt did the deleting, so the call returns normally. A 404 on the first attempt still raises, as does one after a 429 (which deleted nothing) or one saying chat or broadcast is switched off (chat-disabled, broadcast-disabled).

To treat "already gone" as done, catch it:

def end_show(room_id: str) -> None:
    try:
        rtc.delete_room(room_id)
    except StreamError as err:
        if not err.is_not_found:  # no-room: it had already ended
            raise

Where your API secret goes

The SDK sends your API secret in the X-API-Key header of requests to base_url, and nowhere else:

• get_rates() sends nothing to rates_base_url.
• A redirect is never followed with it (below).
• It never appears in an error, in a log line the SDK writes, or in the client object: see Safe to log.

Every request carries User-Agent: voltkit-rtc-python/<version> and nothing else about your system.

Redirects are never followed

A request that carries your API secret never follows a redirect: following one would hand X-API-Key to whatever host it named. A 3xx answer fails at once with the code redirected: status is the 3xx, the message names where it pointed, and it is not retried, because the same base_url redirects the same way.

If your base_url reaches the service through a redirect (an http:// address upgraded to https://, or a www. rewrite in front), set it to the exact scheme and host of your API base URL, as the Credentials tab shows it. get_rates() sends no secret and may follow redirects.

Safe to log

An error never carries your API secret, or the credentials of a URL (user:pass@ in base_url or rates_base_url), so logger.error("%s", err) and logger.exception(...) can take it whole. Wherever the transport, a dropped connection, a redirect or the service's reply quotes either one, message and code say [redacted] instead, however it was encoded: as is, URL-encoded by any encoder in either hex case, form-encoded, JSON-escaped (\/, /), as HTML character references (/, /, /), or any of those escaped once more (%252F, \"). A key over 256 characters is not looked for: words at least as long as it are shown as [not shown: it could not be checked for the API key]. A long message is cut only after that, so a cut cannot leave part of the secret behind. A network error keeps nothing of the transport's own exception, not even as __context__.

The client object keeps them out of sight too. repr(client), str(client) and vars(client) never show the secret, and show base_url and rates_base_url with any credentials as [redacted]; requests still go to the URLs as given. pickle and copy refuse the client with a TypeError: a client is made where its key is, not moved. A TransportRequest prints with X-API-Key, Authorization and URL credentials as [redacted], for a transport of your own that logs what it sends.

print(rtc)  # StreamClient(base_url='https://api.voltkit.in', rates_base_url='https://api.voltkit.in', timeout=10.0, retries=2)

What a call returns is another matter: token, hostToken and publish["whipUrl"] are credentials for one room or stream. Hand each to the one client that needs it, and never log them.

Errors

A call that fails raises a StreamError, a subclass of Exception. Branch on code or the predicates, and log message. A mistake in your own setup is a TypeError or ValueError instead, raised before anything is sent: see Mistakes caught before a request. Every code from every VoltKit SDK, and what to do about it, is on Errors.

Handling a failed call

A token endpoint, mapping each answer from VoltKit to an answer for the client. With Flask:

import logging
from typing import Optional

from flask import Flask, Response, jsonify, request

from voltkit_rtc.server import StreamClient, StreamError

app = Flask(__name__)
log = logging.getLogger("token-server")
rtc = StreamClient.from_env()  # made once, shared by every request


def current_user() -> Optional[str]:
    """Who is calling. A placeholder: your session or JWT check goes here."""
    header = request.headers.get("Authorization", "")
    if not header.startswith("Bearer "):
        return None
    return header[7:].strip() or None


@app.post("/shows/<show_id>/join")
def join(show_id: str) -> tuple[Response, int]:
    user = current_user()
    if user is None:
        return jsonify(error="sign in first"), 401
    try:
        grant = rtc.create_token("show-%s" % show_id, identity=user, role="viewer")  # your rules decide the role
    except StreamError as err:
        log.error("create_token failed: %s", err)  # safe: the API secret is never in it
        return error_response(err)
    body = jsonify(token=grant["token"], wsUrl=grant["wsUrl"], expiresAt=grant["expiresAt"])
    body.headers["Cache-Control"] = "no-store"  # a token is a credential for one room
    return body, 200


def error_response(err: StreamError) -> tuple[Response, int]:
    """What the user is told, by code. Nothing of VoltKit's wording leaks past here."""
    if err.is_not_found:  # no-room
        return jsonify(error="This show has ended"), 404
    if err.is_out_of_minutes:  # 402: quota-exhausted, daily-limit-reached, monthly-limit-reached
        return jsonify(error="This show cannot take viewers right now"), 402
    if err.code == "viewer-quota":  # 429: the room is full
        return jsonify(error="This show is full"), 429
    if err.retryable:  # network, 429, 5xx: the SDK has already retried
        response = jsonify(error="Try again in a moment")
        response.headers["Retry-After"] = "10"
        return response, 503
    # unauthorized, tenant-suspended, bad-request, redirected: a problem on your side, answered as a 500
    return jsonify(error="Something went wrong"), 500

With FastAPI the endpoint is a plain def, so FastAPI runs the blocking call in its thread pool and the event loop stays free:

from typing import Optional

from fastapi import FastAPI, Header
from fastapi.responses import JSONResponse

from voltkit_rtc.server import StreamClient, StreamError

app = FastAPI()
rtc = StreamClient.from_env()


@app.post("/shows/{show_id}/join")
def join(show_id: str, authorization: Optional[str] = Header(default=None)) -> JSONResponse:
    user = authorization[7:].strip() if authorization and authorization.startswith("Bearer ") else ""
    if not user:
        return JSONResponse({"error": "sign in first"}, status_code=401)
    try:
        grant = rtc.create_token("show-%s" % show_id, identity=user, role="viewer")
    except StreamError as err:
        if err.is_not_found:
            return JSONResponse({"error": "This show has ended"}, status_code=404)
        if err.is_out_of_minutes:
            return JSONResponse({"error": "This show cannot take viewers right now"}, status_code=402)
        if err.code == "viewer-quota":
            return JSONResponse({"error": "This show is full"}, status_code=429)
        if err.retryable:
            return JSONResponse({"error": "Try again in a moment"}, status_code=503, headers={"Retry-After": "10"})
        return JSONResponse({"error": "Something went wrong"}, status_code=500)
    return JSONResponse(
        {"token": grant["token"], "wsUrl": grant["wsUrl"], "expiresAt": grant["expiresAt"]},
        headers={"Cache-Control": "no-store"},
    )

With Django, as a view (path("shows/<str:show_id>/join", join) in your urls.py):

from django.http import HttpRequest, JsonResponse
from django.views.decorators.http import require_POST

from voltkit_rtc.server import StreamClient, StreamError

rtc = StreamClient.from_env()  # at import: one client for the process


@require_POST
def join(request: HttpRequest, show_id: str) -> JsonResponse:
    if not request.user.is_authenticated:
        return JsonResponse({"error": "sign in first"}, status=401)
    try:
        grant = rtc.create_token("show-%s" % show_id, identity="user-%s" % request.user.pk, role="viewer")
    except StreamError as err:
        if err.is_not_found:
            return JsonResponse({"error": "This show has ended"}, status=404)
        if err.is_out_of_minutes:
            return JsonResponse({"error": "This show cannot take viewers right now"}, status=402)
        if err.code == "viewer-quota":
            return JsonResponse({"error": "This show is full"}, status=429)
        if err.retryable:
            return JsonResponse({"error": "Try again in a moment"}, status=503, headers={"Retry-After": "10"})
        return JsonResponse({"error": "Something went wrong"}, status=500)
    response = JsonResponse({"token": grant["token"], "wsUrl": grant["wsUrl"], "expiresAt": grant["expiresAt"]})
    response["Cache-Control"] = "no-store"
    return response

The error mapping is the same everywhere. The SDK's examples/ folder has the Flask and FastAPI endpoints and a webhook receiver as runnable programs.

What a StreamError carries

A StreamError pickles and copies, so one raised in a worker process (multiprocessing, ProcessPoolExecutor, a task queue) reaches the parent with all four attributes. Messages name things as the REST API and every VoltKit SDK do (roomId, baseUrl), so a log line reads the same whichever backend wrote it.

What to do, code by code

Every code a call can fail with. "Retried" means on a call that is safe to repeat; create_room, create_stream and send_message are sent once.

Mistakes caught before a request

These are a TypeError or a ValueError, never a StreamError, so a handler cannot mistake them for an answer from VoltKit. They fail at once, are never retried, and never quote your secret or a URL:

Whitespace around the secret, and around the URLs, is fine: it is trimmed before sending. A timeout above 2^31−1 ms (about 24.8 days), inf included, is held there; a retries of any size is accepted, and the waits stop doubling at the same cap.

Error codes

Every code a VoltKit RTC SDK can report is in one shared registry, exported in its order. REST_ERROR_CODES bounds a metrics label:

from voltkit_rtc.server import ERROR_CODES, REST_ERROR_CODES, error_info, is_terminal

ERROR_CODES["capacity-exhausted"]  # ErrorInfo(origin="server", terminal=False, retryable=True)
error_info("no-room")  # the same lookup; None for a code this release does not know
is_terminal("no-room")  # True: a client SDK must not retry a join refused with it

known = set(REST_ERROR_CODES)  # the codes a call from this SDK can fail with


def metrics_label(err: StreamError) -> str:
    return err.code if err.code in known else "other"  # http-<status> and the website API's codes fall under "other"

Webhooks

VoltKit signs every delivery: X-Stream-Signature: t=<unix seconds>,v1=<hex>, where v1 is HMAC-SHA256 over <timestamp>.<raw body> with your signing secret. parse_webhook verifies and decodes in one step, so a handler cannot forget to check. With Flask, the raw body is request.get_data():

import os

from flask import Flask, request

from voltkit_rtc.server import StreamError, parse_webhook

app = Flask(__name__)
# The current secret, and the previous one during a rotation (see below). Empty entries are skipped.
SECRETS = [s for s in (os.environ.get("VOLTKIT_RTC_WEBHOOK_SECRET"), os.environ.get("VOLTKIT_RTC_WEBHOOK_SECRET_OLD")) if s]


@app.post("/hooks/voltkit")
def hook() -> tuple[str, int]:
    try:
        event = parse_webhook(request.get_data(), request.headers.get("X-Stream-Signature"), SECRETS)
    except StreamError:
        return "", 401  # bad-signature: not from VoltKit, or too old
    # TypeError or ValueError (no secret, not the raw body): a mistake in your setup; let it reach your error handler.
    queue.put(event)  # answer first, do the work after, deduplicating on event["id"]
    return "", 200

With FastAPI or Starlette the raw body is await request.body(); with Django, request.body. Both must be read before anything parses the JSON:

from fastapi import FastAPI, Request, Response

from voltkit_rtc.server import StreamError, parse_webhook

app = FastAPI()


@app.post("/hooks/voltkit")
async def hook(request: Request) -> Response:
    raw = await request.body()
    try:
        event = parse_webhook(raw, request.headers.get("x-stream-signature"), SECRETS)
    except StreamError:
        return Response(status_code=401)
    await queue.put(event)
    return Response(status_code=200)

from django.http import HttpRequest, HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from voltkit_rtc.server import StreamError, parse_webhook


@csrf_exempt  # VoltKit has no CSRF token; the signature is the proof
@require_POST
def hook(request: HttpRequest) -> HttpResponse:
    try:
        event = parse_webhook(request.body, request.headers.get("X-Stream-Signature"), SECRETS)
    except StreamError:
        return HttpResponse(status=401)
    handle_later(event)  # a Celery task, a queue: never the work itself
    return HttpResponse(status=200)

• Pass the raw body. The signature is HMAC-SHA256 over <timestamp>.<raw body>, so a parsed and re-serialised body can never verify. parse_webhook refuses a parsed object with a TypeError, instead of a quiet failure that would look like an attack. The body may be bytes, a bytearray, a memoryview or a str (taken as UTF-8); the header a str, the bytes a raw WSGI or ASGI layer hands over, or a list of either. See Keep the raw body.
• What it checks. The timestamp must be within tolerance_seconds of your clock (default 300, in either direction, inclusive), so a captured request cannot be replayed later, and a v1 signature must match one of your secrets, compared in constant time with hmac.compare_digest, every secret against every signature. To widen the window, pass tolerance_seconds=600; now= replaces the clock in tests. NaN or a negative raises a ValueError instead of switching the check off.
• What it raises. A StreamError with code bad-signature (status 401) when the delivery does not verify: answer 401, and VoltKit retries it. A TypeError (a parsed body, no secret) or a ValueError (a bad tolerance_seconds) is a mistake in your setup: let it reach your error handler.
• verify_webhook takes the same arguments and returns True or False instead of the event.
• The secret is used exactly as given. Nothing is trimmed: a trailing newline from an env file makes every delivery a 401. strip() it yourself when you read it.

The event is a StreamEvent: {"id", "type", "tenantId", "roomId"?, "createdAt", "data"?}; data varies by type. Three things to get right:

• Answer first. A slow handler is retried, so your latency becomes duplicate deliveries. Retries come at 5 s, 30 s, 2 min, 10 min and 30 min; a 410 disables the endpoint.
• Deduplicate on event["id"]. Delivery is at-least-once by design; the id is stable across retries (also in the X-Stream-Delivery header, with X-Stream-Attempt).
• Do not assume order. A participant.left can arrive before the participant.joined it followed; pair them by identity and joinedAt.

Every event type and its fields are on Webhooks. The SDK's examples/webhook_receiver.py is this handler as a runnable program, with a worker thread and an in-memory dedupe.

Rotating the signing secret

When you press Rotate secret with an overlap (24 hours by default, or 1 hour), VoltKit signs every delivery with both the new secret and the previous one until the overlap ends: X-Stream-Signature: t=<ts>,v1=<new>,v1=<previous>. parse_webhook and verify_webhook accept a delivery when any v1 value matches any secret you pass. To rotate without dropping a delivery:

1. Deploy the handler above, which takes a list. With VOLTKIT_RTC_WEBHOOK_SECRET_OLD unset, it verifies exactly as before.
2. Press Rotate secret. Move the current secret to VOLTKIT_RTC_WEBHOOK_SECRET_OLD, put the new one in VOLTKIT_RTC_WEBHOOK_SECRET, and deploy before the overlap ends.
3. After the overlap, remove VOLTKIT_RTC_WEBHOOK_SECRET_OLD.

See Rotating the signing secret.

Logging and threads

The SDK logs nothing unless you ask. The logger voltkit_rtc.server gets one DEBUG line per attempt with the method, the path, the status or the error code, the elapsed time and the attempt number, and never a header, a body, a token or a URL:

import logging

logging.getLogger("voltkit_rtc.server").setLevel(logging.DEBUG)  # e.g. "GET /v1/usage: 200 after 41ms (attempt 1 of 3)"

One client per process, shared by every thread: it holds no state outside a call, and the default transport opens one connection per attempt, so there is no pool to guard. Under asyncio, the SDK is synchronous: run a call in a worker thread so the event loop stays free (FastAPI does this for a plain def endpoint on its own):

import asyncio
from concurrent.futures import ThreadPoolExecutor


async def join(room_id: str, identity: str) -> dict:
    return await asyncio.to_thread(rtc.create_token, room_id, identity, "viewer")


with ThreadPoolExecutor(max_workers=8) as pool:  # or many threads at once, the same client
    tokens = list(pool.map(lambda identity: rtc.create_token("show-42", identity, "viewer"), ["user-1", "user-2", "user-3"]))

Across processes (multiprocessing, a ProcessPoolExecutor, a task queue), make a client in each process rather than passing one: the client refuses to be pickled, since that would copy the secret. A StreamError raised in a worker does pickle, and reaches the parent whole.

Testing

Pass a transport to test your own code without a network. A transport is any object with send(request: TransportRequest) -> TransportResponse; the retries, redaction and parsing sit above it and behave as they do in production. Answer with the service's error body, {"error": "<code>", "message": "..."}, and its status to drive each branch of your error handling; retries=0 keeps a failing test fast:

import json

from voltkit_rtc.server import StreamClient, StreamError, TransportRequest, TransportResponse


class EndedShow:
    """Answers every request as the service does when the room is gone."""

    def __init__(self) -> None:
        self.calls: list[str] = []

    def send(self, request: TransportRequest) -> TransportResponse:
        self.calls.append("%s %s" % (request.method, request.url))  # headers, body (bytes or None), timeout, follow_redirects too
        body = {"error": "no-room", "message": "room not found"}
        return TransportResponse(404, {}, json.dumps(body).encode())


fake = EndedShow()
client = StreamClient("https://rtc.test", "sk_test", transport=fake, retries=0)
try:
    client.create_token("show-42", "user-99", "viewer")
except StreamError as err:
    assert err.is_not_found and err.code == "no-room"
assert fake.calls == ["POST https://rtc.test/v1/rooms/show-42/tokens"]

A transport that raises TimeoutError stands for the deadline passing; an OSError, http.client.HTTPException or voltkit_rtc.server.TransportError for any other failure in transit. sleep= replaces the wait between attempts and jitter= its randomness (a function returning a float in [0, 1)), so a test of your retry handling runs in no time. To test a webhook handler, sign a body yourself and pin the clock:

import hashlib
import hmac

from voltkit_rtc.server import parse_webhook


def signed_header(secret: str, body: bytes, ts: int = 1787000000) -> str:
    digest = hmac.new(secret.encode(), b"%d." % ts + body, hashlib.sha256).hexdigest()
    return "t=%d,v1=%s" % (ts, digest)


body = b'{"id":"evt_1","type":"webhook.test","tenantId":"t","createdAt":1787000000000}'
event = parse_webhook(body, signed_header("whsec_test", body), "whsec_test", now=1787000000)  # the clock at the timestamp
assert event["id"] == "evt_1"

What to do when

The key is wrong

Every call fails with unauthorized (401), not retried. The secret in VOLTKIT_RTC_SECRET is wrong or revoked, or VOLTKIT_RTC_URL is another project's. Check both at startup with a cheap read, so you find out before your users do:

try:
    rtc.get_usage()
except StreamError as err:
    if err.code == "unauthorized":
        raise SystemExit("VoltKit RTC: the API secret or the base URL is wrong (%s)" % err)  # the message never holds the key
    raise

A key with a stray line break or an accented character from a mangled file is refused by the constructor itself, with a TypeError that says to check how the key is loaded.

The account is out of credit or at a limit

New rooms, tokens and messages fail with is_out_of_minutes: quota-exhausted, daily-limit-reached or monthly-limit-reached, all 402 and not retried; rooms already live are warned and then ended. Answer the viewer with 402 and a message the project owner can act on (recharge, or wait for the limit to reset), and read get_balance() to say so before anyone tries: canJoin False and blocked with the code. billing-unavailable (503) is different: wallet billing is not ready, and a retry does not help; contact support.

The room is full

viewer-quota (429, is_quota): the room is at its viewer limit. Tell the viewer it is full; the SDK has already retried, so do not loop. room-quota (429) is your project's concurrent-room limit instead: end rooms you no longer need with delete_room, or raise the limit. rate-limited (429) is your project's API rate limit: spread calls out.

The room has ended

no-room (404, is_not_found): the room is gone, or was never yours. For a viewer, say the show has ended. For the host coming back, call ensure_room with the same room_id: it makes the room again if it is gone, and returns it with a fresh host token if it is not, existed telling you which.

The network drops mid-call

Two different things. A call to the API that loses its connection, before or after the headers, fails with network (status 0, retryable True) and is retried on every call that is safe to repeat; the attempt's deadline covers the body, so a reply that stalls after its headers fails the same way. When the error reaches you, the retries are spent: answer 503 with Retry-After, and try again from a job or the user's next action. For create_room, create_stream and send_message, see A create may or may not have landed. A participant whose connection drops is not your backend's problem: the client SDKs reconnect on their own, and your webhook receives participant.left with reason disconnected only when they do not come back. Your server is never in the media path.

A create may or may not have landed

create_room, create_stream and send_message are sent once. A 429 means nothing was done: send again after a pause. A network error or a 5xx may have done the work before the reply was lost: check first, then repeat, as the table under Calls that are never repeated says. For rooms, ensure_room with your own id is both the check and the repeat.

The webhook does not verify

parse_webhook raises bad-signature, and VoltKit shows the 401s in the delivery log. In order of likelihood: the body was parsed and re-serialised before it reached parse_webhook (a JSON middleware in front of the handler, or request.json instead of request.get_data()); the secret is the wrong one, or has whitespace around it (nothing is trimmed: use it exactly as the dashboard shows it); the secret was rotated and the receiver holds only the old one (pass both, as above); your server's clock is more than five minutes off (t is checked both ways; fix the clock rather than widening tolerance_seconds); the header reached you as several values (pass the list: it is joined with commas). VoltKit retries a 401, so a fix within the retry window drops nothing. A TypeError instead of a 401 is a wiring mistake the SDK refuses to hide: no secret configured, or not the raw body.

Behind a proxy or a redirect

The default transport speaks to base_url directly and ignores HTTP_PROXY and HTTPS_PROXY; the secret travels inside TLS, which a proxy that only forwards the connection never sees. To go through a proxy, or for a corporate CA, mTLS or HTTP/2, pass a transport of your own. HttpClientTransport(ssl_context=...) takes your own ssl.SSLContext for a private CA; for a proxy, httpx is the shortest route:

from typing import Optional

import httpx

from voltkit_rtc.server import StreamClient, TransportRequest, TransportResponse


class HttpxTransport:
    """The SDK over httpx: a proxy, your own TLS settings, HTTP/2."""

    def __init__(self, proxy: Optional[str] = None) -> None:
        self._client = httpx.Client(proxy=proxy)

    def send(self, request: TransportRequest) -> TransportResponse:
        reply = self._client.request(
            request.method,
            request.url,
            headers=dict(request.headers),
            content=request.body,
            timeout=request.timeout,  # httpx's timeout is per socket operation, not the whole exchange
            follow_redirects=request.follow_redirects,  # False with the key: the 3xx must come back as the reply
        )
        return TransportResponse(reply.status_code, dict(reply.headers), reply.content)


rtc = StreamClient(base_url, api_key, transport=HttpxTransport(proxy="http://proxy.internal:3128"))

Whatever the transport, it must hand a 3xx back unfollowed when request.follow_redirects is False, so the key is never sent elsewhere. A base_url that answers with a redirect, such as an http:// address upgraded to https:// or a www. rewrite, fails every call with redirected: set it to the exact scheme and host the Credentials tab shows. A proxy that answers with its own page for an error, or for a success, comes back as http-<status> with the page's first words in the message, redacted.

A slow request in a web handler

A call can take about 31 seconds with the defaults (three attempts of 10 seconds plus the waits), which is longer than a user waits. The SDK is synchronous and has no cancellation token, so set the budget when you make the client: a second client with a short timeout and retries=0 for handlers, and the default one for background jobs, where waiting is fine:

fast = StreamClient.from_env(timeout=3, retries=0)  # for handlers a user is waiting on: at most 3 seconds, one attempt
jobs = StreamClient.from_env()  # for queues and cron: the full budget

Answer 503 with Retry-After when fast reports a retryable error, and never retry in the handler itself.

Rotating a secret

Deploy the receiver with both secrets before pressing Rotate secret, then swap the variables and deploy again before the overlap ends: the steps are under Rotating the signing secret. Rotating the API secret is simpler: issue the new one from the Credentials tab, deploy it, then revoke the old one; an unauthorized after that means some instance still holds the old key.

Versions

voltkit_rtc.server.__version__ is the installed version, and User-Agent: voltkit-rtc-python/<version> goes on every request. Upgrade with pip install -U voltkit-rtc-server; the changelog ships in the sdist. Until 1.0, a breaking change is a minor bump and everything else a patch.

Without the SDK

Nothing in the SDK is required. Any HTTP client in any language can make the same calls, with your secret in X-API-Key: see the API reference. http.client or requests will do, as long as you keep what the SDK keeps: a deadline that covers the body, no redirects with the key, and the key out of your logs.

Help

Questions and bug reports: contact@voltkit.in. Send the error's code and message, which never contain your secret, and never the secret itself. Getting technical support lists what else to include.

Broadcast

Not offered on VoltKit cloud today; here for deployments that enable broadcast. On VoltKit cloud every one of these calls fails with broadcast-disabled (404, is_broadcast_disabled), and the methods are kept for parity with the API and the other server SDKs.

live = rtc.create_stream("show-42", playback="public")
# -> {"streamId", "streamKey", "state": "idle", "playback", "playbackUrls": {"llhls", "webrtc"},
#     "publish": {"whipUrl", "token", "expiresAt"}}      the publish URL is returned ONLY here; treat it like a password

# Mint another publish URL when you need one:
pub = rtc.create_stream_token("show-42", action="publish")  # -> {"whipUrl", "token", "expiresAt"}

# With playback="token", each viewer needs URLs minted for them. action defaults to "play".
play = rtc.create_stream_token("show-42", identity="user-99", ttl_seconds=3600)  # -> {"llhls", "webrtc", "token", "expiresAt"}

rtc.get_stream("show-42")  # state, viewers, peakViewers
rtc.list_streams()  # a list of streams
rtc.update_stream("show-42", playback="token")  # who may watch, or metadata
rtc.delete_stream("show-42")  # takes it off air; its URLs stop working

publish["whipUrl"] carries the publish token, so treat it like a password. ttl_seconds is capped at 24 hours. viewers counts the playback sessions the media server serves itself, not an audience behind a CDN. create_stream is never retried, for the same reason as create_room. See Live broadcast.