Robot and factory¶
Every keyword the Robot factory accepts and every method the engine or hardware robot it returns has.
Robot(name, mode=...) is a factory function, not a class: it returns a simulation engine in mode="sim" (the default) or a strands_robots.hardware_robot.Robot in mode="real"; both expose send_action, get_observation, run_policy and cleanup, but run_policy differs: hardware takes a Policy first, sim a robot name and a provider, or a built Policy as policy_object=. Below: every factory keyword and method of the returned object.
The factory¶
strands_robots.robot.Robot ¶
Robot(name: str, mode: Literal['sim'] = ..., backend: str = ..., urdf_path: str | None = ..., cameras: dict[str, dict[str, Any]] | None = ..., position: list[float] | None = ..., data_config: str | None = ..., *, orientation: list[float] | None = ..., keyframe: str | int | None = ..., driver: str = ..., tool_name: str | None = ..., **kwargs: Any) -> Simulation
Robot(name: str, mode: Literal['real'], backend: str = ..., urdf_path: str | None = ..., cameras: dict[str, dict[str, Any]] | None = ..., position: list[float] | None = ..., data_config: str | None = ..., *, orientation: list[float] | None = ..., keyframe: str | int | None = ..., driver: Literal['strands'], tool_name: str | None = ..., **kwargs: Any) -> HardwareDriver
Robot(name: str, mode: Literal['real'], backend: str = ..., urdf_path: str | None = ..., cameras: dict[str, dict[str, Any]] | None = ..., position: list[float] | None = ..., data_config: str | None = ..., *, orientation: list[float] | None = ..., keyframe: str | int | None = ..., driver: str = ..., tool_name: str | None = ..., **kwargs: Any) -> HardwareRobot
Robot(name: str, mode: Literal['auto'] | str = ..., backend: str = ..., urdf_path: str | None = ..., cameras: dict[str, dict[str, Any]] | None = ..., position: list[float] | None = ..., data_config: str | None = ..., *, orientation: list[float] | None = ..., keyframe: str | int | None = ..., driver: str = ..., tool_name: str | None = ..., **kwargs: Any) -> Simulation | HardwareRobot
Robot(name: str, mode: str = 'sim', backend: str = 'mujoco', urdf_path: str | None = None, cameras: dict[str, dict[str, Any]] | None = None, position: list[float] | None = None, data_config: str | None = None, mesh: bool | None = None, peer_id: str | None = None, *, orientation: list[float] | None = None, keyframe: str | int | None = None, driver: str = 'auto', tool_name: str | None = None, **kwargs: Any) -> Simulation | HardwareRobot | HardwareDriver
Create a robot - returns a Simulation or HardwareRobot instance.
This is a convenience factory, NOT a wrapper class. You get the real backend instance back - with full access to all its methods.
Defaults to simulation mode so that Robot("so100") never
accidentally sends commands to physical hardware. Use
mode="real" to explicitly opt into hardware control.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Robot name ("so100", "aloha", "unitree_g1", "panda", ...)
Accepts any alias defined in |
required |
mode
|
str
|
"sim" (default - safe), "real" (explicit hardware), or "auto" (probes USB for servo controllers, falls back to sim). Case-insensitive; surrounding whitespace ignored. |
'sim'
|
backend
|
str
|
Simulation backend name or alias, resolved through
|
'mujoco'
|
urdf_path
|
str | None
|
Explicit path to URDF/MJCF file. If not provided,
resolved via |
None
|
cameras
|
dict[str, dict[str, Any]] | None
|
Camera config for real hardware. Example:: {"wrist": {"type": "opencv", "index_or_path": "/dev/video0", "fps": 30}} Note: In |
None
|
position
|
list[float] | None
|
Robot base position in the sim world, |
None
|
data_config
|
str | None
|
Data configuration name for observation/action schema.
Honoured in both modes:
|
None
|
orientation
|
list[float] | None
|
Robot base orientation in the sim world as a quaternion
|
None
|
keyframe
|
str | int | None
|
Spawn the robot in a canonical pose declared by a
|
None
|
mesh
|
bool | None
|
Attach a Zenoh fleet-coordination mesh. |
None
|
peer_id
|
str | None
|
Optional mesh peer identifier. Auto-generated when omitted. |
None
|
driver
|
str
|
Which implementation drives the robot in |
'auto'
|
tool_name
|
str | None
|
The name the agent sees this robot under. Defaults to
|
None
|
**kwargs
|
Any
|
Forwarded to the underlying backend constructor. |
{}
|
Returns:
| Type | Description |
|---|---|
Simulation | Robot | HardwareDriver
|
|
Simulation | Robot | HardwareDriver
|
|
Simulation | Robot | HardwareDriver
|
Either one reports |
Simulation | Robot | HardwareDriver
|
methods take; |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
ImportError
|
If a known backend's optional dependency is missing
(e.g. |
RuntimeError
|
If the sim world or robot fails to initialize. |
Examples::
# Simulation (default - safe)
sim = Robot("so100")
# Explicit MJCF model path
sim = Robot("my_arm", urdf_path="path/to/robot.xml")
# Real hardware (explicit opt-in)
hw = Robot("so100", mode="real", cameras={...})
# Auto-detect (probes USB, falls back to sim)
robot = Robot("so100", mode="auto")
# The 5-line promise (defaults to sim - safe, no hardware needed)
from strands_robots import Robot
from strands import Agent
robot = Robot("so100") # mode="sim" (default)
agent = Agent(tools=[robot])
agent("Pick up the red cube")
The hardware robot¶
Returned by Robot(name, mode="real"). Also usable directly when you already hold a driver.
strands_robots.hardware_robot.Robot ¶
Robot(tool_name: str, robot: Robot | RobotConfig | str, cameras: dict[str, dict[str, Any]] | None = None, action_horizon: int = 8, data_config: str | Any | None = None, control_frequency: float = 50.0, ros2_bridge: bool = False, ros2_domain: int = 0, ros2_commands: bool = True, ros2_transport: str = 'rclpy', joint_limits: dict[str, tuple[float, float]] | None = None, dds_security_config: dict[str, str] | None = None, foxglove: bool | str = False, foxglove_mcap: str | PathLike[str] | None = None, foxglove_services: bool = False, **kwargs: Any)
Bases: TeleopMixin, AgentTool
Universal robot control with async task execution and status reporting.
Initialize Robot with async capabilities.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tool_name
|
str
|
Name for this robot tool |
required |
robot
|
Robot | RobotConfig | str
|
LeRobot Robot instance, RobotConfig, or robot type string |
required |
cameras
|
dict[str, dict[str, Any]] | None
|
Camera configuration dict:
{"wrist": {"type": "opencv", "index_or_path": "/dev/video0", "fps": 30}}
Each key names one camera and must be a bare token of letters,
digits, |
None
|
action_horizon
|
int
|
Actions consumed from each inferred policy chunk
before re-querying. Must be a positive integer - it is a lower
bound on the chunk slice the task loop applies
( |
8
|
data_config
|
str | Any | None
|
Data configuration (for GR00T compatibility) |
None
|
control_frequency
|
float
|
Control loop frequency in Hz (default: 50Hz).
Must be a positive finite number - it is the divisor of the
loop's per-action period ( |
50.0
|
ros2_bridge
|
bool
|
When True, publish this robot's live observation
( |
False
|
ros2_domain
|
int
|
ROS 2 domain id ( |
0
|
ros2_commands
|
bool
|
When True (default), the bridge also subscribes to
|
True
|
ros2_transport
|
str
|
Which ROS 2 backend the bridge uses:
|
'rclpy'
|
joint_limits
|
dict[str, tuple[float, float]] | None
|
Optional |
None
|
dds_security_config
|
dict[str, str] | None
|
Optional DDS Security credentials
( |
None
|
foxglove
|
bool | str
|
|
False
|
foxglove_mcap
|
str | PathLike[str] | None
|
Path of a new MCAP file the same channels are
recorded to. Requires |
None
|
foxglove_services
|
bool
|
When |
False
|
**kwargs
|
Any
|
Robot-specific parameters (port, etc.) |
{}
|
foxglove_url
property
¶
The live Foxglove WebSocket URL, or None when no Foxglove bridge runs.
foxglove_link
property
¶
A foxglove:// deep link to this robot's server, or None.
publish_ros_observation ¶
Read the robot's current observation once and publish it on ROS 2.
On-demand counterpart to the per-step publishing inside a running task:
lets an agent turn an idle, connected robot into a live ROS 2 device
without starting a control task. Requires ros2_bridge=True at
construction.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
skip_images
|
bool
|
When True, publish |
False
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
|
dict[str, Any]
|
|
dict[str, Any]
|
disabled - tools never raise past dispatch. |
start_task ¶
start_task(instruction: str, policy_port: int | None = None, policy_host: str = 'localhost', policy_provider: str = 'lerobot_local', duration: float = 30.0, **policy_kwargs: Any) -> dict[str, Any]
Start robot task asynchronously and return immediately.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
instruction
|
str
|
Natural-language instruction passed to the policy. |
required |
policy_port
|
int | None
|
Port of the policy server to query. Required when
|
None
|
policy_host
|
str
|
Host of the policy server. |
'localhost'
|
policy_provider
|
str
|
Provider name used to build the policy. |
'lerobot_local'
|
duration
|
float
|
Wall-clock budget in seconds. Must be positive and finite; it is validated before the task is submitted, so a budget the loop cannot honor is reported here instead of as a started task that commands nothing (or never ends). |
30.0
|
**policy_kwargs
|
Any
|
Checkpoint/provider keywords ( |
{}
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Tool-shaped result confirming the task started, or an error naming |
dict[str, Any]
|
the offending parameter. A second task is refused while another |
dict[str, Any]
|
rollout is in flight - including during its connect/policy-build |
dict[str, Any]
|
bring-up - because the arm has a single command bus. A robot that |
dict[str, Any]
|
|
dict[str, Any]
|
executor and bridges are gone, so a rollout started now would |
dict[str, Any]
|
command the arm zero times. |
run_policy ¶
run_policy(policy_object: Policy, instruction: str = '', duration: float = 30.0, n_steps: int | None = None) -> dict[str, Any]
Run a pre-built policy object on the real robot (blocking).
Hardware counterpart of Simulation.run_policy(policy_object=...):
drive a policy already constructed in-process (e.g. via
create_policy(...) around a local checkpoint) without standing up
a policy server on a port. Reuses the exact start_task control
loop - connect (with half-open rollback), state-key initialization,
the RTC control-frequency / observed-delay contract, and
resolve_chunk_length chunk consumption - so a pre-built object
and a server-backed provider behave identically on the wire.
Blocking: returns when duration elapses, after n_steps
applied actions, or when stop_task() is called from another
thread. For the server-backed provider path (and for fire-and-forget
execution) use start_task.
Exclusive: the arm has a single command bus, so this is refused while
another rollout is in flight - including while that rollout is still
connecting or building its policy, before it reports RUNNING.
Two rollouts admitted at once would interleave send_action writes
on one half-duplex bus and share the single task-state slot.
Terminal after shutdown: cleanup / stop release the task
executor, the mesh and the ROS bridges and cannot be undone, so a
rollout requested afterwards is refused rather than admitted to command
the arm zero times. A rollout already running when the shutdown lands is
reported STOPPED - a shutdown truncates a task exactly as
stop_task does, so its step count is a partial one.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
policy_object
|
Policy
|
A constructed |
required |
instruction
|
str
|
Natural-language instruction passed to the policy on
every |
''
|
duration
|
float
|
Wall-clock budget in seconds (same default as
|
30.0
|
n_steps
|
int | None
|
Optional cap on applied actions (mirrors the sim
|
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Tool-shaped result: a text summary plus a |
dict[str, Any]
|
carrying |
dict[str, Any]
|
/ |
dict[str, Any]
|
the rollout already in flight. |
get_task_status ¶
Report the task state, then the device it would drive.
The task machine used to be the whole answer: an arm whose port does
not exist on this host read Robot Status: IDLE, byte-identical to
a connected, healthy arm at rest, and the difference only surfaced as
a connect failure inside the first task. The measured facts were
already gathered by :meth:get_status for Python callers; the tool
action now prints the same facts under the task state and carries
them as a json block.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
|
dict[str, Any]
|
before), the device lines from :meth: |
dict[str, Any]
|
and a second content block holding those facts as JSON. |
stop_task ¶
Stop the current task, including one that is still connecting.
This is the interrupt an operator (or the fleet {"action": "stop"}
dispatch, via :class:~strands_robots.mesh.core.Mesh) reaches for, so
it has to hold for a
task in ANY stage that can still command the arm - not only the one
stage whose status happens to be RUNNING.
_execute_task_async sits in CONNECTING for the whole hardware
bring-up: a motors-bus handshake plus warmup_s per camera, seconds
on a real arm and longer on a multi-camera rig, followed by the policy
build. A stop pressed in that window used to be answered with
status="success" and "No task running to stop", and the arm then
moved anyway once the bring-up finished - the operator was told the
interrupt was handled while the rollout it was meant to cancel was
still pending.
A status write cannot express the request on its own, because
_execute_task_async writes RUNNING once bring-up completes and
that overwrites any STOPPED recorded before it. So the request is
latched in an event that is set here FIRST, before the status is even
read, and cleared only when a new task starts. The rollout honors the
latch at each stage boundary and in its loop condition.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A tool-shaped result confirming the stop, or - for a robot that is |
dict[str, Any]
|
genuinely idle or already in a terminal state - reporting that |
dict[str, Any]
|
there was nothing to stop. Both are |
dict[str, Any]
|
idle robot to stop is satisfied, not an error. |
stream
async
¶
stream(tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> AsyncGenerator[ToolResultEvent | ToolInterruptEvent, None]
Stream robot task execution with async actions.
cleanup ¶
Cleanup resources and stop any running tasks.
Terminal: this latches a shutdown, releases the task executor, tears
down the mesh and ROS bridges, and disconnects the robot -- the motors
bus and every camera -- so no device node stays held. The one exception
is a teleop loop that did not join: the devices are left open rather
than closed under a live writer, and the reason is recorded at ERROR
with the remedy. There is no
restart, so
run_policy / start_task / the execute action refuse
permanently afterwards rather than admit a rollout that would command
the arm zero times, and a rollout still in flight when this runs is
abandoned at its next stage check -- rather than finishing a bring-up
this teardown has already made pointless -- and reported STOPPED
rather than COMPLETED. Construct a new
Robot to run another task.
get_status
async
¶
Report the device's measured state plus this robot's task state.
Two fields answer different questions about cameras, and the pair is what makes an unhealthy device attributable:
camerasenumerates the camera names the device is configured for - which image streams exist at all.cameras_connectedmaps each live camera to its ownis_connectedreading - which of those streams are actually up.
is_connected on a lerobot arm is bus and all(cameras), so a
single dropped camera pulls it to False without naming which of the
N+1 facts fell. Reported beside the per-camera readings, a False
becomes attributable: a camera reading False is the culprit, and
every camera reading True leaves the motor bus as the one by
elimination.
Best-effort per camera, mirroring the observation path: a camera whose
is_connected probe raises is omitted from cameras_connected
rather than failing the whole probe, so a name present in cameras
and absent from cameras_connected is one whose state could not be
read. That same raise also reaches the arm's aggregate is_connected,
which reports None for it rather than degrading this probe to its
error shape - so the attribution survives the failure it exists to
attribute. A device exposing no live camera objects reports an empty map.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Status dict of measured facts. An unexpected failure anywhere in |
dict[str, Any]
|
the probe degrades to ``{"robot_name", "error", "is_connected": |
dict[str, Any]
|
False, "task_status": "error"}`` rather than propagating, so a |
dict[str, Any]
|
supervising agent reads a verdict instead of taking an exception. |
get_observation ¶
Read the arm once through lerobot: joints and camera frames.
The hardware half of the call mode="sim" answers, so a loop written
against the simulation keeps its shape when mode="real". The keys
are lerobot's ("<motor>.pos" plus one per camera), not the
simulation model's joint names. Connects on first use, as
:meth:send_action does, and reads under the bus lock the mesh shares.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
robot_name
|
str | None
|
Ignored (single robot). Present for parity with sim. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
The lerobot robot's observation dict. |
Raises:
| Type | Description |
|---|---|
Exception
|
Whatever the connect or the read raised, unchanged - a port that will not open is the caller's to see, not an empty observation. |
send_action ¶
Apply a single action to the hardware robot (TeleopMixin contract).
Synchronous so it can be driven from the :class:TeleopMixin teleop
loop thread. Ensures the underlying lerobot robot is connected, then
delegates through :func:strands_robots.bus_access.write_action, so a
teleop or ROS 2 command shares the bus with the mesh's readers instead of
racing them. robot_name is accepted
for parity with the multi-robot simulation host but ignored here - a
hardware Robot wraps exactly one device.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
action
|
dict[str, Any]
|
Flat |
required |
robot_name
|
str | None
|
Ignored (single robot). Present for mixin parity. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Status dict ( |
dict[str, Any]
|
errors without exceptions tearing down the hot loop. |
stop
async
¶
Stop the robot and release everything it holds.
Terminal, exactly as :meth:cleanup is -- this is the async spelling
of it, and it delegates every step rather than performing any itself.
The disconnect in particular belongs to that cleanup rather than ahead
of it. lerobot gates Robot.disconnect() on is_connected
(bus.is_connected and all(cam.is_connected ...)) and raises
DeviceNotConnectedError when it is false, so disconnecting here
first meant that stopping a robot which was never connected -- or one
left half-open by a failed bring-up -- raised before :meth:cleanup
was reached. Every terminal guarantee was then silently skipped: no
shutdown latch, a task executor still accepting work, and any device a
half-open connect had opened still held, with no entry point left that
would close it.
Ordering matters as much as reachability. :meth:cleanup closes the
devices last, once the teleop loop, the task executor, the mesh and
the ROS bridge are all down, because :meth:send_action re-opens the
robot lazily on a command that finds it disconnected. Disconnecting
here put that close ahead of every one of those command sources.
Runs off the event loop: :meth:cleanup joins the task executor and
closes a serial port, both of which block.
start_teleop_publish ¶
start_teleop_publish(teleoperator: Any, device_name: str = 'leader', method: str = 'arm', hz: float = 50.0) -> dict[str, Any]
Start publishing teleoperator actions to the mesh.
This makes the robot a teleop source: another peer on the mesh
can call start_teleop_receive(source_peer_id=self.peer_id) to
have its hardware follow along.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
teleoperator
|
Any
|
Any object with a callable |
required |
device_name
|
str
|
Name for this input stream (e.g. "leader", "gamepad"). |
'leader'
|
method
|
str
|
Input method label ("arm", "gamepad", "keyboard", "phone"). |
'arm'
|
hz
|
float
|
Publishing frequency in Hz. Must be a positive finite number;
the publish loop's period is |
50.0
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Status dict with topic and peer_id for the receiver to use, or an |
dict[str, Any]
|
error dict when the mesh is inactive, |
dict[str, Any]
|
polled, |
dict[str, Any]
|
not a rate the publish loop can honor. |
stop_teleop ¶
Stop all or a specific teleop publisher/receiver.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
device_name
|
str | None
|
If provided, stop only the named publisher/receiver. If None, stop all. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Stats from stopped sessions. |
strands_robots.hardware_robot.TaskStatus ¶
Bases: Enum
Robot task execution status
strands_robots.hardware_robot.RobotTaskState
dataclass
¶
RobotTaskState(status: TaskStatus = TaskStatus.IDLE, instruction: str = '', start_mono: float = 0.0, duration: float = 0.0, step_count: int = 0, error_message: str = '', task_future: Future | None = None, policy: Any = None)
Robot task execution state
Teleoperator¶
strands_robots.teleoperator.Teleoperator ¶
Create a teleoperator (input device) - returns a raw lerobot Teleoperator.
Convenience factory, NOT a wrapper. You get the real lerobot
Teleoperator instance back with full access to all its methods.
Unlike :func:Robot, a teleoperator has no "sim" mode - it is always a
real input device (or a mock). It is NOT connected here; call
.connect() (or attach it to a robot and call robot.teleoperate(),
which connects lazily).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Teleoperator type ("so101_leader", "so100_leader", "gamepad",
"keyboard", "phone", "koch_leader", ...). Any type registered
via lerobot's |
required |
id
|
str | None
|
Optional instance identifier. Namespaces the calibration file
(e.g. |
None
|
**kwargs
|
Any
|
Device-specific config forwarded to the resolved lerobot
|
{}
|
Returns:
| Type | Description |
|---|---|
Teleoperator
|
A connected-on-demand lerobot |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
TypeError
|
If lerobot returns a non-dataclass config class. |
Examples::
leader = Teleoperator("so101_leader", port="/dev/ttyACM1", id="blue")
pad = Teleoperator("gamepad", use_gripper=True)
kb = Teleoperator("keyboard")