Robot app | LiveKit Documentation

Overview

Teleoperation is not limited to any particular language, but this guide focuses on native and embedded robot apps built with C++ , Rust , or Python .

Building a robot teleoperation app typically involves the following steps:

Publish camera views.
Receive control commands.
Apply commands safely.

Publish camera views

Publish each camera view from the robot as a named video track. For example, a mobile robot might publish front_camera and rear_camera tracks. The operator subscribes to the views it needs and renders the frames in the control interface.

The client SDK accepts frames from your app, encodes them, and publishes them to the room. LiveKit then forwards each track to subscribed operators. Keeping camera views in separate tracks lets the operator select views independently and lets LiveKit manage each stream according to the subscriber's needs.

Video latency affects how quickly an operator can react. Use hardware-accelerated encoding where available, and test your app with the low-latency settings that match its network conditions and tolerance for jitter.

The following examples publish a front_camera video track and push RGBA frames captured by app code. Replace the camera calls with your camera driver, GStreamer pipeline, or another local capture source:

#include "livekit/local_video_track.h"
#include "livekit/video_source.h"

constexpr int kWidth = 1280;
constexpr int kHeight = 720;

auto source = std::make_shared<livekit::VideoSource>(kWidth, kHeight);
auto track = livekit::LocalVideoTrack::createLocalVideoTrack("front_camera", source);

livekit::TrackPublishOptions options;
options.source = livekit::TrackSource::SOURCE_CAMERA;

if (auto local_participant = room->localParticipant().lock()) {
  local_participant->publishTrack(track, options);
}

while (camera.isRunning()) {
  auto frame = livekit::VideoFrame::create(kWidth, kHeight, livekit::VideoBufferType::RGBA);
  camera.readRgba(frame.data(), kWidth, kHeight); // Application-specific camera frame capture call below
  source->captureFrame(frame);
}

use livekit::prelude::*;

let source = NativeVideoSource::new(
    VideoResolution {
        width: 1280,
        height: 720,
    },
    false,
);
let track =
    LocalVideoTrack::create_video_track("front_camera", RtcVideoSource::Native(source.clone()));

let options = TrackPublishOptions {
    source: TrackSource::Camera,
    video_encoding: VideoEncoding {
        max_bitrate: 3_000_000,
        max_framerate: 30.0,
    }
    .into(),
    ..Default::default()
};

room.local_participant()
    .publish_track(LocalTrack::Video(track), options)
    .await?;

while let Some(frame) = camera.next_rgba_frame().await { // Application-specific camera frame capture
    source.capture_frame(&frame);
}

from livekit import rtc

WIDTH = 1280
HEIGHT = 720

source = rtc.VideoSource(WIDTH, HEIGHT)
track = rtc.LocalVideoTrack.create_video_track("front_camera", source)
options = rtc.TrackPublishOptions(
    source=rtc.TrackSource.SOURCE_CAMERA,
    simulcast=True,
    video_encoding=rtc.VideoEncoding(
        max_framerate=30,
        max_bitrate=3_000_000,
    ),
)

publication = await room.local_participant.publish_track(track, options)

async for frame_bytes, capture_time_us in camera.rgba_frames():
    frame = rtc.VideoFrame(WIDTH, HEIGHT, rtc.VideoBufferType.RGBA, frame_bytes)
    source.capture_frame(frame, timestamp_us=capture_time_us)

Publish another named track for each additional camera view. If the robot also streams telemetry, lidar, or diagnostics, use data tracks for high-frequency structured data and RPC for discrete operations that need a response, such as operator control leasing or authentication.

Receive control commands

Subscribe to the operator's control data track, deserialize each frame against the control schema, and translate the resulting command into the local control system. Because data tracks use lossy delivery, the robot should apply the latest valid command and tolerate missing frames.

Register data track handlers before connecting to the room. Otherwise, the client can miss already-published control tracks if events such as DataTrackPublished fire during the connection handshake.

The following examples subscribe to a robot.control data track from the expected operator identity. They decode each frame as UTF-8 JSON and pass decoded commands to your app code:

const std::string kOperatorIdentity = "operator";
const std::string kControlTrackName = "robot.control";

const auto callback_id = room->addOnDataFrameCallback(
    kOperatorIdentity,
    kControlTrackName,
    [&](const std::vector<std::uint8_t>& payload,
        std::optional<std::uint64_t> /*user_timestamp*/) {
      auto command = decodeControlFrame(payload);
      if (!command) {
        return;
      }

      // Application-specific command handling below
      controller.applyLatest(*command);
    });

use futures_util::StreamExt;
use livekit::prelude::*;

const OPERATOR_IDENTITY: &str = "operator";
const CONTROL_TRACK_NAME: &str = "robot.control";

while let Some(event) = room_events.recv().await {
    if let RoomEvent::DataTrackPublished(track) = event {
        if track.publisher_identity() != OPERATOR_IDENTITY
            || track.info().name() != CONTROL_TRACK_NAME
        {
            continue;
        }

        let controller = controller.clone();
        tokio::spawn(async move {
            let Ok(mut stream) = track
                .subscribe_with_options(
                    DataTrackSubscribeOptions::new().with_buffer_size(1),
                )
                .await
            else {
                return;
            };

            while let Some(frame) = stream.next().await {
                if let Ok(command) = decode_control_frame(frame.payload()) {
                    // Application-specific command handling below
                    controller.apply_latest(command).await;
                }
            }
        });
    }
}

import asyncio
import json

from livekit import rtc

OPERATOR_IDENTITY = "operator"
CONTROL_TRACK_NAME = "robot.control"


@room.on("data_track_published")
def on_data_track_published(track: rtc.RemoteDataTrack):
    if (
        track.publisher_identity != OPERATOR_IDENTITY
        or track.info.name != CONTROL_TRACK_NAME
    ):
        return

    asyncio.create_task(read_control_track(track))


async def read_control_track(track: rtc.RemoteDataTrack):
    stream = track.subscribe(buffer_size=1)

    async for frame in stream:
        try:
            payload = json.loads(frame.payload.decode("utf-8"))
            command = validate_control_frame(payload)
        except (json.JSONDecodeError, UnicodeDecodeError, ValueError):
            continue

        # Application-specific command handling below
        await controller.apply_latest(command)

Keep the data track buffer small for control input. A larger buffer can preserve more frames, but stale commands are usually worse than dropped commands for continuous teleoperation input.

Apply commands safely

Treat every command received from LiveKit as untrusted input before applying it to an actuator. These are general guidelines that don't all apply to every robot or control system. At a minimum, the robot app should do the following:

Authorize the operator: Accept commands only from an expected participant identity and issue tokens with the minimum required participant permissions.
Enforce one controller: Use an explicit control lease when more than one operator can join the room.
Validate every frame: Reject malformed, out-of-range, expired, or out-of-order commands.
Stop on lost input: Move to a safe state when the command deadline passes, the data track closes, or the controlling participant disconnects.
Clamp commands locally: Apply the robot's position, velocity, and acceleration limits on the robot, not only in the operator interface.

These app-level safeguards complement the authentication, encryption, and transport provided by LiveKit. They don't replace the hardware interlocks and safety systems required for your robot.

After validation, translate normalized control values into the local control system. Keep this translation deterministic and local to the robot:

def apply_drive_command(command: ControlCommand):
    steering = clamp(command.control_values["steering"], -1.0, 1.0)
    throttle = clamp(command.control_values["throttle"], -1.0, 1.0)

    left_velocity = clamp(throttle - steering, -1.0, 1.0) * MAX_WHEEL_VELOCITY
    right_velocity = clamp(throttle + steering, -1.0, 1.0) * MAX_WHEEL_VELOCITY

    motor_controller.set_velocity(left_velocity, right_velocity)

Drive the actuator only while valid commands are being received from the command app. Run it behind a watchdog so that if the latest valid command expires, the control track closes, or the controlling participant disconnects, the robot stops or moves to another safe state.

Additional resources

Use these implementations as references for robot teleoperation apps.

Rover teleop robot

Python robot participant for a Raspberry Pi rover that publishes camera video and applies remote drive commands.

Pan-tilt teleop robot

C++ robot participant that publishes RealSense video, receives velocity commands, and gates control with RPC.

C++ video example

Native example that creates local audio and video sources, publishes tracks, and captures frames.

C++ data track example

Native example that publishes and subscribes to named data tracks.