Strands Robots

Control, simulate, and train robots with natural language

Strands Docs ◆ MuJoCo ◆ NVIDIA GR00T ◆ LeRobot ◆ Robots Sim ◆ Project Board

strands-robots gives a Strands Agent hands. One Robot() call returns a MuJoCo simulation (default - no GPU, no hardware) or a real robot - same code, same natural-language control, both auto-joined to a peer-to-peer mesh.

from strands import Agent
from strands_robots import Robot

robot = Robot("so100")              # MuJoCo sim by default; mode="real" for hardware
Agent(tools=[robot])("pick up the red cube")

One agent, the whole robotics loop

Teleoperate a real arm to collect demos, fine-tune a policy on them, run it in sim and on hardware, hand work to a fleet peer, and expose it all on ROS 2 - one library, one mental model. Every line below is a distinct capability:

from strands import Agent
from strands_robots import Robot
from strands_robots.tools import train_policy

# 1. TELEOPERATE a real SO-101 with its leader arm and RECORD demos as a
#    LeRobotDataset (one prompt drives cameras + teleop + recording).
follower = Robot("so101", mode="real", port="/dev/ttyACM0",
                 cameras={"front": {"type": "opencv", "index_or_path": "/dev/video0"}})
follower.attach_teleop("so101_leader", port="/dev/ttyACM1", id="leader")
Agent(tools=[follower])(
    "start_recording(repo_id='me/pick', root='/tmp/pick', fps=30, "
    "task='pick up the cube'); teleoperate for 60s; stop_recording"
)

# 2. POST-TUNE a policy on those demos (LoRA fine-tune; GPU box).
train_policy(action="train", provider="lerobot_local",
             dataset_root="/tmp/pick", base_model="lerobot/smolvla_base",
             output_dir="/tmp/pick_ckpt", method="lora", steps=20000)

# 3. RUN the tuned checkpoint - same policy on a MuJoCo twin AND the real arm.
twin = Robot("so101")                                              # sim twin, no hardware
twin.run_policy(robot_name="so101", policy_provider="lerobot_local",
                policy_config={"pretrained_name_or_path": "/tmp/pick_ckpt"}, duration=10.0)
follower.start_task("pick up the cube", policy_provider="lerobot_local",
                    policy_port=None, duration=10.0)               # real arm, in-process

# 4. COORDINATE a fleet - tell a mesh peer to assist, in natural language.
follower.mesh.tell(follower.mesh.peers[0]["peer_id"], "hold the tray steady")

# 5. EXPOSE the running sim on ROS 2 - rviz / nav2 / any ros2 node can subscribe.
from strands_robots.simulation import Simulation
sim = Simulation(ros2_bridge=True); sim.create_world(); sim.add_robot("so101")
sim.step(100)   # publishes /so101/joint_states + camera image_raw on the ROS 2 graph

Steps 1 and 3-real need hardware; step 2 needs a GPU. Everything runs in sim with no hardware (Robot("so101")), so you can exercise the whole loop today.

Why strands-robots

• Sim-first, safe by default. Robot("so100") spins up a MuJoCo world. You never accidentally drive real servos - mode="real" is an explicit opt-in.
• 70+ robots, 8 categories. Arms, humanoids, quadrupeds, hands, drones, bimanual rigs - resolved from a single registry with auto-download of assets.
• Any policy. VLA models (NVIDIA GR00T, LeRobot ACT/Pi0/SmolVLA/Diffusion), plus classical motion planners, MPC, and scripted controllers behind one ABC.
• Mesh networking built in. Every robot is a Zenoh peer. tell() another robot what to do; broadcast an E-STOP; bridge to AWS IoT Core for fleets.
• 67-action simulation tool. World building, physics, rendering, domain randomization, procedural terrain (create_world(terrain="rough"|"stairs"|"pyramid"|"slope") for locomotion), and LeRobotDataset recording - all agent-callable.
• ROS 2 interop. Observe + command any ROS 2 graph (use_ros), act as a robot with no rclpy (use_rtps), or expose a running sim as a ROS node.
• One mental model. Sim and hardware share the same policy interface, the same mesh, and the same natural-language control surface.

How it works

graph LR
    A[Natural Language<br/>'Pick up the red block'] --> B[Strands Agent]
    B --> C[Robot<br/>sim or real]
    C --> D[Policy Provider<br/>GR00T / Cosmos 3 / LeRobot / planner / mock]
    D --> E[Action Chunk]
    E --> F[MuJoCo Sim<br/>or Hardware]
    F -->|observation| C

    classDef input fill:#2ea44f,stroke:#1b7735,color:#fff
    classDef agent fill:#0969da,stroke:#044289,color:#fff
    classDef policy fill:#8250df,stroke:#5a32a3,color:#fff
    classDef hardware fill:#bf8700,stroke:#875e00,color:#fff

    class A input
    class B,C agent
    class D,E policy
    class F hardware

Installation

Examples use uv (curl -LsSf https://astral.sh/uv/install.sh | sh); plain pip works too.

uv pip install strands-robots

The base install is light (numpy, opencv-headless, Pillow). Pull in only the extras you need:

# Most users start here:
uv pip install "strands-robots[sim-mujoco]"

# Real hardware + local policies:
uv pip install "strands-robots[sim-mujoco,lerobot]"

# MolmoAct2 VLA (transformers-native; resolves from PyPI via lerobot >= 0.6):
uv pip install "strands-robots[molmoact2]"

# Everything:
uv pip install "strands-robots[all]"

The Isaac Sim GPU backend is a built-in, in-tree peer of mujoco and newton (it lives at strands_robots.simulation.isaac). Its pip-installable helpers ship in the sim-isaac extra, but the Isaac Sim runtime itself (~30 GB) is provisioned separately - via its own pip wheels on Python 3.12 (pip install 'isaacsim[all,extscache]==6.0.*' --extra-index-url https://pypi.nvidia.com, with coverage/EULA caveats documented in docs/simulation/isaac.md), the Omniverse Launcher, Isaac Lab, or the NGC docker image. Install the helpers with pip install 'strands-robots[sim-isaac]', then select the backend with create_simulation("isaac") - see Simulation (MuJoCo) and docs/simulation/isaac.md.

From source:

git clone https://github.com/strands-labs/robots
cd robots
uv pip install -e ".[all,dev]"

Quick starts

Simulation (no GPU, no hardware)

from strands import Agent
from strands_robots import Robot

robot = Robot("so100") # MuJoCo simulation
agent = Agent(tools=[robot])
agent("Wave the arm using the mock policy for 200 steps, then render a top-down view")

Robot("so100") returns a Simulation instance - the full 67-action simulation AgentTool. Drive it in natural language through an Agent, call its methods directly (robot.render(camera_name="topdown")), or dispatch an action by calling it (robot(action="render", camera_name="topdown")). See Simulation.

Note: Robot("so100") already creates the world and adds the robot for you. Do not call create_world() again on the returned instance - it will error with "World already exists." The create_world() / add_robot() sequence shown in Simulation (MuJoCo) is for the low-level Simulation(...) constructor, which starts empty.

Real hardware + GR00T

from strands import Agent
from strands_robots import Robot, gr00t_inference

robot = Robot(
    "so101",
    mode="real",
    cameras={
        "front": {"type": "opencv", "index_or_path": "/dev/video0", "fps": 30},
        "wrist": {"type": "opencv", "index_or_path": "/dev/video2", "fps": 30},
    },
    port="/dev/ttyACM0",
    data_config="so100_dualcam",
)

agent = Agent(tools=[robot, gr00t_inference])

# Start the GR00T inference service (Docker, Jetson/x86 GPU)
agent.tool.gr00t_inference(
    action="start",
    checkpoint_path="/data/checkpoints/model",
    port=8000,
    data_config="so100_dualcam",
)

agent("Use so101 to pick up the red block with the GR00T policy on port 8000")

Local LeRobot policy (no inference server)

from strands_robots import create_policy

# Direct HuggingFace inference - ACT, Pi0, SmolVLA, Diffusion, ...
policy = create_policy("lerobot/act_aloha_sim_transfer_cube_human")

Teleoperation (leader arms, gamepads, WASD)

Drive any real robot - or a simulation - from one or more LeRobot teleoperators. Teleoperator() mirrors the Robot() factory; attach_teleop() + teleoperate() run the control loop.

from strands_robots import Robot, Teleoperator

# Leader arm -> follower arm (both speak {motor}.pos -> zero config)
follower = Robot("so101", mode="real", port="/dev/ttyACM0")
follower.attach_teleop("so101_leader", port="/dev/ttyACM1", id="leader")
follower.teleoperate()                       # Ctrl+C or stop_teleoperate()

# Earth Rover Mini+ with WASD keys (velocity keys -> zero config)
rover = Robot("earthrover_mini_plus", mode="real", robot_ip="192.168.1.151")
rover.attach_teleop("keyboard_rover")        # W/A/S/D
rover.teleoperate(block=True, duration=30)

# Cross-vocabulary or sim teleop -> supply a map_fn(action) -> action
robot.attach_teleop("keyboard_ee", map_fn=my_ik)   # EE deltas -> joint .pos
robot.teleoperate(publish=True)              # also stream over the mesh

17 teleoperators (so100/so101/koch/omx/openarm leaders, bi_* leaders, gamepad, keyboard, keyboard_ee, keyboard_rover, phone, reachy2_teleoperator, unitree_g1, homunculus arm/glove) drive 14 robots. Zero-config when action keys match; otherwise pass map_fn. Full matrix + recipes: Teleoperation docs.

Recording & streaming datasets

The physical-AI data loop, end to end: record a LeRobotDataset from sim or hardware, stream it straight back for eval/training (no full download), and optionally dump it to a mutable Hugging Face Storage Bucket. Needs the lerobot extra (which bundles datasets + av + torchcodec).

from strands import Agent
from strands_robots import Robot

sim = Robot("so100", mesh=False)
agent = Agent(tools=[sim])

# 1. COLLECT — one natural-language prompt drives scene + cameras + policy + record.
agent(
    "Create a world with the so100 robot, add a red cube and a front camera, "
    "start recording (repo_id='local/demo', root='/tmp/demo', fps=30, "
    "overwrite=True, task='pick up the red cube'), run the mock policy for "
    "60 steps, then stop recording."
)

# 2. STREAM — read it back lazily; camera frames decode on the fly from the MP4
#    shards, state/action from parquet. Nothing is re-materialized to disk.
reader = sim.stream_dataset("local/demo", root="/tmp/demo", shuffle=False)
for frame in reader:
    frame["observation.images.front"]   # (3, H, W) tensor, decoded from video
    frame["observation.state"]          # joint vector
    frame["action"]
    break

stream_dataset() is the in-process read counterpart to start_recording/stop_recording. For full training, the upstream trainer uses the same engine — lerobot-train --policy.type=act --dataset.repo_id=... --dataset.streaming=true --num_workers=4 (the lerobot-train entry point wraps python -m lerobot.scripts.lerobot_train; flags are draccus --dotted.key=value form).

Verify episode integrity. A recording's ground truth is the parquet under meta/episodes/, not the count a model narrates while collecting. Collect episodes with a deterministic Python loop (one run_policy(..., n_episodes=1) plus save_episode() per episode) rather than trusting a model to count its own tool calls, then confirm the dataset holds the episodes you intended - in-process or from the shell:

sim.verify_dataset_episodes(expected=20)   # reads parquet; status="error" on a mega-episode

# exit 0 = pass, 1 = fail, so it drops straight into CI as a dataset gate
strands-robots verify-dataset /tmp/demo --expected 20

This catches the "mega-episode" corruption class - a run that buffered every frame into one episode_index=0 episode while reporting 20/20 - plus meta/info.json vs parquet drift and zero-length episodes.

Dump to a Storage Bucket during collection (mutable, Xet-deduplicated — the Phase 1/2 collection target that avoids git-LFS history bloat) with one kwarg:

sim.stop_recording(bucket="your-org/robot-fave")   # → hf://buckets/your-org/robot-fave/demo

Requires the hf CLI with the buckets/sync subcommands (pip install -U "huggingface_hub>=1.5" + hf auth login — those subcommands first ship in 1.5.0; every earlier release, including 1.0–1.4.x, installs an hf entry point without them).

The bucket read side (stream_dataset(..., repo_type="bucket")) needs strands-robots >= 0.5.1, so upgrade with -U rather than a bare pip install "strands-robots[...]" — pip reports Requirement already satisfied against a pre-existing older release and upgrades nothing, and on 0.4.1 the read raises TypeError: open() got an unexpected keyword argument 'repo_type'.

Any on-disk dataset directory can be synced (or daily re-synced) without a live recording session — one recorded earlier in the process, or on hardware via lerobot-record:

from strands_robots import sync_dataset_to_bucket

sync_dataset_to_bucket("/tmp/demo", "your-org/robot-fave")
# → {"status": "success", "bucket_uri": "hf://buckets/your-org/robot-fave/demo"}

run_id defaults to the directory name; pass run_id="nightly" to choose the bucket subpath, and delete=True for mirror semantics.

Proprio-only / no video (e.g. edge devices without a torchcodec wheel): sim.stream_dataset(repo_id, drop_videos=True, delta_timestamps={...}) streams state/action only and never touches the video decoder. drop_videos=True requires a delta_timestamps with at least one non-video key (e.g. {"observation.state": [0.0], "action": [0.0]}) - without one, every feature including video would stream, so the call raises ValueError instead of silently no-opping.

macOS note (zero-touch). torchcodec links ffmpeg via @rpath, and Homebrew's ffmpeg (/opt/homebrew/lib) is not on the default dyld search path — so video decode would normally fail with Library not loaded: @rpath/libavutil.NN.dylib. On import strands_robots we auto-detect this and put Homebrew's ffmpeg on DYLD_FALLBACK_LIBRARY_PATH (re-exec'ing the interpreter once for a plain script run; never inside Jupyter/REPL/pytest, where it just prints the one-line export to run). It's a no-op off macOS, without torchcodec, or when the var is already set. Disable with STRANDS_ROBOTS_NO_DYLD_SHIM=1. See examples/06_agent_collect_and_stream.py.

See also Recording & datasets for the DatasetRecorder direct API and append/resume workflow.

The Robot() factory

Robot() is a factory, not a wrapper - you get the real backend instance back with all its methods.

Robot("so100")                       # mode="sim"  (default, safe)
Robot("so100", mode="real")          # explicit hardware opt-in
Robot("so100", mode="auto")          # probe USB for servos, fall back to sim
Robot("my_arm", urdf_path="arm.xml") # bring your own MJCF/URDF

Safety/validation rules: - Defaults to sim. Real hardware is always an explicit mode="real". - cameras= is rejected in sim mode - add sim cameras via the add_camera action after creation. - Unknown robot names raise ValueError unless you pass urdf_path=. - STRANDS_ROBOT_MODE overrides detection; a typo'd value logs a warning and falls back to sim.

Supported robots

70+ robots across 8 categories, resolved from registry/robots.json. Assets (MJCF + meshes) auto-download from robot_descriptions / MuJoCo Menagerie on first use. List them at runtime with from strands_robots import list_robots; list_robots().

Hardware-capable (drivable with mode="real" via LeRobot): so100, so101, koch, omx, hope_jr, aloha, bi_openarm, reachy2, unitree_g1, lekiwi, earthrover. All are simulatable.

Adding a robot

There are two paths, depending on whether the robot needs project-specific metadata:

1. Standard robot_descriptions robot (zero config). Any MJCF robot shipped by robot_descriptions resolves automatically without a robots.json entry - the asset is discovered and downloaded on first use:

```python from strands_robots import Robot, list_discoverable

sim = Robot("iiwa14") # discovered, not in robots.json print(list_discoverable()) # the MJCF long tail you can load directly ```

A curated robots.json entry always wins over discovery, so overriding a discovered robot later is non-breaking.

1. Custom or metadata-rich robot. If the robot needs a non-default joint count, hardware port, aliases, scene tweaks, or local mesh overrides, add a curated entry. For a robot that belongs in the shipped catalog, add it to registry/robots.json and open a PR. For a machine-local robot, register it at runtime instead of editing the package:

```python from strands_robots.registry import register_robot

register_robot(name="my_arm", model_xml="my_arm.xml", asset_dir="~/robots/my_arm", joints=7, category="arm") ```

Tools reference

Import any of these and pass to Agent(tools=[...]). Each is a Strands AgentTool returning {"status", "content"}.

agent.tool.gr00t_inference(
    action="start",
    checkpoint_path="/data/checkpoints/model",
    port=8000,
    use_tensorrt=True,
    vit_dtype="fp8",     # ViT:  fp16 | fp8
    llm_dtype="nvfp4",   # LLM:  fp16 | nvfp4 | fp8
    dit_dtype="fp8",     # DiT:  fp16 | fp8
)

Policy providers

All policies implement one ABC - async get_actions(observation, instruction, **kwargs). The interface is deliberately agnostic about how actions are produced, so it fits both VLA models and classical controllers.

from strands_robots import create_policy

create_policy("mock")                                  # sinusoidal test actions
create_policy("groot", port=5555)                      # NVIDIA GR00T via ZMQ
create_policy("zmq://localhost:5555")                  # same, by URL
create_policy("cosmos3", embodiment="droid", port=8000)  # NVIDIA Cosmos 3 via WebSocket
create_policy("lerobot/act_aloha_sim_transfer_cube")   # local HF inference

classDiagram
    class Policy {
        <<abstract>>
        +get_actions(obs, instruction, **kwargs)
        +set_robot_state_keys(keys)
        +requires_images
        +reset(seed)
        +provider_name
    }
    class Gr00tPolicy
    class Cosmos3Policy
    class LerobotLocalPolicy
    class MockPolicy
    class YourPolicy
    Policy <|-- Gr00tPolicy
    Policy <|-- Cosmos3Policy
    Policy <|-- LerobotLocalPolicy
    Policy <|-- MockPolicy
    Policy <|-- YourPolicy

Security: lerobot_local loads HuggingFace models with trust_remote_code=True (arbitrary code execution). You must opt in with export STRANDS_TRUST_REMOTE_CODE=1. Only load models you trust.

Cosmos 3 (NVIDIA omnimodal VLA - service mode)

nvidia/Cosmos3-Nano-Policy-DROID via a self-contained WebSocket client (cosmos3 / c3 / cosmos3://host:port); no openpi-client dep, no numpy<2 pin, so it composes with lerobot in one env.

uv sync --all-extras --group=cu130-train --group=policy-server
python -m cosmos_framework.scripts.action_policy_server_robolab \
    --checkpoint-path nvidia/Cosmos3-Nano-Policy-DROID --port 8000
curl http://localhost:8000/healthz   # -> 200 when ready (~4 min cold)

uv pip install -e '.[sim-mujoco]'
uv pip install 'strands-robots[cosmos3-service]'

from strands_robots.policies import create_policy

policy = create_policy("cosmos3", embodiment="droid", port=8000)
policy.set_robot_state_keys([f"joint_{i}" for i in range(7)] + ["gripper"])
chunk = policy.get_actions_sync(observation, "pick up the cube")
# chunk == [{"joint_0": .., ..., "gripper": ..}, ...]  (one dict per timestep)

policy = create_policy(
    "cosmos3", embodiment="droid", port=8000,
    observation_mapping={
        "wrist":     "observation/wrist_image_left",
        "exterior":  "observation/exterior_image_1_left",
        "exterior2": "observation/exterior_image_2_left",
    },
)

MUJOCO_GL=egl python examples/vla/cosmos3_sim_rollout.py --record /tmp/c3.mp4

Non-VLA policies (motion planners, MPC, scripted)

The same interface fits cuRobo, MoveIt2, OMPL, MPC, and pure-IK / scripted trajectories - anything mapping (observation, goal) to joint targets. Non-VLA providers set requires_images = False (skip camera rendering) and read their goal from well-known **kwargs keys instead of parsing the instruction string:

Providers MUST ignore unknown **kwargs rather than raising, so callers can pass shared keys across providers without coupling to a backend.

from typing import Any
from strands_robots.policies import Policy, register_policy, create_policy


class ReachPolicy(Policy):
    """Linear interpolation from current joint state to target_joints."""

    def __init__(self, steps: int = 32, **_: Any) -> None:
        self._keys: list[str] = []
        self._steps = steps

    @property
    def provider_name(self) -> str:
        return "reach"

    @property
    def requires_images(self) -> bool:
        return False  # joint-state only -- skip camera rendering

    def set_robot_state_keys(self, robot_state_keys: list[str]) -> None:
        self._keys = list(robot_state_keys)

    async def get_actions(self, observation_dict, instruction, **kwargs):
        target = kwargs.get("target_joints")
        if target is None:
            raise ValueError("ReachPolicy requires target_joints kwarg")
        state = observation_dict.get("observation.state", [0.0] * len(self._keys))
        out = []
        for s in range(1, self._steps + 1):
            alpha = s / self._steps
            out.append({k: (1 - alpha) * state[i] + alpha * target[k]
                        for i, k in enumerate(self._keys)})
        return out


register_policy("reach", lambda: ReachPolicy, aliases=["lerp"])
policy = create_policy("reach")

from strands_robots.policies import create_policy

# Collision-aware planning (GPU, in-process); plan is cached, streamed per tick.
policy = create_policy("curobo", robot_config="franka.yml", action_horizon=16)
actions = policy.get_actions_sync(
    {"observation.state": [0.0, -0.79, 0.0, -2.36, 0.0, 1.57, 0.79]},
    "reach for the red block",                  # ignored by planners
    target_pose=[0.5, 0.0, 0.4, 1.0, 0.0, 0.0, 0.0],
)

Training providers

create_trainer is the training-side peer of create_policy: the provider name is the same one used for inference, so create_policy("groot") and create_trainer("groot") refer to one provider. It returns a Trainer you drive with a TrainSpec, and the train_policy tool wraps this same factory for agent-callable post-tuning.

from strands_robots.training import create_trainer, TrainSpec

# Imitation / post-tuning on a recorded LeRobotDataset:
trainer = create_trainer("lerobot_local")
trainer.train(TrainSpec(dataset_root="/tmp/pick", base_model="lerobot/smolvla_base",
                        output_dir="/tmp/pick_ckpt", method="lora", steps=20000))

# From-scratch reinforcement learning in sim (no dataset):
ppo = create_trainer("ppo")   # or create_trainer("fast_sac")

The RL trainers (ppo, fast_sac) subclass BaseRLAlgo and share the same validate -> prepare -> train -> export lifecycle as the imitation trainers. They collect trajectories through VecSimEnv (N independent SimEnv as one batched env) and score with BaseRLAlgo.evaluate(). The training package stays torch-free until an RL provider is resolved on first use.

Simulation (MuJoCo)

Robot("so100") (sim mode) returns a Simulation - a MuJoCo-backed AgentTool exposing 67 actions for world composition, physics, rendering, policy execution, and dataset recording. Build it directly when you want full control:

from strands_robots.simulation import Simulation

sim = Simulation(tool_name="sim", mesh=False)
sim.create_world()
sim.add_robot(name="arm", data_config="so100")
sim.add_object(name="cube", shape="box", position=[0.3, 0, 0.05])
sim.add_camera(name="topdown", position=[0, 0, 1.5], target=[0, 0, 0])

# Wrist camera: mount ON the gripper body so it tracks the arm like the real
# SO101/SO100 hardware cam. position/target are in the body's LOCAL frame.
# Body names are namespaced "<robot>/<body>" (e.g. "arm/gripper").
sim.add_camera(name="wrist", position=[0, -0.05, 0], target=[0, -0.15, 0],
               parent_body="arm/gripper")

sim.run_policy(robot_name="arm", policy_provider="mock", n_steps=200,
               control_frequency=50.0)

frame = sim.render(camera_name="topdown")   # {status, content:[text, image]}

Self-healing: unknown parameters are rejected with "Unknown parameter X for action Y. Valid: [...]", missing required params produce "Action X requires parameter Y.", and vectors/dtypes are validated before MuJoCo sees them - so the agent learns the contract without crashing the process.

Third-party backends. create_simulation(name) discovers backends beyond the built-in mujoco/newton/isaac registry via Python entry points. A sibling package registers its SimEngine subclasses under the strands_robots.backends group in its pyproject.toml, and they become available on pip install without patching this package:

[project.entry-points."strands_robots.backends"]
my_engine = "my_pkg.backend:MyEngine"

Built-in backends always take precedence over plugins of the same name, plugin discovery is lazy (it never slows cold import), and list_backends() returns the merged builtin + plugin set. Requesting a known-but-uninstalled plugin backend (e.g. create_simulation("isaac") without the plugin) raises a ValueError carrying the exact install hint. See docs/simulation/isaac.md for the Isaac Sim backend's install, usage, config, and STRANDS_ISAAC_* env vars.

Mesh networking

Every Robot() and Simulation() is automatically a peer on a local Zenoh mesh - no setup. Peers on the same host discover each other out of the box (gossip scouting plus a shared local endpoint), sharing a single ref-counted zenoh.Session per process. Cross-host discovery is deliberately explicit: point peers at each other with ZENOH_CONNECT (e.g. tcp/10.0.0.1:7447). Multicast scouting is off by default - it lets any device on the LAN enumerate and attract the fleet - and is opt-in via STRANDS_MESH_MULTICAST=true, which logs a loud warning.

from strands_robots import Robot

a = Robot("so100")              # auto-joins the mesh
b = Robot("so100")              # second peer (another process)
print(a.mesh.peers)             # list[dict] - discovers b
print(a.mesh.peers_by_id[b.peer_id])   # dict[peer_id -> info] for O(1) lookup
info = a.mesh.get_peer(b.peer_id)      # None-safe single lookup

a.mesh.tell(b.peer_id, "pick up the cube")
a.mesh.emergency_stop()         # broadcast E-STOP, audited to disk

tell() routes to hardware and sim peers. Each payload is forwarded to the sink that reads it: constructor extras (model_path, server_address, ...) via policy_config, and the per-call goal (target_pose, target_joints, world_update) via policy_kwargs, which the runner hands to every get_actions() call. So a planner-style policy on a sim peer sees the goal payload it needs:

a.mesh.tell(
    b.peer_id,
    "reach for the red block",
    policy_provider="curobo",
    target_pose=[0.3, 0.0, 0.4, 1.0, 0.0, 0.0, 0.0],
    robot_name="arm_left",      # disambiguate in multi-robot sims
    duration=10.0,
)

Expose the mesh to an agent with the robot_mesh tool (peers, status, tell, send, broadcast, stop, emergency_stop, subscribe, watch, inbox). Disable globally with STRANDS_MESH=false or per-robot with Robot("so100", mesh=False). Install with uv pip install "strands-robots[mesh]".

For frictionless single-machine experiments, set STRANDS_MESH_LOCAL_DEV=1 - one env var that runs the mesh without mTLS/ACL on localhost. It defaults the auth mode to none and satisfies the insecure-acknowledgement second factor by itself, so you don't also need STRANDS_MESH_I_KNOW_THIS_IS_INSECURE=1. An explicit STRANDS_MESH_AUTH_MODE=mtls still wins. Never set STRANDS_MESH_LOCAL_DEV on a shared or production network.

AWS IoT Core transport (fleets)

For robots across networks, bridge the mesh to AWS IoT Core over MQTT5/mTLS, with Device Shadow mirroring, S3 camera offload, and account-wide Fleet Provisioning. Hardened with CA pinning, strict thing-name validation, deny-by-default IoT policy scoping, and a safety audit log. Install with uv pip install "strands-robots[mesh-iot]". See the Configuration matrix for the STRANDS_MESH_* knobs.

ROS 2 interoperability

strands-robots speaks ROS 2 from four complementary angles - a Strands agent can observe, command, be, and expose a ROS 2 system. Full guide: ROS 2 Integration / docs/ros2-integration.md.

A Strands agent (Claude Opus via Amazon Bedrock) given the use_ros tool drives a real ROS 2 turtlesim in a closed-loop square - reading pose, correcting heading, re-driving - over 43 in-process tool calls. Runnable: examples/ros2/use_ros/.

# Observe + command a live ROS 2 graph, in plain English:
from strands import Agent
from strands_robots.tools import use_ros
Agent(tools=[use_ros])("list the topics, drive /turtle1 forward, confirm the pose changed")

# Or expose a simulation as a ROS 2 node any tool can subscribe to:
from strands_robots.simulation import Simulation
sim = Simulation(ros2_bridge=True)
sim.create_world(); sim.add_robot("so101")
sim.step(10)   # publishes /so101/joint_states + camera image_raw on the ROS 2 domain

rclpy ships with a sourced ROS 2 distro (not on PyPI). The [ros2] extra adds only the pip-installable cyclonedds binding that use_rtps uses - so the pure-RTPS path needs no ROS install at all. Every surface degrades to a clear, structured error when its backend is unavailable; the default install never touches ROS 2.

Configuration

Environment variables

Asset cache

~/.strands_robots/
└── assets/           # auto-downloaded MJCF + meshes
    ├── trs_so_arm100/
    ├── franka_emika_panda/
    └── ...

Clear with rm -rf ~/.strands_robots/assets/; relocate with export STRANDS_ASSETS_DIR=/path/to/dir.

Benchmarks

strands-robots ships a LIBERO benchmark integration on the MuJoCo backend - byte-equivalent to upstream LIBERO at the model level, reaching success_rate >= 0.92 on libero-10/SCENE5. Register declarative benchmarks from file and evaluate policies via the list_benchmarks, register_benchmark_from_file, and evaluate_benchmark simulation actions. Install with uv pip install "strands-robots[benchmark-libero]".

Project structure

strands_robots/
├── __init__.py            # Lazy-loaded public API (Robot, Simulation, policies)
├── robot.py               # Robot() factory (sim/real/auto dispatch)
├── hardware_robot.py      # HardwareRobot - async LeRobot control
├── policies/
│   ├── base.py            # Policy ABC
│   ├── factory.py         # create_policy() + runtime registration
│   ├── mock.py            # MockPolicy (non-VLA reference)
│   ├── groot/             # NVIDIA GR00T (ZMQ/HTTP client + data configs)
│   └── lerobot_local/     # Direct HuggingFace inference (RTC, processors)
├── registry/              # robots.json (70+) + policies.json + loaders
├── simulation/
│   ├── base.py            # SimEngine ABC
│   ├── factory.py         # create_simulation() + backend registry
│   ├── models.py          # SimWorld / SimRobot / SimObject / SimCamera
│   └── mujoco/            # MuJoCo backend (67-action AgentTool)
├── rendering/             # Hybrid rendering: CameraParams, backgrounds (panorama/3DGS),
│                          #   HybridCompositor, encode_clip / mjpeg_frames
├── mesh/                  # Zenoh mesh: core, sensors, input, audit, transport, iot
├── benchmarks/libero/     # LIBERO suite + BDDL parser + adapter
└── tools/                 # gr00t_inference, lerobot_*, pose, serial, robot_mesh

Development

uv pip install -e ".[all,dev]"

hatch run test          # unit tests
hatch run test-integ    # integration tests (GPU + model weights)
hatch run lint          # ruff check + format --check + mypy
hatch run format        # ruff check --fix + ruff format

Python 3.12+ required. See AGENTS.md for conventions and the accumulated code-review learnings.

Security

Found a vulnerability? Do not open a public issue. Follow the disclosure process in SECURITY.md (AWS VDP / HackerOne).

Note the trust_remote_code gate on lerobot_local (see Policy providers) and the mesh CA-pinning / thing-name validation controls in the Configuration matrix.

Contributing

Issues and PRs welcome. Track work on the Strands Labs - Robots project board; it is the source of truth for roadmap and follow-ups.

• GitHub Issues
• Pull Requests

License

Apache-2.0 - see LICENSE.

Links