> ## Documentation Index
> Fetch the complete documentation index at: https://servo.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Robot Integration

> Connect cameras, joint sensors, and motor controllers to hosted VLA models.

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:

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

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

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

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

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

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

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

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

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

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

1. **Active Zero-Velocity Hold**: Servo catches the exception immediately and commands actuators to lock position or hold zero velocity.
2. **Telemetry Preservation**: All video frames, network round-trip traces, and joint timestamps leading up to the exact moment of failure are flushed and preserved.
3. **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`:

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

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

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

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

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

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

```bash theme={null}
servo check-rig rig.py           # against connected hardware
servo check-rig rig.py --mock    # synthetic frames and joints (no robot required)
```

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

1. **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.
2. **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.
3. **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.