Skip to main content
The Realtime API enables you to transform live video streams with minimal latency using WebRTC. Perfect for building camera effects, video conferencing filters, VR/AR applications, and interactive live streaming.

​
Quick Start

from decart import DecartClient, SetInput, models
from decart.realtime import RealtimeClient, RealtimeConnectOptions
from decart.types import ModelState, Prompt

model = models.realtime("lucy_2_rt")

# Get user's camera stream
stream = await get_camera_stream(
    audio=True,
    video={
        "frame_rate": model.fps,
        "width": model.width,
        "height": model.height,
    },
)

# Create client
client = DecartClient(api_key="your-api-key-here")

# Connect to realtime API
realtime_client = await RealtimeClient.connect(
    base_url=client.base_url,
    api_key=client.api_key,
    local_track=stream.video,  # Pass video track
    options=RealtimeConnectOptions(
        model=model,
        on_remote_stream=lambda transformed_stream: (
            # Handle the transformed video in your app
            handle_stream(transformed_stream)
        ),
        initial_state=ModelState(
            prompt=Prompt(text="Transform into this character", enhance=True),
        ),
    ),
)

# Update prompt and reference image atomically (async method)
await realtime_client.set(SetInput(
    prompt="Add silver futuristic sunglasses",
    image="https://example.com/character.jpg",
    enhance=True,
))

# Disconnect when done (async method)
await realtime_client.disconnect()

​
Creating Client Tokens

When your Python backend serves client applications (browsers, mobile apps), create client tokens for secure client-side authentication.
Learn more about client tokens and why they’re important for security.

​
Backend Examples

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

app = FastAPI()

# Create client at module level for connection reuse
decart_client = DecartClient(api_key=os.environ["DECART_API_KEY"])

@app.post("/api/realtime-token")
async def create_realtime_token():
    """Generate a client token for client-side realtime connections."""
    try:
        token = await decart_client.tokens.create()
        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()
Add authentication to your endpoint to ensure only authorized users can request client tokens. The examples above are simplified for clarity.

​
Connecting

​
Getting Camera Access

Request access to the user’s camera using your platform’s camera API: 
stream = await get_camera_stream(
    audio=True,
    video={
        "frame_rate": 25,
        "width": 1280,
        "height": 704,
    },
)
Use the model’s fps, width, and height properties to ensure optimal performance.

​
Establishing Connection

Connect to the Realtime API with your media stream: 
from decart.realtime import RealtimeClient, RealtimeConnectOptions
from decart.types import ModelState, Prompt

realtime_client = await RealtimeClient.connect(
    base_url=client.base_url,
    api_key=client.api_key,
    local_track=stream.video,  # Video track from camera
    options=RealtimeConnectOptions(
        model=models.realtime("mirage_v2"),
        on_remote_stream=lambda transformed_stream: (
            # Display the transformed video
            handle_stream(transformed_stream)
        ),
        initial_state=ModelState(
            prompt=Prompt(
                text="Lego World",
                enhance=True,  # Let Decart enhance the prompt (recommended)
            ),
        ),
    ),
)
Parameters:
  • base_url (required) - API base URL from client
  • api_key (required) - Your Decart API key
  • local_track (required) - Video MediaStreamTrack from camera
  • options (required) - RealtimeConnectOptions with:
    • model (required) - Realtime model from models.realtime()
    • on_remote_stream (required) - Callback that receives the transformed video stream
    • initial_state (optional) - ModelState with:
      • prompt - Prompt object (text, enhance)
      • image - Initial image as bytes, URL string, or file path string (used with live_avatar model)

​
Managing Prompts

Change the transformation style dynamically without reconnecting: 
# Simple prompt with automatic enhancement (async method)
await realtime_client.set_prompt("Anime style")

# Custom detailed prompt without enhancement
await realtime_client.set_prompt(
    "A detailed artistic style with vibrant colors and dramatic lighting",
    enhance=False
)
Parameters:
  • prompt: str (required) - Text description of desired style
  • enhance: bool (optional) - Whether to enhance the prompt (default: True)
Returns: None (async method)
Prompt enhancement uses Decart’s AI to expand simple prompts for better results. Disable it if you want full control over the exact prompt.

​
Unified State Update

The set() method lets you update the prompt, reference image, or both in a single atomic call. For Lucy 2 (lucy_2_rt), it ensures prompt and image changes are applied together. 
from decart import SetInput

# Update prompt only
await realtime_client.set(SetInput(prompt="Anime style", enhance=True))

# Update reference image only
await realtime_client.set(SetInput(image=image_bytes))

# Update both atomically — prompt and image are applied together
await realtime_client.set(SetInput(
    prompt="Transform into this character",
    image="https://example.com/character.jpg",
    enhance=True,
))
Parameters (SetInput):
  • prompt: Optional[str] - Text description of desired style (at least one of prompt or image is required)
  • enhance: bool - Whether to enhance the prompt (default: True)
  • image: Optional[Union[bytes, str]] - Reference image as bytes or URL string, or None to clear
Use set() instead of separate set_prompt() and set_image() calls when you need to change both at once. This avoids intermediate states where only one value has updated.

​
Connection State

Monitor and react to connection state changes: 
# Check state synchronously
is_connected = realtime_client.is_connected()  # boolean
state = realtime_client.get_connection_state()  # "connected" | "connecting" | "generating" | "reconnecting" | "disconnected"

# Listen to state changes
def on_connection_change(state):
    print(f"Connection state: {state}")
    
    if state == "disconnected":
        # Handle disconnection
        show_reconnect_button()
    elif state == "connected":
        # Handle successful connection
        hide_reconnect_button()

realtime_client.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 was lost unexpectedly; the SDK is automatically retrying
  • "disconnected" — Not connected (initial state, after disconnect(), or after reconnect failure)
The SDK automatically reconnects when an unexpected disconnection occurs (e.g., network interruption). During auto-reconnect, the state transitions to "reconnecting" while the SDK retries with exponential backoff (up to 5 attempts). If all retries fail, the state moves to "disconnected" and an error event is emitted.

​
Error Handling

Handle errors with the error event: 
from decart import (
    InvalidAPIKeyError,
    WebRTCError,
    ModelNotFoundError,
    InvalidInputError,
    DecartSDKError,
)

def on_error(error: DecartSDKError):
    print(f"SDK error: {error.__class__.__name__}, {error.message}")

    if isinstance(error, InvalidAPIKeyError):
        show_error("Invalid API key. Please check your credentials.")
    elif isinstance(error, WebRTCError):
        show_error("Connection error. Please check your network.")
    elif isinstance(error, ModelNotFoundError):
        show_error("Model not found. Please check the model name.")
    elif isinstance(error, InvalidInputError):
        show_error(f"Invalid input: {error.message}")
    else:
        show_error(f"Error: {error.message}")

realtime_client.on("error", on_error)
Exception Types:
  • InvalidAPIKeyError - API key is invalid or missing
  • WebRTCError - WebRTC connection failed
  • ModelNotFoundError - Specified model doesn’t exist
  • InvalidInputError - Invalid input parameters
  • DecartSDKError - Base class for all SDK errors
Use isinstance() checks with specific exception types for better error handling.

​
Session Management

Access the current session ID: 
session_id = realtime_client.session_id
print(f"Current session: {session_id}")
This can be useful for logging, analytics, or debugging.

​
Session Viewing (Subscribe)

You can let other clients watch an active realtime session as read-only viewers using the subscribe feature. The producer session exposes a subscribe_token that viewers use to connect.

​
Getting a Subscribe Token

After connecting, the producer’s subscribe_token is automatically populated: 
realtime_client = await RealtimeClient.connect(
    base_url=client.base_url,
    api_key=client.api_key,
    local_track=stream.video,
    options=RealtimeConnectOptions(
        model=model,
        on_remote_stream=handle_stream,
        initial_state=ModelState(prompt=Prompt(text="Anime style")),
    ),
)

# Share this token with viewers (e.g., via your backend)
token = realtime_client.subscribe_token
print(f"Subscribe token: {token}")

​
Subscribing to a Session

Viewers use the subscribe token to receive the transformed video stream in read-only mode: 
from decart import DecartClient
from decart.realtime import RealtimeClient, SubscribeOptions

client = DecartClient(api_key="viewer-api-key")

subscriber = await RealtimeClient.subscribe(
    base_url=client.base_url,
    api_key=client.api_key,
    options=SubscribeOptions(
        token=subscribe_token,  # Token from the producer
        on_remote_stream=lambda stream: handle_viewer_stream(stream),
    ),
)

# Monitor connection state
def on_connection_change(state):
    print(f"Viewer connection: {state}")

subscriber.on("connection_change", on_connection_change)

# Handle errors
def on_error(error):
    print(f"Viewer error: {error}")

subscriber.on("error", on_error)

# Disconnect when done
await subscriber.disconnect()
Subscribe clients are receive-only — they cannot send prompts or modify the session. The subscriber sees exactly what the producer’s on_remote_stream receives.

​
Cleanup

Always disconnect when done to free up resources: 
# Disconnect from the service (async method)
await realtime_client.disconnect()

# Remove event listeners
realtime_client.off("connection_change", on_connection_change)
realtime_client.off("error", on_error)

# Stop the local media stream
stop_camera_stream(stream)
Failing to disconnect can leave WebRTC connections open and waste resources.

​
Complete Example

Here’s a full application with all features: 
from decart import DecartClient, models
from decart.realtime import RealtimeClient, RealtimeConnectOptions
from decart.types import ModelState, Prompt
from decart import DecartSDKError
import asyncio
import os

async def setup_realtime_video():
    try:
        # Get camera stream with optimal settings
        model = models.realtime("mirage_v2")
        stream = await get_camera_stream(
            audio=True,
            video={
                "frame_rate": model.fps,
                "width": model.width,
                "height": model.height,
            },
        )

        # Display input video
        display_input_video(stream)

        # Create client
        client = DecartClient(api_key=os.getenv("DECART_API_KEY"))

        # Connect to Realtime API
        realtime_client = await RealtimeClient.connect(
            base_url=client.base_url,
            api_key=client.api_key,
            local_track=stream.video,
            options=RealtimeConnectOptions(
                model=model,
                on_remote_stream=lambda transformed_stream: (
                    display_output_video(transformed_stream)
                ),
                initial_state=ModelState(
                    prompt=Prompt(
                        text="Studio Ghibli animation style",
                        enhance=True,
                    ),
                ),
            ),
        )

        # Handle connection state changes
        def on_connection_change(state):
            print(f"Status: {state}")
            update_status_ui(state)

        # Handle errors
        def on_error(error: DecartSDKError):
            print(f"Realtime error: {error}")
            show_error_ui(error.message)

        realtime_client.on("connection_change", on_connection_change)
        realtime_client.on("error", on_error)

        # Allow user to change styles (async)
        async def change_style(new_style):
            await realtime_client.set_prompt(new_style, enhance=True)
        
        # Example: await change_style("Cyberpunk style")

        # Cleanup handler (async)
        async def cleanup():
            await realtime_client.disconnect()
            await client.close()
            stop_camera_stream(stream)

        return realtime_client
    except Exception as error:
        print(f"Failed to setup realtime video: {error}")
        raise error

# Initialize
asyncio.run(setup_realtime_video())

​
Best Practices

Always use the model’s fps, width, and height properties when requesting camera access to ensure optimal performance and compatibility.
model = models.realtime("mirage_v2")
stream = await get_camera_stream(
    video={
        "frame_rate": model.fps,
        "width": model.width,
        "height": model.height,
    },
)
For best results, keep enhance=True (default) to let Decart’s AI enhance your prompts. Only disable it if you need exact prompt control.
Always listen to connection_change events to update your UI and handle reconnection logic gracefully.
Always call disconnect() and stop media streams when done to avoid memory leaks and unnecessary resource usage.

​
API Reference

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

Classmethod that connects to the realtime transformation service. Parameters:
  • base_url: str - API base URL (from client.base_url)
  • api_key: str - Your Decart API key
  • local_track: MediaStreamTrack - Video track from camera stream
  • options: RealtimeConnectOptions - Connection options
    • model: ModelDefinition - Realtime model from models.realtime()
    • on_remote_stream: Callable[[MediaStreamTrack], None] - Callback for transformed video
    • initial_state: ModelState (optional) - Initial state
      • prompt: Prompt - Prompt object (text, enhance)
      • image: Optional[Union[bytes, str]] - Initial image as bytes, URL, or file path (used with live_avatar) Returns: RealtimeClient - Connected realtime client instance
Example: 
from decart.realtime import RealtimeClient, RealtimeConnectOptions
from decart.types import ModelState, Prompt

realtime_client = await RealtimeClient.connect(
    base_url=client.base_url,
    api_key=client.api_key,
    local_track=stream.video,
    options=RealtimeConnectOptions(
        model=models.realtime("mirage_v2"),
        on_remote_stream=handle_stream,
        initial_state=ModelState(
            prompt=Prompt(text="Anime", enhance=True),
        ),
    ),
)

​
await realtime_client.set(input)

Updates the prompt, reference image, or both atomically (async method). Parameters:
  • input: SetInput - Object with at least one of:
    • prompt: Optional[str] - Text description of desired style
    • enhance: bool - Whether to enhance the prompt (default: True)
    • image: Optional[Union[bytes, str]] - Reference image, or None to clear
Returns: None

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

Changes the transformation style (async method). Parameters:
  • prompt: str - Text description of desired style
  • enhance: bool - Whether to enhance the prompt (default: True)
Returns: None

​
realtime_client.is_connected()

Check if currently connected. Returns: bool

​
realtime_client.get_connection_state()

Get current connection state. Returns: Literal["connected", "connecting", "generating", "reconnecting", "disconnected"]

​
realtime_client.session_id

The ID of the current realtime inference session. Type: Optional[str]

​
realtime_client.subscribe_token

An opaque token that can be shared with other clients to let them watch this session in read-only mode via RealtimeClient.subscribe(). Type: Optional[str]

​
await realtime_client.disconnect()

Closes the connection and cleans up resources (async method). Returns: None

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

Classmethod that connects to an existing realtime session as a read-only viewer. Parameters:
  • base_url: str - API base URL (from client.base_url)
  • api_key: str - Your Decart API key
  • options: SubscribeOptions - Subscribe options
    • token: str - Subscribe token from the producer’s realtime_client.subscribe_token
    • on_remote_stream: Callable[[MediaStreamTrack], None] - Callback for the video stream
Returns: SubscribeClient - Connected subscribe client The subscribe client exposes:
  • is_connected() - Check connection status
  • get_connection_state() - Get current state
  • await disconnect() - Close the connection
  • on(event, callback) / off(event, callback) - Listen to connection_change and error events

​
Events

​
connection_change

Fired when connection state changes. Callback: (state: Literal["connected", "connecting", "generating", "reconnecting", "disconnected"]) -> None

​
error

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

​
generation_tick

Fired periodically during generation with billing information. Use this to track session duration and display usage to users. Callback: (message: GenerationTickMessage) -> None 
from decart.realtime import GenerationTickMessage

def on_generation_tick(message: GenerationTickMessage):
    print(f"Generation running for {message.seconds} seconds")
    update_billing_ui(message.seconds)

realtime_client.on("generation_tick", on_generation_tick)

​
Next Steps

Process API

Learn how to generate and transform media with the Process API

Examples

See complete example applications