> ## 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 from Python

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:

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

```python theme={null}
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](/getting-started/client-tokens) instead of shipping your permanent API key.

### Backend Examples

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

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

<Tip>Authenticate the token endpoint before issuing — anyone who can call it can spend credits against your account.</Tip>

## 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`:

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

* [`realtime_synthetic.py`](https://github.com/DecartAI/decart-python/blob/main/examples/realtime_synthetic.py) — push generated frames at the model's `fps`.
* [`realtime_file.py`](https://github.com/DecartAI/decart-python/blob/main/examples/realtime_file.py) — read frames from a video file via OpenCV.

For a real camera, use any LiveKit-compatible capturer or feed frames yourself.

<Tip>Use the model's `fps`, `width`, and `height` for the source so capture matches what the server expects.</Tip>

### Establishing the connection

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

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

### Consuming the remote track

`on_remote_stream` hands you a LiveKit `RemoteVideoTrack`. Wrap it with `rtc.VideoStream` to receive decoded frames:

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

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

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

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

<Note>Prompt enhancement uses Decart's AI to expand simple prompts for better results. Disable it when you need exact prompt control.</Note>

## Unified State Update

`set()` replaces the entire session state in a single atomic call. Fields you omit are cleared.

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

<Tip>Prefer `set()` over separate `set_prompt()` and `set_image()` calls when you're changing both — it avoids intermediate states the model briefly renders.</Tip>

## Connection State

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

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

## Error Handling

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

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

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

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

<Note>Subscribers are receive-only — they cannot send prompts or images. They see exactly what the producer's `on_remote_stream` receives.</Note>

## Cleanup

```python theme={null}
await realtime.disconnect()

realtime.off("connection_change", on_connection_change)
realtime.off("error", on_error)
```

<Warning>Forgetting to `disconnect()` leaves the LiveKit room open and continues to consume credits.</Warning>

## Best Practices

<AccordionGroup>
  <Accordion title="Match the source to the model">
    Build `rtc.VideoSource(model.width, model.height)` and push frames at `model.fps`. Mismatched dimensions get rescaled server-side and waste bandwidth.
  </Accordion>

  <Accordion title="Seed the first frame">
    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.
  </Accordion>

  <Accordion title="Use `set()` for combined changes">
    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.
  </Accordion>

  <Accordion title="Always disconnect">
    `await realtime.disconnect()` from a `finally` block or async-context exit. The LiveKit room and HTTP session aren't reclaimed until you do.
  </Accordion>

  <Accordion title="Run on Python 3.10+">
    The SDK targets Python ≥ 3.10. The `decart[realtime]` extra adds `livekit` and `tenacity` on top of the core dependencies.
  </Accordion>
</AccordionGroup>

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

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

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