# Servo — full documentation for LLM agents

> Generated from the docs sources by `python -m repo_tools generate docs`. Per-page markdown lives under /docs/md/; the index is /llms.txt.

# Introduction

> Connect physical robots to hosted VLA models with low-latency streaming.

Servo connects physical robot hardware to cloud-hosted Vision-Language-Action (VLA) models with
low-latency streaming, smooth action execution, and built-in safety controls.

Instead of forcing your robot into rigid custom frameworks or writing fragile glue code,
Servo connects model inputs and outputs directly to your existing Python functions, camera streams,
and motor controllers.

> **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` explicitly from `servo.universal` — they don't yet resolve as `servo.connect` etc.

## Why Servo

1. **Zero Glue Code**: Model inputs (such as camera feeds and joint angles) and outputs (target joint actions) bind directly to your existing Python callables, ROS 2 topics, or hardware camera handles.
2. **Smooth 30 Hz Motion**: Pipelined action chunk prefetching streams actions continuously without chunk-boundary pauses or arm stutter.
3. **Built-in Safety**: If a motor driver raises an exception (overcurrent, joint limits exceeded, or E-stop), Servo catches the fault immediately, commands an active zero-velocity hold, and logs diagnostic telemetry.
4. **Environment Isolation**: Your robot drivers can run in their native environment while Servo's background engine handles hardware-accelerated video compression and low-latency network transport.
5. **Live Monitoring & Interventions**: Review real-time camera feeds in your browser and seamlessly transition between autonomous policy control and manual teleoperation.

## How it works

Connecting a robot takes just a few lines of Python:

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

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

# 2. Connect model inputs and outputs
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 an autonomous task
report = session.run(task="pick up the red cup", timeout_s=20.0)
print(report)
```

Servo automatically starts its local background daemon (`servo daemon`) on first use to manage video compression, shared memory buffers, and streaming transport. For dedicated production robot cells, you can install the daemon as a supervised system service with `servo daemon --install-service`.

Before moving physical hardware, you can also run zero-motion preflight verification using `servo check-rig` to validate all camera streams, network connections, and motor bindings safely.

## Next steps

- Follow the [Quickstart](/docs/md/quickstart.md) to test a simulated rollout on your laptop in 5 minutes.
- Read the [Robot Integration Guide](/docs/md/guides/robots.md) for complete details on cameras, motor contracts, and ROS 2.
- Learn how to manage multi-robot deployments in [Robot Fleets](/docs/md/guides/fleet-deployment.md).
- Explore all classes and methods in the [Python API Reference](/docs/md/reference/python-api.md).

---

# Quickstart

> Get started with Servo and connect your robot in 5 minutes.

Connect your robot to cloud-hosted Vision-Language-Action (VLA) models in minutes. Servo
connects model inputs and outputs directly to your existing Python functions, camera streams,
or motor drivers without requiring custom wrappers or 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.

## Prerequisites

Before starting, make sure you have:
- Python 3.10+ installed
- Network access to your Servo control-plane URL
- A Servo account or API key

---

## Step 1: Install and authenticate

Install the Servo client package:

```bash
pip install 'servo-client>=0.4,<0.5'
export SERVO_BASE_URL="https://<your-servo-control-plane>"
```

On a development machine with a browser, sign in interactively:

```bash
servo login
```

For headless robot computers (such as onboard IPCs accessed over SSH), generate an API key on your development machine with `servo key create --label robot-cell-01`, then export it on the robot:

```bash
export SERVO_API_KEY="sk_servo_..."
```

---

## Step 2: Test on your laptop (no hardware needed)

You do not need physical hardware on hand to start developing. Use `mock=True` to simulate inputs and verify the complete workflow on your development computer:

```python
from servo.universal import connect

# Connect to a hosted model with mock=True for instant local testing
session = connect(
    "allenai/MolmoAct2-BimanualYAM",
    region="us-west-2",
    mock=True,
)

# Run a test episode
report = session.run(task="pick up the red cup", timeout_s=5.0)
print(report)
```

```text
[Servo RunReport: COMPLETED]
Duration: 5.00s | Steps: 150 (30.0 Hz) | Status: completed
```

This simulates camera frames and joint telemetry, validates communication with the model endpoint, and returns a structured `RunReport`.

---

## Step 3: Connect your real robot (`rig.py`)

When you are ready to connect physical hardware, create `rig.py` to bind your real cameras and motor methods using `observe` and `act`:

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

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

# 2. Connect model inputs and outputs to your hardware
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 live episode
report = session.run(task="pick up the red cup", timeout_s=20.0)
print(report)
```

Servo automatically starts its local background daemon (`servo daemon`) on first use if it isn't already running. For production robot cells, run `servo daemon --install-service` once to install it as a supervised system service.

### Understanding `observe` and `act`

| Channel | Role | What to provide | Notes |
|---|---|---|---|
| `observe` | Sensory Inputs | Dictionary of camera handles (`camera()`, from `servo.universal`) and state callables | Keys match the model's required sensor slots (e.g. `top`, `left`, `right`, `state`). |
| `act` | Motor Outputs | Function accepting continuous target joint arrays (e.g. `(14,) float32`) | Called at 30 Hz. Return `None` on success, `{"done": True}` to complete early, or raise to trigger safety hold. |

---

## Step 4: Zero-motion preflight (`servo check-rig`)

Before powering on physical motors, run zero-motion preflight verification against your script:

```bash
servo check-rig rig.py
```

`servo check-rig` imports `rig.py` and validates all camera connections, network latency, and motor bindings up to the point of motion without commanding any motor movement:

```text
[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
✔ Actuator Sink: Registered & Responsive
✔ Remote Cockpit: Ready at https://servo.run/cockpit/bimanual-yam-01
STATUS: READY FOR LIVE EXECUTION.
```

To validate scripts before hardware arrives, run `servo check-rig rig.py --mock`.

---

## Step 5: Camera discovery

To find the serial numbers of USB or RealSense cameras attached to your computer, run:

```bash
servo camera list
```

```text
Found 3 cameras:
  - Intel RealSense Depth Camera 405 | Serial: 402323071794 | RGB: /dev/video4  | Res: 640x360
  - Intel RealSense Depth Camera 405 | Serial: 402323071792 | RGB: /dev/video10 | Res: 640x360
  - Intel RealSense Depth Camera 435 | Serial: 250423122040 | RGB: /dev/video16 | Res: 640x360
```

Pass the serial strings directly into `camera(serial="...")` (from `servo.universal`). Using hardware serial numbers ensures camera streams never swap if USB ports or device indices change after a reboot.

---

## Step 6: Live monitoring and interventions

While an episode runs, you can monitor and control execution interactively:

1. **Remote Web Cockpit**: Open the URL displayed by `servo check-rig` to view real-time camera streams and latency charts in your browser.
2. **Teleoperation Interventions**: Take manual control using a gamepad or keyboard from the cockpit at any time. Servo smoothly blends velocity trajectories to eliminate motion jolts.
3. **Programmatic Pause & Resume**:
```python
   # Pause autonomous actions (holds current position safely):
   session.pause()

   # Resume policy execution:
   session.resume()
```

---

## Direct control loops (`servo.Policy`)

If you prefer managing your own `while` loop (for custom safety filters, step-by-step logging, or custom simulation wrappers), use `servo.Policy`:

```python
import servo
from my_robot.env import RobotEnv

env = RobotEnv()
policy = servo.Policy("allenai/MolmoAct2-BimanualYAM", robot="bimanual-yam-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="pick up the red cup",
    )

    env.step(action)
```

`policy(...)` returns the first row of the chunk the model returned, directly usable by `env.step(action)`. 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.

---

## Next steps

- Read the in-depth [Robot Integration Guide](/docs/md/guides/robots.md) to learn about ROS 2 topics, custom sensor contracts, and fault handling.
- Scale up to multi-robot deployments in [Robot Fleets](/docs/md/guides/fleet-deployment.md).
- Review all methods and configuration options in the [Python API Reference](/docs/md/reference/python-api.md).

---

# 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
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. For example:
- `allenai/MolmoAct2-BimanualYAM` expects 3 camera streams (`top`, `left`, `right`), a 14-dimensional kinematic joint vector (`state`), and outputs a 14-dimensional joint target action at 30 Hz.
- `openpi/pi0_droid` expects 2 wrist/exterior cameras, a joint position vector, and outputs joint targets.

Before writing integration code, discover what any model requires using terminal inspection or Python:

### 1. Terminal inspection (`servo inspect`)

```bash
servo inspect allenai/MolmoAct2-BimanualYAM
```

```text
Model: allenai/MolmoAct2-BimanualYAM
Architecture: MolmoAct2 (30 Hz action chunking, horizon: 30)

Required Observation Slots (`observe`):
  - 'top':   Image (360, 640, 3) uint8 (overhead camera)
  - 'left':  Image (360, 640, 3) uint8 (left wrist camera)
  - 'right': Image (360, 640, 3) uint8 (right wrist camera)
  - 'state': Vector (14,) float32 in radians [left arm (0..5), left gripper, right arm (0..5), right gripper]

Required Actuator Sink (`act`):
  - Continuous joint target vector (14,) float32 in radians commanded at 30 Hz
```

### 2. 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
from servo.universal import inspect

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

### 3. Immediate preflight error detection

If your `observe` dictionary has a missing slot or a typo (such as `"front"` instead of `"top"`), `servo.connect()` detects it immediately at startup before moving physical hardware:

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

### 4. 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
from servo.models import MolmoAct2BimanualYAM
from servo.universal import camera, connect

session = connect(
    MolmoAct2BimanualYAM,
    observe={
        "top": camera(serial="250423122040"),
        "left": camera(serial="402323071792"),
        "right": camera(serial="402323071794"),
        "state": arm.get_joint_positions,
    },
    act=arm.apply_joint_targets,
)
```

---

## 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
from servo.universal import camera

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

To list plugged-in camera serial numbers, run `servo camera list`.

### Option B: Python callable functions

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

```python
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
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
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
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
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
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
# 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
# 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
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
# 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
# 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
servo check-rig rig.py           # against connected hardware
servo check-rig rig.py --mock    # synthetic frames and joints (no robot required)
```

```text
[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.

---

# Robot Fleets

> Share managed model capacity across multiple robots with automatic scaling.

A fleet pools hosted model capacity across a cluster of robots, automatically scaling GPU compute based on the maximum number of robots active simultaneously (`peak_active`).

Each robot runs `servo daemon` locally and connects its sensors and motor controllers directly to the shared fleet.

> **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` explicitly from `servo.universal` — they don't yet resolve as `servo.connect` etc.

---

## 1. Assign stable robot identities

To connect a robot to a fleet, assign it a stable name (and optional site/labels) in `servo.connect()`:

```python
from servo.universal import connect

session = connect(
    "allenai/MolmoAct2-BimanualYAM",
    robot="yam-cell-01",
    region="us-west-2",
    site="sf-lab",
    labels={"line": "assembly"},
    observe={...},
    act=...,
)
```

The first connection with a given name registers the robot in your organization; subsequent connections attach to that existing record. Registration derives a configuration digest from the robot's actual bound `observe`/`act` sources, so there's no CLI equivalent to pre-register a name before hardware and a driver exist — `connect(robot=...)` is the only registration path today.

List registered robots in your organization with:

```bash
servo robot list
```

---

## 2. Deploy shared fleet capacity

From your central management script or developer machine, provision shared GPU capacity for your fleet:

```python
import os
import servo

sv = servo.Servo(
    base_url=os.environ["SERVO_BASE_URL"],
    api_key=os.environ["SERVO_API_KEY"],
)
model = sv.models.get("pi0.5")

fleet = sv.fleets.deploy(
    model,
    robots=["yam-cell-01", "yam-cell-02"],
    name="yam-assembly",
    peak_active=2,
    readiness="always_ready",
)

print(f"Fleet deployed: {fleet.id}")
```

`peak_active` defines the maximum number of robots that run concurrently. Servo automatically manages GPU provisioning to guarantee 30 Hz control loop latency for all active sessions.

For larger deployments, you can dynamically select robots using site and label selectors:

```python
fleet = sv.fleets.deploy(
    model,
    selector={"site": "sf-lab", "labels": {"line": "assembly"}},
    name="yam-assembly",
    peak_active=40,
    readiness="scheduled",
    schedule=[
        {
            "starts_at": "2026-09-03T06:00:00-07:00",
            "ends_at": "2026-09-03T14:00:00-07:00",
        }
    ],
)
```

### Readiness modes

| Mode | Behavior |
| --- | --- |
| `always_ready` | Keeps GPU capacity warm in the cloud for zero-latency rollout starts. |
| `scheduled` | Automatically warms GPU instances before scheduled operating shifts. |
| `on_demand` | Provisions GPU capacity when robots initiate rollouts. |

---

## 3. Run each robot locally

On each robot computer, install the Servo daemon as a supervised system service (which provides automatic restarts on reboot or failure):

```bash
servo daemon --install-service
```

Run zero-motion preflight verification:

```bash
servo check-rig rig.py
```

In `rig.py`, point `servo.connect` to your fleet name instead of a single standalone model:

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

arm = BimanualYam()

session = connect(
    fleet="yam-assembly",
    robot="yam-cell-01",
    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,
)

# For automated cells, listen for remote start commands:
session.listen()
```

---

## 4. Roll out model updates without downtime

When a newly trained or fine-tuned checkpoint (`ckpt_...`) is ready, update the fleet without disrupting active robot operations:

```python
fleet.update("ckpt_20260904_pi05_dexterity")
print(fleet.update_status)
```

- **Zero-Downtime Cutover**: Active episodes finish uninterrupted on the current release. Once the new model checkpoint is warm and verified, subsequent rollouts automatically route to it.
- **Instant Rollback**: If an issue occurs with a new checkpoint, call `fleet.rollback()` at any time to immediately revert to the prior healthy version:
```python
  fleet.rollback()
```

---

# Authentication

> Sign in to the CLI and authenticate robot applications.

## CLI login

`servo login` opens a browser for interactive authentication on development machines, storing credentials
locally in `~/.config/servo/token.json`.

Set the control-plane URL supplied by your Servo administrator, then sign in:

```bash
export SERVO_BASE_URL="https://<your-servo-control-plane>"
servo login
servo whoami
```

Login state is stored on the current computer. Pass `--base-url` to select a different control
plane for one command.

For **headless robot computers** (such as remote IPCs accessed over SSH), do not run `servo login` on
the robot. Instead, create an API key on your development computer and export `SERVO_API_KEY` on the robot.

## Robot applications and automation

Python applications and noninteractive CLI commands use an organization API key:

```bash
servo key create --label yam-cell-01
export SERVO_API_KEY="sk_servo_..."
```

Servo displays the secret once. Store it in the robot computer's secret manager. The key
authorizes the application; the robot's stable name and `rob_*` ID identify the physical rig. A
robot computer with `SERVO_API_KEY` set can call `sv.robots.setup(...)` or run `servo robot list`
without `servo login`.

List and revoke keys with:

```bash
servo key list --active
servo key revoke <key-id>
```

## Python

```python
import os

import servo

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

Missing or invalid credentials return `401`. A valid credential without permission for an
operation returns `403`.

---

# Python API

> Robot integration, functional policy execution, and fleet administration.

The Servo Python SDK provides two primary layers:
1. **Robot Integration & Policy Execution**: Connect robot hardware directly to hosted VLA models with `servo.connect()`, `servo.Policy()`, `servo.camera()`, and `servo.inspect()`.
2. **Resource Management & Administration**: Manage fleet deployments, query available models, and register robots with `servo.Servo`.

The primary types are `servo.Policy`, `servo.Session`, `servo.Observation`, `servo.ActionPrediction`, `servo.Servo`, `servo.Robot`, `servo.Model`, `servo.HostedDeployment`, and `servo.Fleet`.

> **Preview.** The `servo.connect` API is currently in developer preview and not yet shipped in general availability. Early feedback and testing are welcome. `connect`/`camera`/`inspect`/`gym_step` are imported explicitly from `servo.universal` in the examples below — they don't yet resolve as `servo.connect` etc.

## Robot Integration API

### `servo.connect`

Binds named model input and output slots to Python callables, ROS 2 topics, or hardware handles:

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

arm = BimanualYam()

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,
)

session.run(task="pick up the red cup")
```

#### Parameters:
`connect()` runs in one of two mutually exclusive modes — pass `model` for a dedicated session, or `fleet` + `robot` to draw from pooled fleet capacity (see [Robot Fleets](/docs/md/guides/fleet-deployment.md)):
- `model` (*str | ModelSpec*): The model identifier string (e.g. `"allenai/MolmoAct2-BimanualYAM"`) or a typed model constant from `servo.models` (e.g. `servo.models.MolmoAct2BimanualYAM`). String literals and typed models enable static type checking and IDE key autocomplete. Mutually exclusive with `fleet`.
- `fleet` (*str*, optional): Fleet name from `sv.fleets.deploy(..., name=...)`. Selects the model implicitly — pass with `robot`, not `model`.
- `robot` (*str*, optional with `model`, required with `fleet`): A stable robot name. The first call with a given name registers the robot in your organization; subsequent calls attach to that record. Omit entirely for an unregistered, standalone session.
- `site`, `labels` (*str*, *dict[str, str]*, optional): Location and grouping metadata attached on first registration with `robot=`; used by `sv.fleets.deploy(selector=...)`.
- `observe` (*dict[str, Source] | TypedDict | Callable[[], dict[str, Any]]*): Observation channel. Either a slot dictionary mapping model sensor names to `servo.universal.Camera` handles (the type returned by `camera()`), callables, or ROS topic strings, or a zero-argument callable returning the observation dictionary read in one call. Keys are statically type-checked against the target model's `TypedDict` schema.
- `act` (*Callable[[np.ndarray], None | dict] | dict[str, str]*): Actuator sink accepting continuous target positions `(N,) float32` at the control rate (e.g. 30 Hz), or a dict of ROS topic strings. Return `None` for success or `{"done": True}` for early completion; raise to fault (triggers zero-velocity hold). Wrap an existing Gym env with `servo.gym_step(env.step)` rather than returning a raw step tuple.
- `region` (*str*, optional): Target compute region nearest to the physical robot (default: `"us-west-2"`).
- `ros_node` (*rclpy.node.Node*, optional): Active ROS 2 node when binding topic strings.
- `mock` (*bool*, optional): Run against synthetic camera frames and simulated joint physics instead of real Sources/Sinks — validate the rest of the pipeline with no robot on hand (default: `False`).

### `servo.Policy`

Creates a high-performance functional policy callable for custom control loops:

```python
import servo

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

action = policy(
    top=frame_top,
    left=frame_left,
    right=frame_right,
    state=joint_positions,
    task="clean the tabletop",
)
```

Each call returns the first row of the chunk the model returned, directly usable by `env.step(action)`. `instruction=` is accepted as an alias for `task=`. **Current limitation**: unlike `session.run()`'s real background-prefetching control loop, `policy(...)` does not yet buffer or prefetch — every call is one real network round trip. Call it no faster than your model's round-trip allows, or add your own buffering for a tighter loop.

### `servo.camera`

Creates a persistent hardware camera handle opened via zero-copy V4L2 DMA-BUF:

```python
from servo.universal import camera

# Recommended: Persistent hardware serial number (never drifts across reboots or USB replugs):
cam_overhead = camera(serial="250423122040")
cam_wrist = camera(serial="402323071792")

# Single-webcam local debugging:
cam_debug = camera(0)

# Automatic hardware silicon BGR -> RGB color conversion:
cam_bgr = camera(serial="250423122040", color_space="bgr")
```

### `servo.inspect`

Inspects the sensory input schema a model expects for a given robot -- the contract is a
property of the (model, robot) pairing, not the model alone, so `robot` is required:

```python
from servo.universal import inspect

contract = inspect("allenai/MolmoAct2-BimanualYAM", "yam-cell-01")
print(contract.inputs)
# {
#   "top": SensorSpec(type="image", shape=(360, 640, 3), dtype="uint8"),
#   "left": SensorSpec(type="image", shape=(360, 640, 3), dtype="uint8"),
#   "right": SensorSpec(type="image", shape=(360, 640, 3), dtype="uint8"),
#   "state": SensorSpec(type="vector", shape=(14,), units="radians"),
# }
print(contract.outputs)
# {} -- actuator/action-space discovery isn't built yet; only observation inputs are real here.
```

#### Parameters:
- `model` (*str*): The model identifier or hosted slug.
- `robot` (*Robot | str*): A registered robot handle or name (same resolution as `connect(robot=...)`).

#### Returns:
- `ModelContract`: Structured schema object exposing `.inputs` (real) and `.outputs` (always `{}` today) with tensor shapes, data types, and physical units.

#### Exceptions:
- `servo.ValidationError`: Raised by `inspect()` itself when the server can't resolve an observation contract for this (model, robot) pair (e.g. no runtime installed for this family/embodiment yet).
- `servo.universal.ContractMismatchError`: Raised by `servo.connect()` when the provided `observe` dictionary keys or `act` callable shape does not match the model's required slot schema.

### Static Type Checking & Model Schemas

Servo provides first-class static typing via PEP 589 `TypedDict` schemas and string-literal `@overload` signatures for `mypy`, `pyright`, and modern IDEs (VS Code, PyCharm).

```python
from servo.models import MolmoAct2BimanualYAM
from servo.types.molmoact2 import MolmoAct2Observation
from servo.universal import camera, connect

# 1. Type-safe session connection with key autocomplete in IDE:
session = connect(
    MolmoAct2BimanualYAM,
    observe={
        "top": camera(serial="250423122040"),
        "left": camera(serial="402323071792"),
        "right": camera(serial="402323071794"),
        "state": arm.get_joint_positions,
    },
    act=arm.apply_joint_targets,
)

# 2. Type-checking custom environment observation methods:
def capture_snapshot() -> MolmoAct2Observation:
    return {
        "top": overhead_cam.get_frame(),
        "left": wrist_cam_l.get_frame(),
        "right": wrist_cam_r.get_frame(),
        "state": robot.get_state(),
    }
```

### `RobotSession`

Returned by `servo.connect()`. Not the same class as the real `servo.Session` returned by
`sv.session(policy)` (see [Low-level inspection loop](#low-level-inspection-loop) below) — that
one is a stepwise `act()`-plus-evidence-recording handle for a caller-driven loop;
`RobotSession` is a high-level rollout controller whose real (non-mock) `.run()` is sugar over
the `robot.run(policy, seconds=...)` verb.

- `session.run(task: str, timeout_s: float | None = None, max_steps: int | None = None, until: Callable[[], bool] | None = None) -> RunReport`: Executes a bounded autonomous episode (rollout) at the model's native rate (e.g. 30 Hz).
- `session.pause() -> None`: Halts action emission and commands zero-velocity holding targets to actuators.
- `session.resume() -> None`: Flushes stale frames, forces an IDR keyframe, prefetches a fresh chunk, and resumes 30 Hz policy execution.
- `session.get_frame(camera_name: str) -> np.ndarray`: Non-blocking preview (`peek()`) from the local shared-memory ring buffer.
- `session.stop() -> None`: Halts autonomous execution and tears down the session.
- `session.listen() -> None`: Blocks. Keeps `observe`/`act` bindings live and reports readiness over the daemon's control-plane connection; rollouts are then triggered by `robot.start()` from another process (requires `robot=` — see [Remote Start and Stop](/docs/md/guides/robots.md#remote-start-and-stop)).

**Current limitation**: a real (non-mock) `session.run()` is not evidenced or recorded
server-side — no rollout record, no LeRobot dataset — because a bare `connect(model, robot=...)`
binding has no deployment to record against. `connect()` warns at construction time
(`UserWarning`) rather than silently dropping episode data; see `ROBOT_INTEGRATION_SPEC.md`.

### `RunReport`

Returned by `session.run()`:
- `report.status` (*str*): `"completed"`, `"timeout"`, `"fault"`, `"intervened"`, or `"halted"`.
- `report.error` (*str | None*): Diagnostic details or exception trace if `report.status == "fault"`, otherwise `None`.
- `report.steps` (*int*): Total control steps executed during the rollout.
- `report.duration_s` (*float*): Total elapsed wall-clock seconds.
- `report.dataset_path` (*str | None*): Reserved for a future recorded-episode path (in LeRobot format); always `None` today — see the current limitation noted above.
- `report.telemetry` (*dict*): Round-trip latency, prefetch jitter, and dropped-frame counts for the run.
- `report.raise_for_status() -> None`: Raises `servo.universal.ActuatorFaultError` if `status == "fault"` or `TimeoutError` if `status == "timeout"`.
- `print(report)` / `repr(report)`: one-glance summary card (status, duration, steps, dataset path, cockpit replay link).

---

## Client & Resource Namespaces

For multi-robot fleet administration, model querying, and hosted capacity provisioning:

```python
import os
import servo

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

| Namespace | Description |
| --- | --- |
| `sv.robots` | Register, list, resolve, and inspect physical robots |
| `sv.models` | Discover models compatible with a robot contract |
| `sv.deployments` | Manage dedicated hosted compute instances |
| `sv.fleets` | Allocate and scale shared hosted capacity across robot fleets |

### Models and deployments

```python
record = sv.robots.get("yam-cell-01")
compatible_models = sv.models.for_robot(record)

model = sv.models.get("pi0.5")
deployment = model.deploy(robot=record).wait(timeout_s=900)
policy = deployment.policy(record, instruction="place the red lid on the black box")
```

Review active capacity with `servo deployment list`, and stop unused capacity with `servo deployment stop <deployment-id>`.

### Remote start and stop (preview)

For a robot bound locally with `session.listen()` (see [Remote Start and Stop](/docs/md/guides/robots.md#remote-start-and-stop)), control it from any other process:

- `robot.start(task: str, timeout_s: float | None = None) -> RemoteRun`: Non-blocking. Raises `servo.universal.RobotNotListening` if nothing is bound.
- `RemoteRun.status` (*str*): `"running"`, `"completed"`, `"fault"`, etc.
- `RemoteRun.pause() / .resume() / .stop() -> None`: Proxy `session.pause()`/`.resume()`/`.stop()` on the listening robot.
- `RemoteRun.wait(timeout_s: float | None = None) -> RunReport`: Blocks until the rollout ends; same `RunReport` shape as a local `session.run()`.

### Low-level inspection loop

For integration debugging, step-by-step telemetry, or manual unit testing:

```python
with robot, sv.session(policy) as session:
    observation = robot.observe()
    prediction = session.act(observation)
    robot.execute(prediction)
```

- `robot.observe()`: Captures camera frames and joint states adhering to the sensor contract.
- `session.act(observation)`: Queries the hosted model and returns an `ActionPrediction` chunk.
- `robot.execute(prediction)`: Applies action-jump limits and streams joint targets to the local controller.

### Fleets

```python
model = sv.models.get("pi0.5")
fleet = sv.fleets.deploy(
    model,
    robots=["yam-cell-01", "yam-cell-02"],
    name="yam-assembly",
    peak_active=2,
)

policy = fleet.policy(record, instruction="place the red lid on the black box")
fleet.update("ckpt_...")
fleet.rollback()
```

---

# API reference

> Servo Public API v0.1.0. Full machine-readable spec: /docs/openapi.json (OpenAPI 3). Authenticate with an organization API key via the `Authorization: Bearer sk_...` header.

## catalog

- `GET /v1/models` — List compatible models
- `GET /v1/models/{slug}` — Get a model

## customer-auth

- `GET /v1/me/session` — Show the signed-in account
- `POST /v1/me/api-keys` — Create an API key
- `GET /v1/me/api-keys` — List API keys
- `DELETE /v1/me/api-keys/{key_id}` — Revoke an API key
- `GET /v1/me/account` — Show account status
- `POST /v1/me/deployments` — Deploy a model for a robot
- `GET /v1/me/deployments` — List hosted deployments
- `DELETE /v1/me/deployments/{deployment_id}` — Stop a hosted deployment

## fleets

- `POST /v1/fleets` — Create a robot fleet
- `GET /v1/fleets` — List robot fleets
- `GET /v1/fleets/{fleet_id}` — Get a robot fleet
- `POST /v1/fleets/{fleet_id}/model` — Update a fleet model
- `POST /v1/fleets/{fleet_id}/rollback` — Roll back a fleet model

## robots

- `GET /v1/robots/catalog` — List supported robot types
- `POST /v1/robots/setup` — Register a robot
- `GET /v1/robots` — List robots
- `GET /v1/robots/{robot_ref}` — Get a robot

## untagged

- `GET /health` — Health