Robot Integration

Integrating physical robots with cloud-hosted Vision-Language-Action (VLA) models requires reliable control rates, deterministic safety bounds, and low-latency network streaming. Servo connects model inputs and outputs directly to whatever your robot provides—Python callables, hardware camera handles, or ROS 2 topics—without requiring custom framework wrappers or brittle configuration files.

Preview. The servo.connect API is currently in developer preview and not yet shipped in general availability. Early feedback and testing are welcome. Code samples on this page import connect/camera/inspect/gym_step explicitly from servo.universal — they don’t yet resolve as servo.connect etc.

The Connection Pattern

Every VLA policy operates on a simple principle: it consumes observations (camera feeds, joint states) and outputs actions (target joint positions or velocities). servo.connect() binds these inputs and outputs in a few lines of Python:

from servo.universal import camera, connect
from my_robot import BimanualYam

# 1. Initialize your robot controller or motor driver
arm = BimanualYam()

# 2. Connect model inputs (observe) and outputs (act)
session = connect(
    "allenai/MolmoAct2-BimanualYAM",
    region="us-west-2",
    observe={
        "top": camera(serial="250423122040"),
        "left": camera(serial="402323071792"),
        "right": camera(serial="402323071794"),
        "state": arm.get_joint_positions,
    },
    act=arm.apply_joint_targets,
)

# 3. Run a bounded autonomous episode
report = session.run(task="pick up the red cup", timeout_s=20.0)
print(f"Status: {report.status}, Steps: {report.steps}, Duration: {report.duration_s:.2f}s")
if report.status == "fault":
    print(f"Hardware fault: {report.error}")

Servo manages the background daemon (servo daemon), camera frame capture, hardware video encoding, and network transport automatically.

Discovering Model Contracts

Different VLA models expect different inputs and outputs. The exact contract is resolved for a hosted checkpoint and registered robot together: a model family can support several embodiments with different sensors. Servo’s built-in local development profile for allenai/MolmoAct2-BimanualYAM uses the top, left, right, and state observation slots. Before writing integration code, inspect the server-resolved contract in Python:

1. Python inspection (`servo.inspect`)

The contract is a property of the (model, robot) pairing, not the model alone, so robot is required — pass a registered robot name (same resolution as connect(robot=...)):

from servo.universal import inspect

contract = inspect("allenai/MolmoAct2-BimanualYAM", "yam-cell-01")
print(contract.inputs.keys())
# dict_keys(['top', 'left', 'right'])
top = contract.inputs["top"]
print(top.type, top.shape, top.dtype)
# image (378, 378, 3) uint8

This current flat inspection projection reports model-prepared observation inputs. It does not yet preserve the full inference contract or expose action outputs; use robot.check(policy) to validate the complete live path.

2. Immediate preflight error detection

For the exact built-in MolmoAct2 development profile, servo.connect() checks the observed slot names immediately at startup. A missing slot or typo (such as "front" instead of "top") fails before a control loop starts:

ContractMismatchError: Model 'allenai/MolmoAct2-BimanualYAM' observation mismatch.
  Missing required slots: ['top']
  Unexpected extra slots: ['front']
  Did you mean 'top' instead of 'front'?

3. Static type checking & IDE autocomplete

The Servo SDK provides TypedDict definitions so your IDE (VS Code, PyCharm) can autocomplete slot names and flag errors with static type checkers like mypy or pyright:

from servo.models import MolmoAct2BimanualYAM, MolmoAct2Observation
from servo.universal import camera, connect

observe: MolmoAct2Observation = {
    "top": camera(serial="250423122040"),
    "left": camera(serial="402323071792"),
    "right": camera(serial="402323071794"),
    "state": arm.get_joint_positions,
}

session = connect(
    MolmoAct2BimanualYAM,
    observe=observe,
    act=arm.apply_joint_targets,
)

The explicit annotation is what enables slot-name checking and IDE completion; connect() does not infer a model-specific TypedDict from a string literal.

Supplying Observations (`observe`)

The observe argument accepts several flexible formats depending on your hardware and software stack:

Option A: Dedicated hardware camera handles

For USB and RealSense cameras, use servo.camera(serial="..."). Servo opens cameras using zero-copy DMA-BUF streaming directly into hardware video encoders:

from servo.universal import camera

observe = {
    "top": camera(serial="250423122040"),
    "left": camera(serial="402323071792"),
    "right": camera(serial="402323071794"),
    "state": arm.get_joint_positions,
}

The customer CLI does not expose automatic camera discovery yet. Generic V4L2 cameras can be opened by a verified device index. For a supported Intel RealSense installation, pass a known vendor serial and verify each role before starting a session.

Option B: Python callable functions

Pass any non-blocking callable returning a NumPy array or PyTorch tensor:

observe = {
    "top": overhead_cam.get_rgb_frame,  # returns (360, 640, 3) uint8
    "left": wrist_cam_l.get_rgb_frame,
    "right": wrist_cam_r.get_rgb_frame,
    "state": robot.get_joint_angles,    # returns (14,) float32 in radians
}

Images: (H, W, 3) uint8 RGB NumPy arrays. (If OpenCV outputs BGR, pass servo.camera(..., color_space="bgr") or convert with cv2.cvtColor).
Joint States: 1D vector (N,) float32 in radians for revolute joints, meters for prismatic joints.

Option C: Unified observation method (Simulators & existing environments)

If you already have a custom driver or simulation environment with an atomic observation function, pass that function directly:

from servo.universal import connect

session = connect(
    "allenai/MolmoAct2-BimanualYAM",
    observe=env.get_observation,  # returns {"top": ..., "left": ..., "right": ..., "state": ...}
    act=env.step,
)

Option D: ROS 2 topic strings

If your robot runs on ROS 2, pass your active node and topic names directly:

from servo.universal import connect

session = connect(
    "allenai/MolmoAct2-BimanualYAM",
    ros_node=node,
    observe={
        "top": "/camera/front/image_raw",
        "left": "/camera/left_wrist/image_raw",
        "right": "/camera/right_wrist/image_raw",
        "state": ["/yam_left/joint_states", "/yam_right/joint_states"],
    },
    act={"left": "/yam_left/joint_command", "right": "/yam_right/joint_command"},
)

Servo automatically subscribes to standard ROS message types (sensor_msgs/Image, sensor_msgs/JointState) and publishes commands.

Commanding Motor Actuators (`act`)

On each control cycle (e.g. 30 Hz), Servo passes the target action array (e.g. (14,) float32) to your act callable:

def apply_action(action: np.ndarray) -> None:
    # Send continuous joint targets to motor controller
    controller.command_positions(action)

Return semantics

Return Value	Meaning	Rollout Behavior
`None`	Dispatched: action sent to motors successfully.	Episode continues at 30 Hz.
`{"done": True}`	Early completion: task achieved early (e.g. sensor detected grip).	Episode ends cleanly ahead of `timeout_s`.
Exception raised	Hardware fault: motor driver encountered an error.	Rollout halts immediately; safety hold engaged.

Wrapping existing Gym environments

If wrapping a standard Gym env.step method, use servo.gym_step to cleanly handle step tuples:

from servo.universal import gym_step

act = gym_step(env.step)

Built-in Safety: Zero-velocity hold on fault

When controlling physical hardware, software errors must never cause runaway motor movements. If your act function raises an exception (such as motor overcurrent, joint limit reached, bus timeout, or E-stop trigger):

Active Zero-Velocity Hold: Servo catches the exception immediately and commands actuators to lock position or hold zero velocity.
Telemetry Preservation: All video frames, network round-trip traces, and joint timestamps leading up to the exact moment of failure are flushed and preserved.
Structured Reporting: session.run() exits safely and returns a RunReport with report.status = "fault" and report.error = str(exception).

Running Episodes & Controlling Motion

Executing an episode (`session.run`)

session.run() executes a bounded autonomous rollout and returns a RunReport:

report = session.run(
    task="pick up the red cup",
    timeout_s=30.0,   # Maximum duration in wall-clock seconds
    max_steps=300,    # Maximum control steps (300 steps at 30 Hz = 10s)
)

print(report.status)       # "completed", "timeout", "fault", "intervened", "halted"
print(report.steps)        # Total control steps executed
print(report.duration_s)   # Total elapsed time in seconds

# Optionally raise an exception if the run failed:
report.raise_for_status()

Current limitations: until=/on_step= are only implemented for mock=True sessions today — passing either to a real (non-mock) run raises rather than silently no-op’ing, since the real chunked control loop has no per-step hook or early-stop predicate yet (only a coarser on_chunk= via robot.run()). report.dataset_path is also always None on a real run today: a bare connect(model, robot=...) binding has no deployment to record episodes against, unlike deployment.policy()/fleet.policy() — see ROBOT_INTEGRATION_SPEC.md’s “Known limitation” note.

Non-blocking camera preview

To display a live camera feed in a local OpenCV window or web viewer without slowing down the 30 Hz control loop:

# Reads latest frame directly from shared memory:
top_frame_rgb = session.get_frame("top")

Pausing and resuming motion

If an obstacle appears or a human enters the workcell:

# Halts action commands and locks actuators in place:
session.pause()

# Resumes autonomous policy execution:
session.resume()

Direct Functional Control Loop (`servo.Policy`)

If your system already manages its own custom while loop (for custom safety filters, low-level trajectory smoothing, or custom step logging), use servo.Policy:

import servo

policy = servo.Policy("allenai/MolmoAct2-BimanualYAM", robot="yam-cell-01")

for step in range(300):
    obs = env.get_obs()

    action = policy(
        top=obs["front_camera"],
        left=obs["left_camera"],
        right=obs["right_camera"],
        state=obs["joint_positions"],
        task="clean the tabletop",
    )

    env.step(action)

Each call to policy(...) returns the first row of the chunk the model returned. It does not buffer or prefetch — every call is one real network round trip — so call it no faster than your model’s round-trip allows, or add your own buffering for a tighter loop.

Remote Operations (`session.listen`)

In automated workcells and production lines, the robot script often runs continuously on the cell IPC while tasks are triggered remotely by a central scheduler, web console, or CI script:

# 1. On the robot: bind hardware once and listen for tasks
from servo.universal import connect

session = connect(
    "allenai/MolmoAct2-BimanualYAM",
    robot="yam-cell-01",
    region="us-west-2",
    observe={...},
    act=...,
)
session.listen()  # Blocks and reports ready state to the control plane

# 2. From an ops console, scheduler, or CI runner:
import os
import servo

sv = servo.Servo(
    base_url=os.environ["SERVO_BASE_URL"],
    api_key=os.environ["SERVO_API_KEY"],
)

robot = sv.robots.get("yam-cell-01")
run = robot.start(task="pick up the red cup", timeout_s=30.0)

# Optional controls during execution:
run.pause()
run.resume()

# Wait for completion:
report = run.wait()
print(f"Task finished with status: {report.status}")

Zero-Motion Preflight (`servo check-rig`)

Before powering on robot motors, run zero-motion preflight against your integration script:

servo check-rig rig.py           # against connected hardware
servo check-rig rig.py --mock    # synthetic frames and joints (no robot required)

[Servo Diagnostic Preflight: Bimanual YAM]
✔ Local Sidecar: Active (/run/servo/agent.sock)
✔ Network Transport: Connected to us-west-2 (RTT: 17.8 ms)
✔ Sensor Sources:
    - 'top':   Stream 0 (640x360 RGB) -> Hardware encoder ready
    - 'left':  Stream 1 (640x360 RGB) -> Hardware encoder ready
    - 'right': Stream 2 (640x360 RGB) -> Hardware encoder ready
    - 'state': 14-D vector verified (values in plausible range)
✔ Actuator Sink: Registered & Responsive
✔ Remote Cockpit: Ready at https://servo.run/cockpit/bimanual-yam-01
STATUS: READY FOR LIVE EXECUTION.

How Servo Manages Streaming Under the Hood

Servo is designed to give you smooth, stutter-free 30 Hz control even when connecting to remote cloud GPUs over the internet:

Local Background Daemon (servo daemon): Decouples your robot control loop from network traffic and video encoding. Video compression happens on hardware silicon without consuming CPU cycles needed by your motor drivers.
Pipelined Action Prefetching: VLA models predict chunks of future actions (e.g. 30 steps ahead). While your robot executes step 15, Servo captures the next observation and prefetches the next action chunk from the cloud. When step 30 finishes, step 31 is already waiting—eliminating boundary pauses.
Smooth Teleoperation Blending: When transitioning between autonomous policy control and human gamepad teleoperation, Servo applies a 50 ms velocity blend to eliminate sudden motion jolts and protect robot gearboxes.

Get started

Use Servo

Robot Integration

The Connection Pattern

Discovering Model Contracts

1. Python inspection (`servo.inspect`)

2. Immediate preflight error detection

3. Static type checking & IDE autocomplete

Supplying Observations (`observe`)

Option A: Dedicated hardware camera handles

Option B: Python callable functions

Option C: Unified observation method (Simulators & existing environments)

Option D: ROS 2 topic strings

Commanding Motor Actuators (`act`)

Return semantics

Wrapping existing Gym environments

Built-in Safety: Zero-velocity hold on fault

Running Episodes & Controlling Motion

Executing an episode (`session.run`)

Non-blocking camera preview

Pausing and resuming motion

Direct Functional Control Loop (`servo.Policy`)

Remote Operations (`session.listen`)

Zero-Motion Preflight (`servo check-rig`)

How Servo Manages Streaming Under the Hood

​The Connection Pattern

​Discovering Model Contracts

​1. Python inspection (servo.inspect)

​2. Immediate preflight error detection

​3. Static type checking & IDE autocomplete

​Supplying Observations (observe)

​Option A: Dedicated hardware camera handles

​Option B: Python callable functions

​Option C: Unified observation method (Simulators & existing environments)

​Option D: ROS 2 topic strings

​Commanding Motor Actuators (act)

​Return semantics

​Wrapping existing Gym environments

​Built-in Safety: Zero-velocity hold on fault

​Running Episodes & Controlling Motion

​Executing an episode (session.run)

​Non-blocking camera preview

​Pausing and resuming motion

​Direct Functional Control Loop (servo.Policy)

​Remote Operations (session.listen)

​Zero-Motion Preflight (servo check-rig)

​How Servo Manages Streaming Under the Hood

The Connection Pattern

Discovering Model Contracts

1. Python inspection (`servo.inspect`)

2. Immediate preflight error detection

3. Static type checking & IDE autocomplete

Supplying Observations (`observe`)

Option A: Dedicated hardware camera handles

Option B: Python callable functions

Option C: Unified observation method (Simulators & existing environments)

Option D: ROS 2 topic strings

Commanding Motor Actuators (`act`)

Return semantics

Wrapping existing Gym environments

Built-in Safety: Zero-velocity hold on fault

Running Episodes & Controlling Motion

Executing an episode (`session.run`)

Non-blocking camera preview

Pausing and resuming motion

Direct Functional Control Loop (`servo.Policy`)

Remote Operations (`session.listen`)

Zero-Motion Preflight (`servo check-rig`)

How Servo Manages Streaming Under the Hood