> ## 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.

# 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 theme={null}
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](/guides/fleet-deployment)):

* `model` (*str*): The model identifier string (e.g. `"allenai/MolmoAct2-BimanualYAM"`) or a string constant from `servo.models` (e.g. `servo.models.MolmoAct2BimanualYAM`). 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] | Callable\[\[], dict\[str, Any]]*, optional with `mock=True`): 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. Annotate a separate dictionary with a model-specific `TypedDict` when you want static slot-name checking.
* `act` (*Callable\[\[np.ndarray], None | dict] | dict\[str, str]*, optional with `mock=True`): 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 a deterministic local fake policy instead of contacting a hosted endpoint (default: `False`). The exact MolmoAct2 development profile can synthesize Sources/Sinks automatically; other model identifiers require explicit `observe` and `act` bindings. Mock mode validates local binding and control-loop behavior, not credentials or network transport.
* `mock_action_dim` (*int*, optional): Positive action-vector dimension for a mock model that has no exact local development profile. It is unnecessary for the built-in MolmoAct2 profile and invalid for non-mock sessions.

### `servo.Policy`

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

```python theme={null}
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 theme={null}
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 theme={null}
from servo.universal import inspect

contract = inspect("allenai/MolmoAct2-BimanualYAM", "yam-cell-01")
print(contract.inputs)
# {
#   "top": SensorSpec(type="image", shape=(378, 378, 3), dtype="uint8"),
#   "left": SensorSpec(type="image", shape=(378, 378, 3), dtype="uint8"),
#   "right": SensorSpec(type="image", shape=(378, 378, 3), dtype="uint8"),
# }
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 | Embodiment | str*): A registered robot or embodiment handle, an
  `emb_*` embodiment ID, or a unique cached embodiment name.

#### Returns:

* `ModelContract`: The current flat compatibility projection. Its `.inputs` preserve basic
  type, shape, and data-type fields; `.outputs` is always `{}` today. It does not yet preserve
  the full server inference contract, including requiredness and action outputs. The current
  MolmoAct2 runtime contract publishes model-prepared camera inputs but not proprioceptive state
  metadata.

#### 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 observed keys do not match an exact local development profile, or when a Source/Sink violates the runtime value contract.

### Static Type Checking & Model Schemas

Servo provides a PEP 589 `TypedDict` schema for its exact MolmoAct2 development profile. Apply the
type explicitly to an observation dictionary for `mypy`, `pyright`, and IDE completion; `connect()`
does not infer that schema from a model string.

```python theme={null}
from servo.models import MolmoAct2BimanualYAM, MolmoAct2Observation
from servo.universal import camera, connect

# Type-check a custom environment observation method:
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(),
    }

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

### `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](/guides/robots#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 theme={null}
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 theme={null}
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](/guides/robots#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 theme={null}
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 theme={null}
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()
```