> ## Documentation Index
> Fetch the complete documentation index at: https://docs.platform.decart.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Realtime API

> Transform video streams in realtime with WebRTC

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

```python theme={null}
from decart import DecartClient, SetInput, models
from decart.realtime import RealtimeClient, RealtimeConnectOptions
from decart.types import ModelState, Prompt

model = models.realtime("lucy-2.1")

# 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")

with open("character.jpg", "rb") as f:
    character_image = f.read()

# 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),
            image=character_image,
        ),
    ),
)

# Update prompt and reference image later 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.

<Info>
  Learn more about [client tokens](/getting-started/client-tokens) and why they're important for security.
</Info>

### Backend Examples

<CodeGroup>
  ```python FastAPI theme={null}
  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(
              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()
  ```

  ```python Flask theme={null}
  from flask import Flask, jsonify
  from decart import DecartClient
  import asyncio
  import os

  app = Flask(__name__)

  # Create client
  decart_client = DecartClient(api_key=os.environ["DECART_API_KEY"])

  @app.route("/api/realtime-token", methods=["POST"])
  def create_realtime_token():
      """Generate a client token for client-side realtime connections."""
      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
  ```
</CodeGroup>

<Tip>
  Add authentication to your endpoint to ensure only authorized users can request client tokens. The examples above are simplified for clarity.
</Tip>

## Connecting

### Getting Camera Access

Request access to the user's camera using your platform's camera API:

```python theme={null}
stream = await get_camera_stream(
    audio=True,
    video={
        "frame_rate": 25,
        "width": 1280,
        "height": 704,
    },
)
```

<Tip>Use the model's `fps`, `width`, and `height` properties to ensure optimal performance.</Tip>

### Establishing Connection

Connect to the Realtime API with your media stream:

```python theme={null}
from decart.realtime import RealtimeClient, RealtimeConnectOptions
from decart.types import ModelState, Prompt

with open("character.jpg", "rb") as f:
    character_image = f.read()

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("lucy-2"),
        on_remote_stream=lambda transformed_stream: (
            # Display the transformed video
            handle_stream(transformed_stream)
        ),
        initial_state=ModelState(
            prompt=Prompt(
                text="Substitute the character in the video with the person in the reference image.",
                enhance=True,  # Let Decart enhance the prompt (recommended)
            ),
            image=character_image,  # Optional reference image
        ),
    ),
)
```

**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` - Reference image (`bytes`, URL string, or file path string)

<Tip>
  Set `initial_state.image` and/or `initial_state.prompt` so the first frame is already transformed — otherwise viewers briefly see the raw camera feed.
</Tip>

## Managing Prompts

Change the transformation style dynamically without reconnecting:

```python theme={null}
# 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)

<Note>Prompt enhancement uses Decart's AI to expand simple prompts for better results. Disable it if you want full control over the exact prompt.</Note>

## Unified State Update

The `set()` method replaces the entire session state in a single atomic call. The new state contains only the fields you provide — any previously set values not included in the call are cleared.

```python theme={null}
from decart import SetInput

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

# Set image only (clears any previously set prompt)
await realtime_client.set(SetInput(image=image_bytes))

# Set both prompt and image 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

<Tip>`set()` replaces the entire state — fields you omit are cleared. Always include every field you want to keep. Use `set()` instead of separate `set_prompt()` and `set_image()` calls to avoid intermediate states.</Tip>

## Connection State

Monitor and react to connection state changes:

```python theme={null}
# 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)

<Info>
  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.
</Info>

## Error Handling

Handle errors with the error event:

```python theme={null}
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

<Tip>Use `isinstance()` checks with specific exception types for better error handling.</Tip>

## Session Management

Access the current session ID:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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()
```

<Note>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.</Note>

## Cleanup

Always disconnect when done to free up resources:

```python theme={null}
# 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)
```

<Warning>Failing to disconnect can leave WebRTC connections open and waste resources.</Warning>

## Complete Example

Here's a full application with all features:

```python theme={null}
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("lucy-restyle-2")
        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

<AccordionGroup>
  <Accordion title="Use model properties for video constraints">
    Always use the model's `fps`, `width`, and `height` properties when requesting camera access to ensure optimal performance and compatibility.

    ```python theme={null}
    model = models.realtime("lucy-restyle-2")
    stream = await get_camera_stream(
        video={
            "frame_rate": model.fps,
            "width": model.width,
            "height": model.height,
        },
    )
    ```
  </Accordion>

  <Accordion title="Enable prompt enhancement">
    For best results, keep `enhance=True` (default) to let Decart's AI enhance your prompts. Only disable it if you need exact prompt control.
  </Accordion>

  <Accordion title="Handle connection state changes">
    Always listen to `connection_change` events to update your UI and handle reconnection logic gracefully.
  </Accordion>

  <Accordion title="Clean up properly">
    Always call `disconnect()` and stop media streams when done to avoid memory leaks and unnecessary resource usage.
  </Accordion>
</AccordionGroup>

## 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
      **Returns:** `RealtimeClient` - Connected realtime client instance

**Example:**

```python theme={null}
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("lucy-restyle-2"),
        on_remote_stream=handle_stream,
        initial_state=ModelState(
            prompt=Prompt(text="Anime", enhance=True),
        ),
    ),
)
```

### `await realtime_client.set(input)`

Replaces the entire session state atomically. Fields not included in the input are cleared (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`

```python theme={null}
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

<CardGroup cols={2}>
  <Card title="Process API" icon="wand-magic-sparkles" href="/sdks/python-process">
    Learn how to generate and transform media with the Process API
  </Card>

  <Card title="Examples" icon="code" href="/examples/real-time-mobile-app">
    See complete example applications
  </Card>
</CardGroup>