Provide model inputs

Pass the observations a trained policy expects without giving Servo a robot-specific camera driver.

Your application captures model inputs and applies returned actions. Read the checkpoint’s own feature names and shapes before writing the NumPy mapping; inspection does not require a registered robot:

robot computer

import servo
import numpy as np

sv = servo.Servo()
inspection = sv.models.inspect_checkpoint("allenai/MolmoAct2-BimanualYAM")
contract = inspection.input_contract
print(contract.resolution)
for feature in contract.inputs or []:
    print(feature.name, feature.modality, feature.shape)
if contract.outputs and "actions" in contract.outputs:
    print(contract.outputs["actions"].shape)

This checkpoint-owned contract lists feature keys, modalities, shapes, and action chunk geometry. It does not assign physical cameras, define state or action semantics, or prove serving support. Fine-tuned Pi0.5 checkpoints can declare different numbers of visual inputs. Missing or contradictory facts remain unknown; the CLI exits nonzero instead of guessing.

observation.state follows the checkpoint’s published state representation; it need not represent joint positions.

Attach capture metadata (optional)

robot computer

inputs = {"metadata": {
    "capture_timestamps_ns": {"wrist": 123456789},
    "capture_clock_ids": {"wrist": "robot-clock-1"},
    "frame_ids": {"wrist": 42},
    "dropped_frames": {"wrist": 0},
    "robot_receive_time_ns": 123457000,
    "robot_receive_clock_id": "robot-clock-1",
}}

Hosted recordings retain the supplied capture timestamps, frame IDs, drops, and clock IDs.

Use the exact feature names from the contract as mapping keys. The customer chooses how to read each physical sensor and state source:

robot computer

import servo

sv = servo.Servo()
robot = sv.robots.get("yam-cell-01")
model = sv.models.get("hf://allenai/MolmoAct2-BimanualYAM")
deployment = model.deploy(robot)
deployment.wait(timeout_s=900)
policy = deployment.policy(robot)
running = True

def capture_inputs_from_customer_robot():
    def read_numpy_array(feature):
        # Return the array from the camera or state reader mapped to this feature name.
        raise NotImplementedError(feature.name)

    return {
        feature.name: read_numpy_array(feature)
        for feature in contract.inputs or []
    }

def apply_actions_to_customer_robot(chunk):
    pass

with sv.session(policy, instruction="pick up the red cup") as session:
    while running:
        inputs = capture_inputs_from_customer_robot()
        chunk = session.predict(inputs=inputs)
        apply_actions_to_customer_robot(chunk)

Servo validates each mapping. Customers own sensor access and applying returned actions.

You now have

• Checkpoint-named NumPy inputs, independent of robot taxonomy.
• Optional camera capture metadata and runtime input validation.
• Customer-owned capture and action application.

Next

Continue with Keep your own loop to schedule action chunks.