Python API

The Servo Python SDK provides two primary layers:

Robot Integration & Policy Execution: Connect robot hardware directly to hosted VLA models with servo.connect(), servo.Policy(), servo.camera(), and servo.inspect().
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:

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

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:

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:

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:

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.

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

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:

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

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

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

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

Get started

Use Servo

Robot Integration API

`servo.connect`

Parameters:

`servo.Policy`

`servo.camera`

`servo.inspect`

Parameters:

Returns:

Exceptions:

Static Type Checking & Model Schemas

`RobotSession`

`RunReport`

Client & Resource Namespaces

Models and deployments

Remote start and stop (preview)

Low-level inspection loop

Fleets

​Robot Integration API

​servo.connect

​Parameters:

​servo.Policy

​servo.camera

​servo.inspect

​Parameters:

​Returns:

​Exceptions:

​Static Type Checking & Model Schemas

​RobotSession

​RunReport

​Client & Resource Namespaces

​Models and deployments

​Remote start and stop (preview)

​Low-level inspection loop

​Fleets

Robot Integration API

`servo.connect`

Parameters:

`servo.Policy`

`servo.camera`

`servo.inspect`

Parameters:

Returns:

Exceptions:

Static Type Checking & Model Schemas

`RobotSession`

`RunReport`

Client & Resource Namespaces

Models and deployments

Remote start and stop (preview)

Low-level inspection loop

Fleets