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_sdk import create_decart_client, models

model = models.realtime("mirage")

# 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 and connect
client = create_decart_client(
    api_key="your-api-key-here",
)

realtime_client = await client.realtime.connect(stream, {
    "model": model,
    "on_remote_stream": lambda transformed_stream: (
        # Handle the transformed video in your app
        handle_stream(transformed_stream)
    ),
    "initial_state": {
        "prompt": {
            "text": "Anime",
            "enrich": True,
        },
    },
})

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

# Disconnect when done
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: 
realtime_client = await client.realtime.connect(stream, {
    "model": models.realtime("mirage"),
    "on_remote_stream": lambda transformed_stream: (
        # Display the transformed video
        handle_stream(transformed_stream)
    ),
    "initial_state": {
        "prompt": {
            "text": "Lego World",
            "enrich": True,  # Let Decart enhance the prompt (recommended)
        },
        "mirror": False,  # Set to True for front-facing cameras
    },
})
Parameters:
  • stream (required) - MediaStream from get_camera_stream
  • model (required) - Realtime model from models.realtime()
  • on_remote_stream (required) - Callback that receives the transformed video stream
  • initial_state.prompt (required) - Initial style prompt
    • text - Style description
    • enrich - Whether to auto-enhance the prompt (default: True)
  • initial_state.mirror (optional) - Enable mirror mode for front-facing cameras

​
Managing Prompts

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

# Custom detailed prompt without enhancement
realtime_client.set_prompt(
    "A detailed artistic style with vibrant colors and dramatic lighting",
    {"enrich": False}
)
Parameters:
  • prompt (required) - Text description of desired style
  • options.enrich (optional) - Whether to enhance the prompt (default: True)
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
realtime_client.set_mirror(True)

# Disable mirror mode
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_sdk import DecartSDKError

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

    if error.code == "INVALID_API_KEY":
        show_error("Invalid API key. Please check your credentials.")
    elif error.code == "WEB_RTC_ERROR":
        show_error("Connection error. Please check your network.")
    elif error.code == "MODEL_NOT_FOUND":
        show_error("Model not found. Please check the model name.")
    else:
        show_error(f"Error: {error.message}")

realtime_client.on("error", on_error)
Error Codes:
  • INVALID_API_KEY - API key is invalid or missing
  • WEB_RTC_ERROR - WebRTC connection failed
  • MODEL_NOT_FOUND - Specified model doesn’t exist
  • INVALID_INPUT - Invalid input parameters

​
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
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_sdk import create_decart_client, models, DecartSDKError
import asyncio
import os

async def setup_realtime_video():
    try:
        # Get camera stream with optimal settings
        model = models.realtime("mirage")
        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 = create_decart_client(
            api_key=os.getenv("DECART_API_KEY"),
        )

        # Connect to Realtime API
        realtime_client = await client.realtime.connect(stream, {
            "model": model,
            "on_remote_stream": lambda transformed_stream: (
                display_output_video(transformed_stream)
            ),
            "initial_state": {
                "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
        def change_style(new_style):
            realtime_client.set_prompt(new_style, {"enrich": True})
        
        # Example: change_style("Cyberpunk style")

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

        # Cleanup handler
        def cleanup():
            realtime_client.disconnect()
            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")
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

​
client.realtime.connect(stream, options)

Connects to the realtime transformation service. Parameters:
  • stream: MediaStream - MediaStream from get_camera_stream
  • options.model: ModelDefinition - Realtime model from models.realtime()
  • options.on_remote_stream: Callable[[MediaStream], None] - Callback for transformed video stream
  • options.initial_state.prompt: dict - Initial transformation prompt
    • text: str - Style description (required)
    • enrich: bool - Whether to enhance the prompt (default: True)
  • options.initial_state.mirror: bool - Enable mirror mode (default: False)
Returns: RealtimeClient - Connected realtime client instance

​
realtime_client.set_prompt(prompt, options?)

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

​
realtime_client.set_mirror(enabled)

Toggles video mirroring. Parameters:
  • enabled: bool - Whether to enable mirror mode

​
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

​
realtime_client.disconnect()

Closes the connection and cleans up resources.

​
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
⌘I