Python API

Deploy models to optimized endpoints, and drive them from your own control loop over the organization network.

Install with pip install servo-client; authenticate with servo login. Runnable examples are in the quickstart.

sv

Typed Servo client for one explicitly selected control-plane origin.

Call	What it does
sv.Servo	Open a client against your organization’s control plane.
sv.for_standalone_endpoint	Create an unauthenticated SDK client for a standalone endpoint session.
sv.session	Open a real-robot control loop for a hosted or standalone policy.
sv.close	Close this client’s HTTP connection pool.

sv.Servo

# sv.Servo
def Servo(base_url=None, timeout=3600.0) -> Servo: ...

Open a client against your organization’s control plane.

Parameter	Type	Default	Description
base_url	str | None	None	Control-plane URL. Defaults to SERVO_BASE_URL, else Servo’s hosted control plane.
timeout	float | None	3600.0	Seconds any one request may take. Serving requests are bounded by their own deadlines, not by this.

sv.for_standalone_endpoint

# sv.for_standalone_endpoint
def for_standalone_endpoint(endpoint_url, *, timeout=3600.0) -> Servo: ...

Create an unauthenticated SDK client for a standalone endpoint session.

Parameter	Type	Default	Description
endpoint_url	str	required	Absolute HTTP(S) URL from the standalone endpoint grant.
timeout	float | None	3600.0	Per-request timeout in seconds.

sv.session

# sv.session
def session(
    policy,
    *,
    instruction=None,
    action_session=None,
    observation_encoding="h264",
    server_telemetry=False,
    record=True,
    record_to=None,
    sample_host_metrics=True,
    host_metrics_interval_s=1.0,
    sensor_history_max_samples=20000,
    sensor_history_max_bytes=64 * 1024 * 1024,
    sensor_history_max_age_ns=10000000000,
    sensor_history_max_consumers=8,
    sensor_history_max_consumer_output_bytes=8 * 1024 * 1024,
) -> Session: ...

Open a real-robot control loop for a hosted or standalone policy.

Parameter	Type	Default	Description
policy	Policy | DirectPolicy	required	Policy to run, as deployment.policy() returns it, or a standalone DirectPolicy returned by servo.direct.attach. A standalone policy uses its grant to control the endpoint and must set record=False or record_to= because it has no hosted cloud recorder.
instruction	str | None	None	Default language instruction for the session, falling back to policy.instruction. A per-act instruction overrides it for that act.
action_session	Any	None	An already-open action session to run over; opened for you when omitted.
observation_encoding	str	"h264"	Wire encoding for camera frames. Only h264 is supported; a session that cannot send H.264 raises rather than silently sending something else.
server_telemetry	bool	False	Ask the server for its full per-prediction telemetry (inference time and latency breakdown). Off by default: predictions then carry only the client-measured timings.
record	bool	True	Record the session for later review.
record_to	str | Path | None	None	Directory or file to write the local recording to. Defaults to the client’s recording location.
sample_host_metrics	bool	True	Sample CPU, memory and link metrics from this computer alongside the session’s own telemetry.
host_metrics_interval_s	float	1.0	Seconds between host metric samples.
sensor_history_max_samples	int	20000	Most snapshots to retain in sensor_history. The oldest is dropped once the buffer is full.
sensor_history_max_bytes	int	64 * 1024 * 1024	Memory ceiling for sensor_history, in bytes.
sensor_history_max_age_ns	int	10000000000	Age at which a retained snapshot is dropped, in nanoseconds, so a stalled consumer cannot read a stale frame.
sensor_history_max_consumers	int	8	Most consumers that may read sensor_history at once.
sensor_history_max_consumer_output_bytes	int	8 * 1024 * 1024	Memory ceiling for one consumer’s selected output, in bytes.

sv.close

# sv.close
def close() -> None: ...

Close this client’s HTTP connection pool.

sv.access

Who holds read or write on a robot.

Call	What it does
sv.access.grant	Give a person or a Google Group exactly role on robot, replacing any other.
sv.access.revoke	Revoke a grant by the address it was made under, or by its listed grant ID.
sv.access.list	Active grants on robot. Organization admins hold every role and are not listed.

sv.access.grant

# sv.access.grant
def grant(subject, *, robot, role) -> None: ...

Give a person or a Google Group exactly role on robot, replacing any other.

Parameter	Type	Default	Description
subject	str	required	Email address of an organization member, or a Google Group address once an admin has connected Google Workspace.
robot	Robot | str	required	Robot name or handle the grant applies to.
role	str	required	read to inspect the robot and its sessions, or write to also start and stop sessions on it.

sv.access.revoke

# sv.access.revoke
def revoke(subject, *, robot) -> None: ...

Revoke a grant by the address it was made under, or by its listed grant ID.

Parameter	Type	Default	Description
subject	str	required	The address the grant was made under, or the grant ID that list reported for it.
robot	Robot | str	required	Robot name or handle the grant applies to.

sv.access.list

# sv.access.list
def list(*, robot) -> list[Grant]: ...

Active grants on robot. Organization admins hold every role and are not listed.

Parameter	Type	Default	Description
robot	Robot | str	required	Robot name or handle to list grants for.

sv.deployments

Hosted deployments: a model running on capacity near a site.

Call	What it does
sv.deployments.get	Fetch a hosted deployment.

sv.deployments.get

# sv.deployments.get
def get(id) -> HostedDeployment: ...

Fetch a hosted deployment.

Parameter	Type	Default	Description
id	str	required	Deployment ID, as servo model deploy and Model.deploy report it (dep_ followed by its hexadecimal suffix).

sv.models

The models your organization can serve: imported checkpoints and catalog entries.

Call	What it does
sv.models.list	List visible models and their current managed serving policy.
sv.models.get	Fetch one model by the slug it was imported under.
sv.models.delete	Delete one of your organization’s models, with its checkpoints and preparations.
sv.models.for_robot	List visible models after verifying access to the registered robot.
sv.models.inspect_checkpoint	Read a Hugging Face checkpoint’s own input and action shapes, without a robot.
sv.models.import_checkpoint	Import a supported checkpoint from Hugging Face or a governed public source.

sv.models.list

# sv.models.list
def list(robot=None) -> list[Model]: ...

List visible models and their current managed serving policy.

Parameter	Type	Default	Description
robot	Robot | str | None	None	Robot name or handle to limit the list to models compatible with it. Omit it for every model in the organization.

sv.models.get

# sv.models.get
def get(slug) -> Model: ...

Fetch one model by the slug it was imported under.

Parameter	Type	Default	Description
slug	str	required	Model identifier, as list and servo model list report it.

sv.models.delete

# sv.models.delete
def delete(model_id) -> None: ...

Delete one of your organization’s models, with its checkpoints and preparations.

Parameter	Type	Default	Description
model_id	str	required	The model’s id or slug, as list reports it.

sv.models.for_robot

# sv.models.for_robot
def for_robot(robot) -> list[Model]: ...

List visible models after verifying access to the registered robot.

Parameter	Type	Default	Description
robot	Robot | str	required	Robot name or handle. You must hold access to it.

sv.models.inspect_checkpoint

# sv.models.inspect_checkpoint
def inspect_checkpoint(
    source,
    *,
    revision="main",
    payload_subdir=None,
    openpi_config_name=None,
    checkpoint_options=None,
) -> ModelInspection: ...

Read a Hugging Face checkpoint’s own input and action shapes, without a robot.

Parameter	Type	Default	Description
source	str	required	Hugging Face URL, hf://ORG/REPO or ORG/REPO.
revision	str	"main"	Commit, tag or branch to read. Pin a commit for a reproducible contract; main moves as the upstream repository moves.
payload_subdir	str | None	None	Path within the repository holding the checkpoint, when it is not at the repository root.
openpi_config_name	str | None	None	OpenPI config to read the contract under, when the repository ships several and names none as its default.
checkpoint_options	Mapping[str, Any] | None	None	Model-owned selections declared by inspection, such as a GR00T embodiment_tag when several complete robot configurations remain. A sole complete configuration is selected from the checkpoint’s own metadata; repository names are not used to guess it.

sv.models.import_checkpoint

# sv.models.import_checkpoint
def import_checkpoint(
    source,
    *,
    name=None,
    slug=None,
    revision="main",
    payload_subdir=None,
    control_source_hz=None,
    morphology=None,
    prompt_key=None,
    image_resolution=None,
    openpi_action_horizon=None,
    openpi_config_name=None,
    checkpoint_options=None,
    robot=None,
    wait=True,
    timeout_s=2400,
) -> Model: ...

Import a supported checkpoint from Hugging Face or a governed public source.

Parameter	Type	Default	Description
source	str	required	Hugging Face URL, hf://ORG/REPO or ORG/REPO.
name	str | None	None	Human-readable name for the imported model. Defaults to the repository’s own name.
slug	str | None	None	Identifier to address the model by later. Defaults to one derived from source.
revision	str	"main"	Commit, tag or branch to import. The resolved commit is pinned, so a later upstream push does not change what you serve.
payload_subdir	str | None	None	Path within the repository holding the checkpoint, when it is not at the repository root.
control_source_hz	float | None	None	Control rate the checkpoint’s demonstrations were recorded at, when the upstream files do not declare it.
morphology	str | None	None	Embodiment the checkpoint was trained for, when the upstream files do not declare it.
prompt_key	str | None	None	Observation key carrying the language instruction, when the checkpoint names it something other than the runtime’s default.
image_resolution	dict[str, tuple[int, int]] | None	None	Per-camera {name: (height, width)} the checkpoint expects, when the upstream files do not declare it.
openpi_action_horizon	int | None	None	Action chunk length for a native JAX/Orbax OpenPI fine-tune; defaults to 50 unless the checkpoint config or caller supplies it.
openpi_config_name	str | None	None	OpenPI config to import under, when the repository ships several and names none as its default.
checkpoint_options	Mapping[str, Any] | None	None	Model-owned selections the inspection lists under checkpoint_native.accepts and requires, such as a GR00T embodiment_tag when multiple complete configurations remain. A sole complete configuration is selected automatically from checkpoint metadata.
robot	Robot | str | None	None	Enrolled robot to bind the inspected contract to, so compatibility is resolved against that embodiment at import rather than at deploy.
wait	bool	True	Wait for the target to load and for the optimized serving profile to qualify. False returns a pending handle to Model.wait later.
timeout_s	float	2400	Seconds to wait when wait is set. A timeout leaves preparation running and raises; it does not cancel the import.

sv.robots

The robots enrolled in your organization.

Call	What it does
sv.robots.get	Fetch one enrolled robot by name or ID.
sv.robots.list

sv.robots.get

# sv.robots.get
def get(id) -> Robot: ...

Fetch one enrolled robot by name or ID.

Parameter	Type	Default	Description
id	str	required	Robot name as it was enrolled (robot-1), or its rbt_ ID.

sv.robots.list

# sv.robots.list
def list() -> list[Robot]: ...

sv.sessions

Session records: who ran what on which robot, open or finished.

Call	What it does
sv.sessions.list	Open sessions, newest first; robot is a name or a robot handle to filter by.
sv.sessions.get	Read one session’s record, open or finished.

sv.sessions.list

# sv.sessions.list
def list(*, robot=None) -> list[SessionRecord]: ...

Open sessions, newest first; robot is a name or a robot handle to filter by.

Parameter	Type	Default	Description
robot	Robot | str | None	None	Robot name or handle to filter by. Omit it for every session you can see across the organization.

sv.sessions.get

# sv.sessions.get
def get(session_id) -> SessionRecord: ...

Read one session’s record, open or finished.

Parameter	Type	Default	Description
session_id	str	required	Session ID, as list reports it.

model

A catalog model (which weights). slug is the stable id; id mirrors it for handle uniformity.

You get one from sv.models.for_robot, sv.models.get, sv.models.import_checkpoint, sv.models.list.

Call	What it does
model.wait	Wait for target load, optimized preparation and cache reuse qualification.
model.checkpoints	The model’s fine-tuned checkpoints (hydrates if partial).
model.deploy	Start, or return, the hosted deployment of this model near a site.
model.compatibility	Report whether this model can drive robot, and what would have to match.

model.wait

# model.wait
def wait(*, timeout_s=2400, poll_interval_s=1, log=True) -> Model: ...

Wait for target load, optimized preparation and cache reuse qualification.

Parameter	Type	Default	Description
timeout_s	float	2400	Seconds to wait. A timeout raises and leaves the work running, so calling again resumes waiting rather than restarting anything.
poll_interval_s	float	1	Seconds between status reads.
log	bool	True	Print each status change as it happens.

model.checkpoints

# model.checkpoints
def checkpoints() -> list["Checkpoint"]: ...

The model’s fine-tuned checkpoints (hydrates if partial).

model.deploy

# model.deploy
def deploy(
    robot=None,
    *,
    region=None,
    city=None,
    latitude=None,
    longitude=None,
    name=None,
    idle_minutes=_IDLE_UNSET,
    max_sessions=None,
    scaling=None,
    wait=True,
) -> HostedDeployment: ...

Start, or return, the hosted deployment of this model near a site.

Parameter	Type	Default	Description
robot	Robot | str | None	None	Enrolled robot whose site to deploy near. Sugar for that robot’s site: the deployment is never bound to the robot.
region	str | None	None	Placement region code, such as eu-central-2.
city	str | None	None	City to deploy near, such as Zurich, Switzerland.
latitude	float | None	None	Latitude to deploy near. Pass longitude with it.
longitude	float | None	None	Longitude to deploy near. Pass latitude with it.
name	str | None	None	Name for this deployment among your organization’s at that site. The default shares one deployment per model and site.
idle_minutes	float | None	_IDLE_UNSET	Release the endpoint after this long with no act and no active sessions. Omit it for the 240 minute default, which is also the most.
max_sessions	int | None	None	How many sessions may run against it at once.
scaling	ServingScalingPolicy | None	None	Typed replica policy: bounded count, per-replica session capacity, cooldown and optional hourly cost limit.
wait	bool	True	Wait until the deployment is ready. False returns at once; call deployment.wait() later.

model.compatibility

# model.compatibility
def compatibility(robot) -> CompatibilityReport: ...

Report whether this model can drive robot, and what would have to match.

Parameter	Type	Default	Description
robot	Robot | str	required	Robot name or handle to check this model’s contract against.

deployment

A hosted endpoint with exact policy() / world() factories.

You get one from checkpoint.deploy, model.deploy, sv.deployments.get.

Call	What it does
deployment.wait	Poll until the deployment reaches a terminal status.
deployment.rollback
deployment.session	Open the control loop on this deployment as robot (default: this enrolled computer).
deployment.policy	Bind this deployment’s exact active generation as an action policy.
deployment.metrics	Task-level episode metrics attributed to this deployment (success/intervention/drift).
deployment.logs	Endpoint log records for this deployment, oldest first.
deployment.delete	Request asynchronous teardown and retain a pollable handle.
deployment.wait_retired	Wait for requested teardown to reach the authoritative retired state.

deployment.wait

# deployment.wait
def wait(*, timeout_s=2400, poll_interval_s=1, log=True) -> HostedDeployment: ...

Poll until the deployment reaches a terminal status.

Parameter	Type	Default	Description
timeout_s	float	2400	Seconds to wait. A timeout raises and leaves provisioning running, so calling again resumes waiting rather than restarting anything.
poll_interval_s	float	1	Seconds between status reads.
log	bool	True	Print each status change as it happens.

deployment.rollback

# deployment.rollback
def rollback() -> HostedDeployment: ...

deployment.session

# deployment.session
def session(
    instruction=None,
    *,
    robot=None,
    server_telemetry=False,
    **kwargs,
) -> Session: ...

Open the control loop on this deployment as robot (default: this enrolled computer).

Parameter	Type	Default	Description
instruction	str | None	None	Default language instruction for every prediction in the session. A per-call instruction overrides it for that request.
robot	Robot | str | None	None	Robot to act as. Defaults to the robot enrolled on this computer.
server_telemetry	bool	False	Ask the server for its full per-prediction telemetry (inference time and latency breakdown). Off by default.

deployment.policy

# deployment.policy
def policy(robot=None, *, instruction=None, record=True, record_to=None) -> Policy: ...

Bind this deployment’s exact active generation as an action policy.

Parameter	Type	Default	Description
robot	Robot | str | None	None	Robot to act as. Defaults to the robot enrolled on this computer.
instruction	str | None	None	Default language instruction for sessions opened on this policy.
record	bool	True	Record sessions opened on this policy for later review.
record_to	str | Path | None	None	Directory or file for the local recording. Defaults to the client’s recording location.

deployment.metrics

# deployment.metrics
def metrics() -> dict[str, Any]: ...

Task-level episode metrics attributed to this deployment (success/intervention/drift).

deployment.logs

# deployment.logs
def logs(*, since_s=86400, limit=100) -> list: ...

Endpoint log records for this deployment, oldest first.

Parameter	Type	Default	Description
since_s	float	86400	How far back to read, in seconds. Must be finite and positive.
limit	int	100	Most records to return, newest first.

deployment.delete

# deployment.delete
def delete() -> HostedDeployment: ...

Request asynchronous teardown and retain a pollable handle.

deployment.wait_retired

# deployment.wait_retired
def wait_retired(*, timeout_s=300, poll_interval_s=1) -> HostedDeployment: ...

Wait for requested teardown to reach the authoritative retired state.

Parameter	Type	Default	Description
timeout_s	float	300	Seconds to wait for the retired status. A timeout raises and leaves teardown running.
poll_interval_s	float	1	Seconds between status reads.

policy

A bound action policy. model_ref is a slug or a ckpt_... id; routing is the binding.

You get one from checkpoint.policy, deployment.policy.

Call	What it does
policy.close	Finalize the persistent manual-prediction session and its recording.

policy.close

# policy.close
def close() -> None: ...

Finalize the persistent manual-prediction session and its recording.

session

A real-robot loop over an action session.

You get one from deployment.session, sv.session.

Call	What it does
session.latest_host_metrics	Return the most recent host metrics snapshot as a JSON-serializable dictionary.
session.open
session.act	One action chunk for observation.
session.predict	Predict from checkpoint-named inputs without reading or commanding hardware.
session.configure_recovery	Register local recovery for a customer-owned loop before its first action.
session.predict_from_sensor_reader	Read customer-owned sensors and predict without copying their buffers.
session.record_sensor_consumer	Record an auxiliary consumer’s selection lineage and optional result.
session.feedback	Report what the robot actually executed, for later review and training.
session.close	Close the session and release its slot on the deployment.

session.latest_host_metrics

# session.latest_host_metrics
def latest_host_metrics() -> dict[str, Any] | None: ...

Return the most recent host metrics snapshot as a JSON-serializable dictionary.

session.open

# session.open
def open() -> Session: ...

session.act

# session.act
def act(observation, instruction=None) -> ActionPrediction: ...

One action chunk for observation.

Parameter	Type	Default	Description
observation	Any	required	Structured observation to act on, carrying the sensor frames and state the policy’s contract declares.
instruction	str | None	None	Language instruction for this act only. Takes precedence over the observation’s own instruction and the session default.

session.predict

# session.predict
def predict(
    *,
    inputs,
    instruction=None,
    noise_seed=None,
    sensor_consumer=None,
    consumer_now=None,
    now_by_clock=None,
    clock_mappings=None,
) -> ActionPrediction: ...

Predict from checkpoint-named inputs without reading or commanding hardware.

Parameter	Type	Default	Description
inputs	Mapping[str, Any]	required	Checkpoint feature names mapped to NumPy arrays matching the inspected shapes and dtypes. Extra keys, metadata among them, are rejected rather than ignored.
instruction	str | None	None	Language instruction for this prediction only, overriding the session default.
noise_seed	int | None	None	Seed for reproducible model sampling in evaluations. Omit it for ordinary sampling.
sensor_consumer	SensorHistoryConsumer | None	None	Consumer selecting a timestamped snapshot from sensor_history, so acquisition cadence stays independent of this call.
consumer_now	ClockStamp | None	None	The consumer’s own clock reading for this request, used to choose which buffered snapshot is current.
now_by_clock	Mapping[str, ClockStamp] | None	None	Current reading of each named clock, for a caller whose sensors timestamp against more than one.
clock_mappings	Mapping[str, ClockMapping] | None	None	How each named clock relates to the others, so snapshots from different sensors can be aligned.

session.configure_recovery

# session.configure_recovery
def configure_recovery(
    *,
    handler,
    controller,
    takeover_timeout_s=1.0,
) -> RecoveryRegistration: ...

Register local recovery for a customer-owned loop before its first action.

Parameter	Type	Default	Description
handler	Callable[[RecoveryContext, RecoveryHandle], Awaitable[None]]	required	Async callable invoked when recovery is needed, receiving the context and a handle to act through.
controller	RecoveryController	required	Your controller, which must route every hardware write through its dispatch methods so recovery can fence them.
takeover_timeout_s	float	1.0	Seconds the handler has to take over before the session treats the controller as unresponsive.

session.predict_from_sensor_reader

# session.predict_from_sensor_reader
def predict_from_sensor_reader(
    reader,
    layout,
    *,
    roles=None,
    instruction=None,
) -> ActionPrediction: ...

Read customer-owned sensors and predict without copying their buffers.

Parameter	Type	Default	Description
reader	Any	required	Your sensor reader, exposing the buffers layout describes.
layout	Any	required	Description of how the reader’s buffers map onto checkpoint feature names, shapes and dtypes.
roles	Any	None	Camera role assignment, when the reader’s names differ from the ones the checkpoint declares.
instruction	str | None	None	Language instruction for this prediction only, overriding the session default.

session.record_sensor_consumer

# session.record_sensor_consumer
def record_sensor_consumer(selection, *, result=None) -> bool: ...

Record an auxiliary consumer’s selection lineage and optional result.

Parameter	Type	Default	Description
selection	Any	required	The SensorSelection the consumer resolved, carrying which snapshot it read and against which clock.
result	Mapping[str, Any] | None	None	Whatever the consumer produced from that snapshot, recorded with the selection for later review.

session.feedback

# session.feedback
def feedback(
    executed_actions,
    *,
    executed_step=None,
    reward=None,
    success=None,
    robot_executed_at_ns=None,
) -> None: ...

Report what the robot actually executed, for later review and training.

Parameter	Type	Default	Description
executed_actions	list[list[float]]	required	The action rows the robot actually applied, which may differ from the predicted chunk after your safety limits.
executed_step	int | None	None	Index within the predicted chunk that execution reached.
reward	float | None	None	Scalar reward for the step, when your task defines one.
success	bool | None	None	Whether the task succeeded, when this reports a finished attempt.
robot_executed_at_ns	int | None	None	When the robot applied the actions, in nanoseconds on the robot’s clock, so server timing can be aligned to it.

session.close

# session.close
def close(success=True, *, suppress_completion_error=False) -> None: ...

Close the session and release its slot on the deployment.

Parameter	Type	Default	Description
success	bool	True	Record the session as having completed its task. Pass False to mark it unsuccessful for later review.
suppress_completion_error	bool	False	Return normally even if the server rejects the completion record, for a caller shutting down anyway.

recovery_registration

Created by Session.configure_recovery; one registration owns one controller.

You get one from session.configure_recovery.

Call	What it does
recovery_registration.admit
recovery_registration.update_context	Publish the controller’s latest view, so a handler sees current state.
recovery_registration.notify_starvation	Report that the action buffer ran dry, triggering recovery.
recovery_registration.interrupt	Hand control to the recovery handler.
recovery_registration.hold	Hold the robot under manual control and return the token that releases it.
recovery_registration.release_hold	Release a manual hold.
recovery_registration.record_action	Record one recovery step the handler took, for the incident record.
recovery_registration.resume	Return control to the policy, reporting whether the handover succeeded.
recovery_registration.stop

recovery_registration.admit

# recovery_registration.admit
def admit() -> int: ...

recovery_registration.update_context

# recovery_registration.update_context
def update_context(*, observations, queue) -> None: ...

Publish the controller’s latest view, so a handler sees current state.

Parameter	Type	Default	Description
observations	Mapping[str, Any]	required	Most recent sensor readings the controller holds.
queue	Mapping[str, Any]	required	Its action-queue state, such as depth and the step it has reached.

recovery_registration.notify_starvation

# recovery_registration.notify_starvation
def notify_starvation(*, queue=None) -> None: ...

Report that the action buffer ran dry, triggering recovery.

Parameter	Type	Default	Description
queue	Mapping[str, Any] | None	None	Queue state at the moment of starvation, recorded with the incident.

recovery_registration.interrupt

# recovery_registration.interrupt
def interrupt(reason) -> None: ...

Hand control to the recovery handler.

Parameter	Type	Default	Description
reason	str	required	Why recovery is needed: buffer_starved or link_lost.

recovery_registration.hold

# recovery_registration.hold
def hold(handle, reason) -> str: ...

Hold the robot under manual control and return the token that releases it.

Parameter	Type	Default	Description
handle	RecoveryHandle	required	The handle for this incident, as the handler received it.
reason	str	required	Why the hold was placed, recorded with the incident.

recovery_registration.release_hold

# recovery_registration.release_hold
def release_hold(handle, token) -> None: ...

Release a manual hold.

Parameter	Type	Default	Description
handle	RecoveryHandle	required	The handle for this incident, as the handler received it.
token	str	required	The token hold returned when the hold was placed.

recovery_registration.record_action

# recovery_registration.record_action
def record_action(handle, name, outcome, details) -> None: ...

Record one recovery step the handler took, for the incident record.

Parameter	Type	Default	Description
handle	RecoveryHandle	required	The handle for this incident, as the handler received it.
name	str	required	What the handler did, such as retract or reseat.
outcome	str	required	How it ended, such as ok or failed.
details	Mapping[str, Any]	required	Whatever else is worth recording about the step.

recovery_registration.resume

# recovery_registration.resume
def resume(handle, *, timeout_s) -> ResumeResult: ...

Return control to the policy, reporting whether the handover succeeded.

Parameter	Type	Default	Description
handle	RecoveryHandle	required	The handle for this incident, as the handler received it.
timeout_s	float	required	Seconds to wait for the policy to take control back. Must be finite and positive.

recovery_registration.stop

# recovery_registration.stop
def stop() -> None: ...

checkpoint

Exact learned weights with lineage. The unit you deploy, export, evaluate, or run.

Call	What it does
checkpoint.deploy	Start, or return, the hosted deployment of this checkpoint near a site.
checkpoint.evaluate	Submit an eval of this checkpoint -> a Job (.wait() -> EvalResult).
checkpoint.evals	All eval scorecards recorded for this checkpoint, as typed EvalResult.
checkpoint.publish_to_hf	Publish this checkpoint’s fine-tuned weights to the Hugging Face Hub -> the live model repo URL.
checkpoint.policy	Bind this checkpoint to robot as an action policy.

checkpoint.deploy

# checkpoint.deploy
def deploy(
    robot=None,
    *,
    region=None,
    city=None,
    latitude=None,
    longitude=None,
    name=None,
    idle_minutes=_IDLE_UNSET,
    max_sessions=None,
    scaling=None,
    wait=True,
) -> HostedDeployment: ...

Start, or return, the hosted deployment of this checkpoint near a site.

Parameter	Type	Default	Description
robot	Robot | str | None	None	Enrolled robot whose site to deploy near. Sugar for that robot’s site: the deployment is never bound to the robot.
region	str | None	None	Placement region code, such as eu-central-2.
city	str | None	None	City to deploy near, such as Zurich, Switzerland.
latitude	float | None	None	Latitude to deploy near. Pass longitude with it.
longitude	float | None	None	Longitude to deploy near. Pass latitude with it.
name	str | None	None	Name for this deployment among your organization’s at that site. The default shares one deployment per model and site.
idle_minutes	float | None	_IDLE_UNSET	Release the endpoint after this long with no act and no active sessions. Omit it for the 240 minute default, which is also the most.
max_sessions	int | None	None	How many sessions may run against it at once.
scaling	ServingScalingPolicy | None	None	Typed replica policy: bounded count, per-replica session capacity, cooldown and optional hourly cost limit.
wait	bool	True	Wait until the deployment is ready. False returns at once; call deployment.wait() later.

checkpoint.evaluate

# checkpoint.evaluate
def evaluate(
    rung="offline",
    benchmark=None,
    dataset_split=None,
    sim_ref=None,
    *,
    sim=None,
    task=None,
    sim_url=None,
    options=None,
    **kwargs,
) -> Job: ...

Submit an eval of this checkpoint -> a Job (.wait() -> EvalResult).

Parameter	Type	Default	Description
rung	str	"offline"	Which rung to evaluate on, such as offline for a recorded dataset or byo_sim for your own simulator. Passing sim_url selects byo_sim for you.
benchmark	str | None	None	Named benchmark to run, when the rung offers more than one.
dataset_split	str | None	None	Split of the benchmark’s dataset to evaluate against, such as a held-out validation split.
sim_ref	str | None	None	Reference to the simulator build to run against.
sim	str | None	None	Simulator to run in, when the rung offers a choice.
task	str | None	None	Task within the benchmark to evaluate, when it defines several.
sim_url	str | None	None	URL of your own running simulator. Passing it selects the byo_sim rung and supplies sim_ref.
options	dict[str, Any] | None	None	Runtime options passed through to the eval job.

checkpoint.evals

# checkpoint.evals
def evals() -> list["EvalResult"]: ...

All eval scorecards recorded for this checkpoint, as typed EvalResult.

checkpoint.publish_to_hf

# checkpoint.publish_to_hf
def publish_to_hf(repo, private=False) -> str: ...

Publish this checkpoint’s fine-tuned weights to the Hugging Face Hub -> the live model repo URL.

Parameter	Type	Default	Description
repo	str	required	Target Hugging Face repository as ORG/REPO.
private	bool	False	Create the repository private rather than public.

checkpoint.policy

# checkpoint.policy
def policy(robot, *, instruction=None) -> Policy: ...

Bind this checkpoint to robot as an action policy.

Parameter	Type	Default	Description
robot	Robot | str	required	Robot the policy acts as. Its embodiment resolves the tensor contract.
instruction	str | None	None	Default language instruction for sessions opened on this policy.

recovery_controller

Customer-owned hardware hooks with SDK-owned final-dispatch arbitration.

Call	What it does
recovery_controller.capture_inputs	Acquire and validate all required sensors locally, outside the dispatch lock.
recovery_controller.clear_policy_queue	Discard queued policy rows. Called under the dispatch lock.
recovery_controller.install_policy_chunk	Replace the policy queue without sending actions; called under the lock.
recovery_controller.send_action	Perform one bounded hardware write; called only under the dispatch lock.
recovery_controller.validate_resume	Check current required-sensor readiness and first-row/controller guards locally.
recovery_controller.send_policy	Dispatch a row only under the generation captured at prediction admission.
recovery_controller.send_recovery	Dispatch a row on behalf of the recovery handler that owns the controller.
recovery_controller.send_operator	One acknowledged hold owner may command the controller locally.

recovery_controller.capture_inputs

# recovery_controller.capture_inputs
def capture_inputs() -> RecoveryObservation: ...

Acquire and validate all required sensors locally, outside the dispatch lock.

recovery_controller.clear_policy_queue

# recovery_controller.clear_policy_queue
def clear_policy_queue() -> None: ...

Discard queued policy rows. Called under the dispatch lock.

recovery_controller.install_policy_chunk

# recovery_controller.install_policy_chunk
def install_policy_chunk(prediction) -> None: ...

Replace the policy queue without sending actions; called under the lock.

Parameter	Type	Default	Description
prediction	ActionPrediction	required	The chunk whose rows replace the queue. Its control generation is what later send_policy calls are checked against.

recovery_controller.send_action

# recovery_controller.send_action
def send_action(action) -> Any: ...

Perform one bounded hardware write; called only under the dispatch lock.

Parameter	Type	Default	Description
action	Any	required	One action row, in whatever form your hardware takes.

recovery_controller.validate_resume

# recovery_controller.validate_resume
def validate_resume(observation, prediction) -> bool: ...

Check current required-sensor readiness and first-row/controller guards locally.

Parameter	Type	Default	Description
observation	RecoveryObservation	required	The sensor snapshot capture_inputs just produced.
prediction	ActionPrediction	required	The chunk about to resume, whose first row the guards check.

recovery_controller.send_policy

# recovery_controller.send_policy
def send_policy(prediction, action) -> Any: ...

Dispatch a row only under the generation captured at prediction admission.

Parameter	Type	Default	Description
prediction	ActionPrediction	required	The chunk this row came from; dispatch is refused once its control generation no longer matches the controller’s.
action	Any	required	One action row, in whatever form your hardware takes.

recovery_controller.send_recovery

# recovery_controller.send_recovery
def send_recovery(incident_id, action) -> Any: ...

Dispatch a row on behalf of the recovery handler that owns the controller.

Parameter	Type	Default	Description
incident_id	str	required	The incident the handler is acting under; dispatch is refused for any other, and while a manual hold is in place.
action	Any	required	One action row, in whatever form your hardware takes.

recovery_controller.send_operator

# recovery_controller.send_operator
def send_operator(hold_token, action) -> Any: ...

One acknowledged hold owner may command the controller locally.

Parameter	Type	Default	Description
hold_token	str	required	The token hold returned, which must be the only hold in place.
action	Any	required	One action row, in whatever form your hardware takes.

recovery_handle

Call	What it does
recovery_handle.resume	Hand control back to the policy, reporting whether the handover succeeded.
recovery_handle.hold	Hold the robot under manual control, returning the token that releases it.
recovery_handle.release_hold	Release a manual hold.
recovery_handle.record_action	Record one step the handler took, for the incident record.

recovery_handle.resume

# recovery_handle.resume
def resume(*, timeout_s=10.0) -> ResumeResult: ...

Hand control back to the policy, reporting whether the handover succeeded.

Parameter	Type	Default	Description
timeout_s	float	10.0	Seconds to wait for the policy to take control back.

recovery_handle.hold

# recovery_handle.hold
def hold(reason) -> str: ...

Hold the robot under manual control, returning the token that releases it.

Parameter	Type	Default	Description
reason	str	required	Why the hold was placed, recorded with the incident.

recovery_handle.release_hold

# recovery_handle.release_hold
def release_hold(token) -> None: ...

Release a manual hold.

Parameter	Type	Default	Description
token	str	required	The token hold returned when the hold was placed.

recovery_handle.record_action

# recovery_handle.record_action
def record_action(name, *, outcome, details) -> None: ...

Record one step the handler took, for the incident record.

Parameter	Type	Default	Description
name	str	required	What the handler did, such as retract or reseat.
outcome	str	required	How it ended, such as ok or failed.
details	Mapping[str, Any]	required	Whatever else is worth recording about the step.