LiveKit Portal API reference

Overview

LiveKit Portal provides a small API surface that works with any robotics stack. It includes optional LeRobot integrations, but doesn't depend on a specific perception, control, or learning framework.

To learn more about Portal and this API, see Concepts.

The typical workflow is:

1. Create a configuration.
2. Construct a Robot or Operator.
3. Register callbacks.
4. Connect to a LiveKit room.
5. Send and receive observations and actions.

pip install livekit-portal      # or: uv add livekit-portal

Portal also provides a native Rust crate:

# Cargo.toml
[dependencies]
livekit-portal = { path = "path/to/portal/livekit-portal" }

The Python bindings ship as the livekit-portal-ffi crate plus a pure-Python package. For platform coverage and source builds, see the LiveKit Portal quickstart.

Declare the schema

The robot and operator must declare the same observation and action schema before connecting. The schema includes each field's name, order, and data type.

The following example declares two cameras and a four-field state and action schema, then sets the capture rate:

from livekit.portal import DType, RobotConfig

cfg = RobotConfig("session-1")

cfg.add_video("camera1")
cfg.add_video("wrist")

cfg.add_state_typed([
    ("joint1", DType.F32),
    ("joint2", DType.F32),
    ("gripper", DType.BOOL),
    ("mode", DType.I8),
])
cfg.add_action_typed([
    ("joint1", DType.F32),
    ("joint2", DType.F32),
    ("gripper", DType.BOOL),
    ("mode", DType.I8),
])

cfg.set_fps(30)

The string passed to RobotConfig is the session name. It is used only as a local label in logs and is independent of the LiveKit room name specified by the access token.

Choosing dtypes

Each field declares a data type that determines its representation on the wire.

Values passed to send_state() and send_action() use normal Python types. LiveKit Portal performs the required conversion when serializing data for transport.

Full config surface

Both RobotConfig and OperatorConfig expose the same configuration options. Options that apply only to one role are ignored by the other.

You can also build the whole config from a file with RobotConfig.from_yaml_file("portal.yaml", "session-1"), which keeps one shareable wire contract for both sides.

Robot

The robot declares its schema, applies incoming actions to the hardware, and publishes video and state on each tick:

import asyncio
import time

from livekit.portal import DType, Robot, RobotConfig

SCHEMA = [("joint1", DType.F32), ("joint2", DType.F32), ("gripper", DType.BOOL)]


async def main(url: str, token: str, hardware) -> None:
    cfg = RobotConfig("session-1")
    cfg.add_video("camera1")
    cfg.add_state_typed(SCHEMA)
    cfg.add_action_typed(SCHEMA)
    cfg.set_fps(30)

    robot = Robot(cfg)

    def on_action(action) -> None:
        # action.values is the typed dict.
        # action.timestamp_us is the operator's clock.
        # action.sender is the operator identity, stamped at the gate.
        # Only the active operator's actions reach this callback.
        hardware.apply(action.values)

    robot.on_action(on_action)
    await robot.connect(url, token)

    try:
        while hardware.running:
            reading = hardware.read()
            ts = int(time.time() * 1_000_000)
            robot.send_video_frame("camera1", reading.rgb, timestamp_us=ts)
            robot.send_state(reading.joints, timestamp_us=ts)
            await asyncio.sleep(1 / 30)
    finally:
        await robot.disconnect()
        robot.close()

Stamp the frame and the state with the same ts. The shared timestamp is what allows the operator to pair them.

If you omit timestamp_us, LiveKit Portal applies the current time at the moment of the call. This is correct when the two calls are adjacent, but incorrect if you capture, perform processing, and then send.

The send_video_frame method infers the width and height from the NumPy array. Specify them explicitly only when passing raw bytes.

Operator

The operator declares the same schema as the robot, runs its policy on each observation, and publishes the resulting actions:

import asyncio

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

SCHEMA = [("joint1", DType.F32), ("joint2", DType.F32), ("gripper", DType.BOOL)]


async def main(url: str, token: str, policy) -> None:
    cfg = OperatorConfig("session-1")
    cfg.add_video("camera1")
    cfg.add_state_typed(SCHEMA)
    cfg.add_action_typed(SCHEMA)
    cfg.set_fps(30)

    op = Operator(cfg)

    def on_observation(obs) -> None:
        frame = obs.frames["camera1"]
        rgb = frame_bytes_to_numpy_rgb(bytes(frame.data), frame.width, frame.height)
        action = policy(rgb, obs.state)
        op.send_action(action, in_reply_to_ts_us=obs.timestamp_us)

    op.on_observation(on_observation)
    await op.connect(url, token)

    # Without this the robot drops every action you send.
    await op.set_active_operator(op.local_identity())

    try:
        await asyncio.Event().wait()
    finally:
        await op.disconnect()
        op.close()

Passing in_reply_to_ts_us is optional. It allows metrics.policy.e2e_us_p50 to report true observation-to-action latency rather than a network round trip.

Lifecycle

A portal joins a room, leaves it, and releases its native handle:

await portal.connect(url, token)   # join the room
await portal.disconnect()          # leave it
portal.close()                     # drop the native handle eagerly

The close() method drops the reference to the native handle. It is optional: the handle is also released when the object is garbage-collected. Call it explicitly in a long-running service that creates portals repeatedly, so that a handle is not held until the next collection.

The connect method raises PortalError.AlreadyConnected if it is called twice.

Send data

Robot only:

robot.send_video_frame(track, frame, width=None, height=None, timestamp_us=None)
robot.send_state(values, timestamp_us=None)

Operator only:

op.send_action(values, timestamp_us=None, in_reply_to_ts_us=None)
op.send_action_chunk(name, data, timestamp_us=None, in_reply_to_ts_us=None)

The send_action_chunk method accepts either a dictionary of per-field columns of length horizon, or a single NumPy array of shape (horizon, n_fields) in declared field order. The array form matches what most VLA policies already emit. Columns of the wrong length are zero-padded.

Chunks travel as byte streams rather than data packets, so a full horizon is not bounded by the 15 KB packet limit.

Receive data

Every callback has a matching pull accessor. Use callbacks for push-driven loops, and use the accessors when your own loop controls the timing. Accessors return the latest value, so a slow reader sees the most recent value rather than a backlog.

Operator:

Robot:

The on_state and on_video_frame callbacks are raw, unmatched streams. They fire on arrival with no matching, which suits a preview pane or a debug log. Use on_observation when frames and state must correspond.

Typed values on receive

The Action, State, Observation, and ActionChunk objects carry typed values by default. The values attribute holds Python-native types according to your declared schema, and the raw_values attribute holds the lossless all-f64 representation. On an Observation, these attributes are named state and raw_state.

def on_action(action):
    action.values["gripper"]    # True, a bool
    action.values["mode"]       # 3, an int
    action.values["joint1"]     # 0.5, a float
    action.raw_values           # Dict[str, float], every field as f64

The Rust core mirrors this: Action, State, and Observation carry values: HashMap<String, TypedValue> alongside raw_values: HashMap<String, f64>.

The active operator

The robot accepts actions from one operator at a time, named by its active_operator pointer. For how the control gate behaves, see Set the active operator.

# Either role
portal.active_operator()                        # Optional[str]
await portal.set_active_operator("policy-v1")   # None clears it
portal.operators()                              # connected operator identities
portal.local_identity()                         # own identity, after connect

# Operator only
op.robot_identity()                             # the robot, once discovered

The set_active_operator method is symmetric. The robot writes its own attribute directly. An operator sends a portal.set_active_operator RPC, and the robot's handler performs the write. In both cases, the change propagates to all participants.

React to changes with three callbacks:

portal.on_operator_joined(lambda identity: ...)
portal.on_operator_left(lambda identity: ...)
portal.on_active_operator_changed(lambda identity: ...)   # identity can be None

Both roles need can_update_own_metadata=True in their token, because Robot and Operator self-set an lk.portal.role attribute on connect. Set the active operator when you generate the token so the robot accepts that operator's actions from the moment it connects:

api.AccessToken(key, secret).with_attributes(
    {"lk.portal.active_operator": "policy-v1"}
)

Video codecs

The add_video(name) method defaults to H.264 on the WebRTC media path. The VP8, VP9, AV1, and H265 codecs are also available; AV1 and H265 require both peers to negotiate them. The max_bitrate_kbps value is a ceiling, not a target.

from livekit.portal import VideoCodec

cfg.add_video("front", max_bitrate_kbps=8000)                  # H264, 8 Mbps cap
cfg.add_video("wide", codec=VideoCodec.VP9, max_bitrate_kbps=4000)

For lossless frames, pass a byte-stream codec. The user-facing API doesn't change: send_video_frame, on_video_frame, get_video_frame, and observations behave identically, and frames arrive as RGB regardless of the codec:

cfg.add_video("front", codec=VideoCodec.MJPEG, quality=90)
cfg.add_video("wrist", codec=VideoCodec.PNG)
cfg.add_video("debug", codec=VideoCodec.RAW)

For guidance on choosing a codec, see Video frame format. Codec details, latency, and per-track fps ceilings are in Frame video .

RPC

Either side can register methods, and either side can invoke them. Use RPC for one-off requests that do not belong in the control loop.

robot.register_rpc_method("home", lambda data: "ok")
robot.unregister_rpc_method("home")

reply = await op.perform_rpc("home", payload="{}")

Handlers must return a string. Full surface, error codes, and payload limits are in RPC .

Metrics

A connected portal exposes runtime metrics for synchronization, round-trip time, and policy latency:

m = portal.metrics()
m.sync.observations_emitted
m.sync.states_dropped
m.rtt.rtt_us_p95
m.policy.e2e_us_p95

portal.reset_metrics()

Every field is documented in Metrics .

Errors

You can catch these PortalError variants:

The ConfigFileError type is separate and originates only from the YAML loader.

Surface summary

A condensed list of every method available to each role:

# data
robot.send_video_frame(track, frame, width=None, height=None, timestamp_us=None)
robot.send_state(values, timestamp_us=None)
robot.on_action(cb)                          # active operator only
robot.on_action_chunk(name, cb)
robot.get_action() / robot.get_action_chunk(name)

# control plane
robot.active_operator() / await robot.set_active_operator(identity)
robot.operators() / robot.local_identity()
robot.on_operator_joined(cb) / robot.on_operator_left(cb)
robot.on_active_operator_changed(cb)

# rpc, metrics, lifecycle
robot.register_rpc_method(name, handler) / robot.unregister_rpc_method(name)
await robot.perform_rpc(method, payload, destination=None, response_timeout_ms=None)
robot.metrics() / robot.reset_metrics()
await robot.connect(url, token) / await robot.disconnect() / robot.close()

# data
op.send_action(values, timestamp_us=None, in_reply_to_ts_us=None)
op.send_action_chunk(name, data, timestamp_us=None, in_reply_to_ts_us=None)
op.on_observation(cb) / op.on_state(cb) / op.on_drop(cb)
op.on_video_frame(track, cb)
op.get_observation() / op.get_state() / op.get_video_frame(track)
op.on_action(cb) / op.get_action()           # requires action subscription

# control plane
op.active_operator() / await op.set_active_operator(identity)
op.operators() / op.robot_identity() / op.local_identity()
op.on_operator_joined(cb) / op.on_operator_left(cb)
op.on_active_operator_changed(cb)

# rpc, metrics, lifecycle
op.register_rpc_method(name, handler) / op.unregister_rpc_method(name)
await op.perform_rpc(method, payload, destination=None, response_timeout_ms=None)
op.metrics() / op.reset_metrics()
await op.connect(url, token) / await op.disconnect() / op.close()

Using LiveKit Portal directly

The Robot and Operator classes are role-specific wrappers over a unified Portal class, which is also exported:

from livekit.portal import Portal, PortalConfig, Role

cfg = PortalConfig("session-1", Role.ROBOT)
portal = Portal(cfg)

The Portal class provides the same behavior as the role-specific wrappers. The gate, the role attribute, and the built-in RPC handler are all present, with no opt-in flag. The only difference is that the type system exposes every method regardless of role, so a method that raises WrongRole at runtime can still be called.

LiveKit recommends using Robot or Operator in new code. Use Portal when the role is genuinely dynamic.

Next steps

Continue with runnable examples and the full Portal documentation: