Skip to main content
The Realtime API transforms a live video track with minimal latency. Signaling is a Decart-owned WebSocket; media flows through a LiveKit room that the SDK joins for you. Pair it with any source that can produce LiveKit video frames — a camera, a file, or synthetic frames.

​
Installation

The realtime client lives behind an optional extra: 
pip install decart[realtime]
This pulls in livekit (LiveKit’s Python client) alongside the core SDK. The decart.realtime module won’t import without it.

​
Quick Start

import asyncio
import os
from decart import DecartClient, SetInput, models
from decart.realtime import RealtimeClient, RealtimeConnectOptions
from decart.types import ModelState, Prompt
from livekit import rtc

async def main():
    async with DecartClient(api_key=os.environ["DECART_API_KEY"]) as client:
        model = models.realtime("lucy-2.1")

        # Build a LiveKit local video track from your own source (camera, file,
        # synthetic frames, ...). See "Creating a video track" below.
        source = rtc.VideoSource(model.width, model.height)
        local_track = rtc.LocalVideoTrack.create_video_track("input", source)

        def on_remote_stream(track: rtc.RemoteVideoTrack):
            async def consume():
                async for event in rtc.VideoStream(track):
                    handle_frame(event.frame)
            asyncio.create_task(consume())

        realtime = await RealtimeClient.connect(
            base_url=client.realtime_base_url,
            api_key=client.api_key,
            local_track=local_track,
            options=RealtimeConnectOptions(
                model=model,
                on_remote_stream=on_remote_stream,
                initial_state=ModelState(
                    prompt=Prompt(text="A cyberpunk cityscape", enhance=True),
                ),
            ),
        )

        # Update the style mid-session — replaces the entire session state.
        await realtime.set(SetInput(prompt="A sunny beach", enhance=True))

        # ... feed frames into `source.capture_frame(...)` from your own loop ...

        await realtime.disconnect()

asyncio.run(main())

​
Creating Client Tokens

When your Python backend serves browser or mobile clients, mint short-lived client tokens instead of shipping your permanent API key.

​
Backend Examples

from fastapi import FastAPI, HTTPException
from decart import DecartClient
import os

app = FastAPI()
decart_client = DecartClient(api_key=os.environ["DECART_API_KEY"])

@app.post("/api/realtime-token")
async def create_realtime_token():
    try:
        token = await decart_client.tokens.create(
            expires_in=300,                # 5 minutes
            allowed_models=["lucy-2.1"],   # restrict to this model
        )
        return {"apiKey": token.api_key, "expiresAt": token.expires_at}
    except Exception:
        raise HTTPException(status_code=500, detail="Failed to create client token")

@app.on_event("shutdown")
async def shutdown():
    await decart_client.close()

from flask import Flask, jsonify
from decart import DecartClient
import asyncio
import os

app = Flask(__name__)
decart_client = DecartClient(api_key=os.environ["DECART_API_KEY"])

@app.route("/api/realtime-token", methods=["POST"])
def create_realtime_token():
    try:
        token = asyncio.run(decart_client.tokens.create(
            expires_in=300,
            allowed_models=["lucy-2.1"],
        ))
        return jsonify({"apiKey": token.api_key, "expiresAt": token.expires_at})
    except Exception:
        return jsonify({"error": "Failed to create client token"}), 500
Authenticate the token endpoint before issuing — anyone who can call it can spend credits against your account.

​
Connecting

​
Creating a video track

The SDK consumes a LiveKit LocalVideoTrack. Build one from any source that can produce RGB frames at the model’s width × height × fps: 
from livekit import rtc

source = rtc.VideoSource(model.width, model.height)
local_track = rtc.LocalVideoTrack.create_video_track("input", source)

# Push frames at the model's fps from your own loop. RGB24 example:
source.capture_frame(rtc.VideoFrame(
    width=model.width,
    height=model.height,
    type=rtc.VideoBufferType.RGB24,
    data=rgb_bytes,
))
The examples/ folder in the SDK shows two common shapes: For a real camera, use any LiveKit-compatible capturer or feed frames yourself.
Use the model’s fps, width, and height for the source so capture matches what the server expects.

​
Establishing the connection

realtime = await RealtimeClient.connect(
    base_url=client.realtime_base_url,
    api_key=client.api_key,
    local_track=local_track,
    options=RealtimeConnectOptions(
        model=models.realtime("lucy-2.1"),
        on_remote_stream=on_remote_stream,
        initial_state=ModelState(
            prompt=Prompt(text="Studio Ghibli style", enhance=True),
            image=character_image_bytes,  # optional reference image
        ),
    ),
)
Parameters:
  • base_url (required) — client.realtime_base_url (wss://api3.decart.ai by default)
  • api_key (required) — your Decart API key or a client token
  • local_track (required) — a LiveKit LocalVideoTrack. Pass None only when subscribing to an existing session.
  • options: RealtimeConnectOptions:
    • model (required) — from models.realtime(...)
    • on_remote_stream (required) — callback receiving the LiveKit RemoteVideoTrack for the transformed output
    • initial_state (optional) — ModelState with prompt and/or image
    • resolution (optional) — "720p" (default) or "1080p" on supported models
    • preferred_video_codec (optional) — "h264" (default) or "vp9". Picks the codec offered to the LiveKit publisher.
Set initial_state.prompt and/or image so the very first frame is already transformed — otherwise viewers briefly see the raw camera feed.

​
Consuming the remote track

on_remote_stream hands you a LiveKit RemoteVideoTrack. Wrap it with rtc.VideoStream to receive decoded frames: 
def on_remote_stream(track: rtc.RemoteVideoTrack):
    async def consume():
        async for event in rtc.VideoStream(track):
            frame = event.frame  # rtc.VideoFrame
            # ... render, encode, or relay frame.data ...
    asyncio.create_task(consume())

​
Output resolution

Opt into 1080p output from supported models; otherwise the server returns 720p: 
options = RealtimeConnectOptions(
    model=models.realtime("lucy-2.1"),
    on_remote_stream=on_remote_stream,
    resolution="1080p",
)

​
Codec preference

preferred_video_codec controls the codec the SDK offers to the LiveKit publisher: 
options = RealtimeConnectOptions(
    model=models.realtime("lucy-2.1"),
    on_remote_stream=on_remote_stream,
    preferred_video_codec="vp9",  # default: "h264"
)

​
Managing Prompts

Change the transformation style mid-session. set_prompt is async and waits for the server ack — it raises on ack failure or timeout. 
await realtime.set_prompt("Anime style")

# Skip prompt enhancement for full control over the exact wording.
await realtime.set_prompt(
    "A detailed artistic style with vibrant colors and dramatic lighting",
    enhance=False,
)
Parameters:
  • prompt: str — style description
  • enhance: bool — auto-enhance the prompt (default: True)
Prompt enhancement uses Decart’s AI to expand simple prompts for better results. Disable it when you need exact prompt control.

​
Unified State Update

set() replaces the entire session state in a single atomic call. Fields you omit are cleared. 
from decart import SetInput

# Prompt only (clears any previously set image)
await realtime.set(SetInput(prompt="Anime style", enhance=True))

# Image only (clears any previously set prompt)
await realtime.set(SetInput(image=image_bytes))

# Both together
await realtime.set(SetInput(
    prompt="Transform into this character",
    image="https://example.com/character.jpg",
    enhance=True,
))
SetInput fields:
  • prompt: Optional[str] — at least one of prompt or image is required
  • enhance: bool — auto-enhance the prompt (default: True)
  • image: Optional[Union[bytes, str]] — reference image as raw bytes, a URL, a data URL, a file path, or raw base64. None clears it.
Prefer set() over separate set_prompt() and set_image() calls when you’re changing both — it avoids intermediate states the model briefly renders.

​
Connection State

state = realtime.get_connection_state()
is_connected = realtime.is_connected()

def on_connection_change(state):
    if state == "disconnected":
        show_reconnect_button()
    elif state == "connected":
        hide_reconnect_button()

realtime.on("connection_change", on_connection_change)
Connection States:
  • "connecting" — initial connection in progress
  • "connected" — connected and ready to send prompts
  • "generating" — actively generating transformed video (sticky until disconnected)
  • "reconnecting" — connection lost; the SDK is automatically retrying
  • "disconnected" — initial state, after disconnect(), or after reconnect failure
The SDK retries with exponential backoff on unexpected disconnects (up to 5 attempts). If every retry fails, the state transitions to "disconnected" and an error event fires.

​
Error Handling

from decart import (
    DecartSDKError,
    InvalidAPIKeyError,
    InvalidInputError,
    ModelNotFoundError,
    WebRTCError,
)

def on_error(error: DecartSDKError):
    if isinstance(error, InvalidAPIKeyError):
        show_error("Invalid API key.")
    elif isinstance(error, WebRTCError):
        show_error("Connection error. Check your network.")
    elif isinstance(error, ModelNotFoundError):
        show_error("Model not found.")
    elif isinstance(error, InvalidInputError):
        show_error(f"Invalid input: {error.message}")
    else:
        show_error(f"Error: {error.message}")

realtime.on("error", on_error)
Exception Types:
  • InvalidAPIKeyError — API key invalid or missing
  • WebRTCError — LiveKit / signaling / ICE failure
  • ModelNotFoundError — model not found
  • InvalidInputError — invalid input parameter
  • DecartSDKError — base class for all SDK errors

​
Generation Ticks

Track session duration for billing or usage display: 
from decart.realtime import GenerationTickMessage

def on_generation_tick(message: GenerationTickMessage):
    update_billing_ui(message.seconds)

realtime.on("generation_tick", on_generation_tick)

​
Session Identifiers

Once the LiveKit room info arrives, the SDK populates a session id and a subscribe token: 
session_id = realtime.session_id          # for logging / analytics
subscribe_token = realtime.subscribe_token # share with viewers for read-only watching

​
Session Viewing (Subscribe)

Other clients can watch an active session as read-only viewers. The producer’s subscribe_token is the handle viewers connect with. 
from decart import DecartClient
from decart.realtime import RealtimeClient, SubscribeOptions

async with DecartClient(api_key=viewer_api_key) as client:
    subscriber = await RealtimeClient.subscribe(
        base_url=client.realtime_base_url,
        api_key=client.api_key,
        options=SubscribeOptions(
            token=subscribe_token,           # from the producer
            on_remote_stream=on_remote_stream,
        ),
    )

    subscriber.on("connection_change", lambda s: print(f"Viewer: {s}"))
    subscriber.on("error", lambda e: print(f"Viewer error: {e}"))

    # ... later
    await subscriber.disconnect()
Subscribers are receive-only — they cannot send prompts or images. They see exactly what the producer’s on_remote_stream receives.

​
Cleanup

await realtime.disconnect()

realtime.off("connection_change", on_connection_change)
realtime.off("error", on_error)
Forgetting to disconnect() leaves the LiveKit room open and continues to consume credits.

​
Best Practices

Build rtc.VideoSource(model.width, model.height) and push frames at model.fps. Mismatched dimensions get rescaled server-side and waste bandwidth.
Pass initial_state.prompt and/or initial_state.image to connect() so the very first generated frame is already transformed. Without it, viewers briefly see the raw input.
When changing prompt and image together, prefer set(SetInput(prompt=..., image=...)) over separate set_prompt / set_image calls — it avoids the intermediate state the model would otherwise render.
await realtime.disconnect() from a finally block or async-context exit. The LiveKit room and HTTP session aren’t reclaimed until you do.
The SDK targets Python ≥ 3.10. The decart[realtime] extra adds livekit and tenacity on top of the core dependencies.

​
API Reference

​
await RealtimeClient.connect(base_url, api_key, local_track, options)

Connects to the realtime transformation service. Returns a connected RealtimeClient. Parameters:
  • base_url: str — client.realtime_base_url
  • api_key: str — your Decart API key or a client token
  • local_track: Optional[rtc.LocalVideoTrack] — input video track
  • options: RealtimeConnectOptions:
    • model: ModelDefinition — from models.realtime(...)
    • on_remote_stream: Callable[[rtc.RemoteVideoTrack], None] — callback for the transformed track
    • initial_state: Optional[ModelState] — initial prompt / image
    • resolution: Optional[Literal["720p", "1080p"]] — output resolution
    • preferred_video_codec: Literal["h264", "vp9"] — codec preference (default "h264")
Raises WebRTCError on signaling, ICE, or LiveKit room failures.

​
await realtime.set(input)

Replaces the entire session state atomically. Fields not included are cleared. Parameters:
  • input: SetInput — at least one of:
    • prompt: Optional[str]
    • enhance: bool (default True)
    • image: Optional[Union[bytes, str]] — bytes, URL, data URL, file path, or raw base64

​
await realtime.set_prompt(prompt, enhance=True)

Changes the prompt and waits for the server ack. Raises DecartSDKError on ack failure or timeout, InvalidInputError on empty prompt.

​
await realtime.set_image(image, prompt=None, enhance=True, timeout=30.0)

Sends (or clears, with image=None) a reference image. Optionally bundles a prompt.

​
realtime.is_connected() / realtime.get_connection_state()

Synchronous accessors for the current state ("connected" | "connecting" | "generating" | "reconnecting" | "disconnected").

​
realtime.session_id / realtime.subscribe_token

Optional[str] properties populated once the LiveKit room info arrives.

​
await realtime.disconnect()

Closes the LiveKit room and the internal HTTP session.

​
await RealtimeClient.subscribe(base_url, api_key, options)

Classmethod that connects to an existing session as a read-only viewer. Returns a SubscribeClient exposing is_connected(), get_connection_state(), disconnect(), and the same on(...) / off(...) event API. SubscribeOptions:
  • token: str — subscribe token from the producer
  • on_remote_stream: Callable[[rtc.RemoteVideoTrack], None]

​
Events

​
connection_change

Fires when the connection state changes. Callback signature: (state: ConnectionState) -> None.

​
error

Fires when an error occurs. Callback signature: (error: DecartSDKError) -> None.

​
generation_tick

Fires periodically during generation with billing info. Callback signature: (message: GenerationTickMessage) -> None. message.seconds is the running session duration.

​
Next Steps

Process API

Generate and transform media synchronously with the Process API

Examples

See complete example applications