LiveKit Portal quickstart | LiveKit Documentation

Overview

This guide builds a minimal LiveKit Portal app from scratch: two Python scripts that connect through a running LiveKit server and exchange a full observation-and-action loop. robot.py publishes video and joint state, and operator_app.py receives them as synchronized observations and sends actions back. It takes about five minutes and runs entirely on your own machine.

You don't need a physical robot. The robot script publishes a synthetic test pattern.

Start from a working example

To run a finished version instead of building it yourself, clone the basic example from the Portal repository. This guide builds the same two-file setup step by step.

Requirements

You need the following to run the quickstart:

Python 3.12: Prebuilt wheels target 3.12. The library supports 3.10 and later, but older versions need a source build.
A package manager: This guide uses pip, with uv shown as an alternative.
A LiveKit server: LiveKit Cloud or a local development server.
Your LiveKit project credentials: LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET.

Install LiveKit Portal

pip install livekit-portal livekit-api numpy

Or with uv :

uv add livekit-portal livekit-api numpy

Each package plays a distinct role: livekit-portal is the library, livekit-api creates access tokens (normally a server-side task), and numpy provides the array type for the frames.

Prebuilt wheels cover CPython 3.12 on Linux x86_64 and aarch64 (glibc 2.35 or later) and macOS Apple Silicon. Any other platform needs a source build.

Set your credentials

export LIVEKIT_URL="wss://your-project.livekit.cloud"
export LIVEKIT_API_KEY="APIxxxxxxxx"
export LIVEKIT_API_SECRET="xxxxxxxxxxxx"

Create tokens

Both scripts need a JWT for the same LiveKit room. Save this as portal_token.py.

Avoid shadowing standard library modules

Do not name these files token.py or operator.py. A module in your working directory takes precedence over the standard library, and shadowing either of those breaks imports across the interpreter.

# portal_token.py
import datetime
import os

from livekit import api
from livekit.protocol.room import RoomConfiguration

ROOM = "portal-quickstart"


def create_token(identity: str) -> str:
    grants = api.VideoGrants(
        room_join=True,
        room=ROOM,
        can_publish=True,
        can_subscribe=True,
        # Required. Robot and Operator each set an lk.portal.role attribute
        # on connect. Without this grant, connect fails.
        can_update_own_metadata=True,
    )
    return (
        api.AccessToken(os.environ["LIVEKIT_API_KEY"], os.environ["LIVEKIT_API_SECRET"])
        .with_identity(identity)
        .with_grants(grants)
        # Low playout delay bounds reduce teleoperation latency.
        .with_room_config(
            RoomConfiguration(name=ROOM, min_playout_delay=0, max_playout_delay=10)
        )
        .with_ttl(datetime.timedelta(hours=6))
        .to_jwt()
    )

Identities must be unique inside a room. There is one robot per session, so "robot" works. Operators choose their own name, such as "policy-v1" or "human-teleop".

Create tokens on a server in production

Creating tokens with your API secret belongs on a server, not in a robot or a browser. It is inline here to keep the quickstart to two files. See Authentication for production token generation.

Write the robot

This runs next to the hardware. It declares what it publishes (one camera and five state fields) and what it accepts (the same five as actions), then sends frames and state at 30 fps.

Save it as robot.py.

# robot.py
import asyncio
import math
import os
import time

import numpy as np
from livekit.portal import DType, Robot, RobotConfig

from portal_token import ROOM, create_token

FPS = 30
WIDTH, HEIGHT = 320, 240

# Both sides must declare the same fields, in the same order, with the
# same dtypes. Mixed dtypes are supported: floats for joints, a bool for
# the gripper, a small int for the control mode.
SCHEMA = [
    ("j1", DType.F32),
    ("j2", DType.F32),
    ("j3", DType.F32),
    ("gripper", DType.BOOL),
    ("mode", DType.I8),
]


def make_frame(phase: float) -> np.ndarray:
    """A moving test pattern. Returns (H, W, 3) uint8 RGB."""
    x = np.arange(WIDTH, dtype=np.float32) / WIDTH
    y = np.arange(HEIGHT, dtype=np.float32)[:, None] / HEIGHT
    r = np.broadcast_to((0.5 + 0.5 * np.sin(2 * math.pi * (x + phase))) * 255, (HEIGHT, WIDTH))
    g = np.broadcast_to((0.5 + 0.5 * np.sin(2 * math.pi * (y + phase))) * 255, (HEIGHT, WIDTH))
    b = np.full((HEIGHT, WIDTH), 128, dtype=np.float32)
    return np.stack([r, g, b], axis=-1).astype(np.uint8)


async def main() -> None:
    cfg = RobotConfig(ROOM)
    cfg.add_video("cam1")
    cfg.add_state_typed(SCHEMA)
    cfg.add_action_typed(SCHEMA)
    cfg.set_fps(FPS)

    robot = Robot(cfg)

    # Actions arrive from the operator that currently holds control.
    # Actions from other operators are dropped before this callback runs.
    def on_action(action) -> None:
        print(f"[robot] action from {action.sender}: {action.values}")

    robot.on_action(on_action)

    # A single request. Either side can register or invoke it.
    robot.register_rpc_method("home", lambda data: "homed")

    await robot.connect(os.environ["LIVEKIT_URL"], create_token("robot"))
    print("[robot] connected")

    try:
        for i in range(FPS * 60):
            phase = i / FPS
            # Use one clock for both the frame and the state so the operator
            # can match them.
            ts = int(time.time() * 1_000_000)
            robot.send_video_frame("cam1", make_frame(phase), timestamp_us=ts)
            robot.send_state(
                {
                    "j1": math.sin(phase),
                    "j2": math.cos(phase),
                    "j3": 0.1 * phase,
                    "gripper": int(phase) % 2 == 0,
                    "mode": int(phase) % 3,
                },
                timestamp_us=ts,
            )
            await asyncio.sleep(1 / FPS)
    finally:
        await robot.disconnect()
        robot.close()


if __name__ == "__main__":
    asyncio.run(main())

Frames must be uint8 NumPy arrays of shape (H, W, 3) in RGB order, and both dimensions must be even. See Concepts.

Write the operator

This runs alongside your policy or teleoperation UI. It declares the same schema, consumes synchronized observations, and publishes actions.

Save it as operator_app.py.

# operator_app.py
import asyncio
import os

from livekit.portal import DType, Operator, OperatorConfig, frame_bytes_to_numpy_rgb

from portal_token import ROOM, create_token

FPS = 30

# Identical to the robot's schema. Same fields, same order, same dtypes.
SCHEMA = [
    ("j1", DType.F32),
    ("j2", DType.F32),
    ("j3", DType.F32),
    ("gripper", DType.BOOL),
    ("mode", DType.I8),
]


async def main() -> None:
    cfg = OperatorConfig(ROOM)
    cfg.add_video("cam1")
    cfg.add_state_typed(SCHEMA)
    cfg.add_action_typed(SCHEMA)
    cfg.set_fps(FPS)

    op = Operator(cfg)
    seen = 0

    def on_observation(obs) -> None:
        nonlocal seen
        seen += 1

        # obs.frames["cam1"] is a VideoFrameData holding packed RGB24 bytes.
        frame = obs.frames["cam1"]
        rgb = frame_bytes_to_numpy_rgb(bytes(frame.data), frame.width, frame.height)

        if seen % FPS == 0:
            print(f"[operator] obs #{seen} frame={rgb.shape} state={obs.state}")

        # Your policy goes here. This example returns the state unchanged.
        action = dict(obs.state)

        # in_reply_to_ts_us identifies the observation this action answers,
        # which makes metrics.policy.e2e_us_* a true latency measurement
        # rather than a network round trip.
        op.send_action(action, in_reply_to_ts_us=obs.timestamp_us)

    op.on_observation(on_observation)

    await op.connect(os.environ["LIVEKIT_URL"], create_token("policy-v1"))
    print("[operator] connected")

    # The robot starts with no active operator and drops every action.
    # Claim control so this operator's actions are accepted.
    await op.set_active_operator(op.local_identity())

    print("[operator] home ->", await op.perform_rpc("home"))

    try:
        await asyncio.sleep(60)
    finally:
        await op.disconnect()
        op.close()


if __name__ == "__main__":
    asyncio.run(main())

Run both sides

Use two terminals in the same directory.

python robot.py           # terminal 1

python operator_app.py    # terminal 2

The operator prints an observation about once a second and the robot prints the actions coming back:

[operator] connected
[operator] home -> homed
[operator] obs #30 frame=(240, 320, 3) state={'j1': 0.84, 'j2': 0.54, 'j3': 0.1, 'gripper': False, 'mode': 1}

If you see this output, your credentials, the native library, and synchronization are all working.

A [state-overflow] and a [sync-drop] warning in the first second are expected. State begins flowing before the video track is ready, so the earliest states have no frames to match against. The warnings stop once the video track is running.

How it works

The robot stamps every frame and every state packet with a single clock. The operator buffers both streams, matches them by that timestamp, and invokes on_observation once per matched pair. Actions travel back on a separate reliable channel, gated so that only the active operator's actions arrive.

This gate is why the operator calls set_active_operator. Without it, the robot drops all actions, which is the most common first-run mistake. See Concepts.

Build from source

Build from source when no prebuilt wheel exists for your platform (Windows, Intel macOS, Python 3.10 or 3.11) or when you are modifying the Rust core. You need a Rust toolchain and uv .

git clone https://github.com/livekit/portal.git
cd portal

bash scripts/build_ffi_python.sh release
cd python && uv sync

The build_ffi_python.sh script runs cargo build -p livekit-portal-ffi, places the platform cdylib next to the Python package, and generates the UniFFI bindings. The first build takes a few minutes and later builds are incremental. Rerun it whenever the Rust code changes.

To depend on that build from another project, install it by path:

uv add --editable /abs/path/to/portal/python/packages/livekit-portal
# or
pip install -e /abs/path/to/portal/python/packages/livekit-portal

If the cdylib is located elsewhere, set LIVEKIT_PORTAL_FFI_LIB to its path.

LeRobot plugins

Two optional plugin packages wrap the code shown above. You pass in your existing LeRobot Robot or Teleoperator, and the remote arm appears as a local LeRobot device.

pip install lerobot-teleoperator-livekit   # robot side
pip install lerobot-robot-livekit          # operator side

The package names are intentionally inverted. The robot host is missing a source of actions, which LeRobot models as a Teleoperator, while the operator host is missing the robot itself. These plugins require Python 3.12 or later, because LeRobot does.

The plugins are a convenience layer over the API on this page, not a replacement for it. Read Concepts first, then the LeRobot plugin reference .

Next steps

The following resources help you build on the Portal quickstart.

Concepts

Roles, the observation model, control handoff, and the frame format.

LiveKit Portal API reference

Config, the Robot and Operator classes, callbacks, send methods, and the control plane.

Examples

Runnable projects, from synthetic video to physical SO-101 arms and cloud inference on Modal.

Full LiveKit Portal documentation

Tuning, frame video codecs, RPC, metrics, troubleshooting, and the wire protocol.