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, models
from decart.realtime import RealtimeClient, RealtimeConnectOptions
from decart.types import ModelState, Prompt

model = models.realtime("mirage_v2")

# 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="Anime", enrich=True),
        ),
    ),
)

# Change style on the fly (async method)
await realtime_client.set_prompt("Cyberpunk city")

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

​
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",
                enrich=True,  # Let Decart enhance the prompt (recommended)
            ),
            mirror=False,  # Set to True for front-facing cameras
        ),
    ),
)
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, enrich)
      • mirror - Enable mirror mode for front-facing cameras

​
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",
    enrich=False
)
Parameters:
  • prompt: str (required) - Text description of desired style
  • enrich: 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.

​
Camera Mirroring

Toggle horizontal mirroring, useful for front-facing cameras: 
# Enable mirror mode (async method)
await realtime_client.set_mirror(True)

# Disable mirror mode
await realtime_client.set_mirror(False)
This flips the video horizontally, making front-facing camera views feel more natural to users.

​
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" | "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)
Use this to update your UI and handle reconnection logic.

​
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.

​
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",
                        enrich=True,
                    ),
                    mirror=True,  # Using front camera
                ),
            ),
        )

        # 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, enrich=True)
        
        # Example: await change_style("Cyberpunk style")

        # Toggle mirror mode (async)
        async def toggle_mirror(enabled):
            await realtime_client.set_mirror(enabled)

        # 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 enrich: 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.
Enable mirror mode when using front-facing cameras to match user expectations from selfie/video call experiences.

​
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, enrich)
      • mirror: bool - Enable mirror mode (default: False)
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", enrich=True),
        ),
    ),
)

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

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

​
await realtime_client.set_mirror(enabled)

Toggles video mirroring (async method). Parameters:
  • enabled: bool - Whether to enable mirror mode
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", "disconnected"]

​
realtime_client.session_id

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

​
await realtime_client.disconnect()

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

​
Events

​
connection_change

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

​
error

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

​
Next Steps

Process API

Learn how to generate and transform media with the Process API

Examples

See complete example applications