Strands Robots

Control, simulate, and train robots with natural language

Strands Docs ◆ MuJoCo ◆ NVIDIA GR00T ◆ LeRobot ◆ 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, and the same opt-in peer-to-peer mesh (mesh=True).

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"}},
                 mesh=True)   # step 4 needs the mesh; joining is opt-in
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:

   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.

2. 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:

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

   model_xml and scene_xml name a file inside asset_dir - that is how every reader joins them back - so a value that leaves it (absolute, or ../) is refused at registration rather than validated against a file the loader will not open.

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.

SageMaker managed training jobs

The sagemaker provider is transport, not behavior: it submits the same TrainSpec as one managed training job and waits for the terminal verdict. The training logic lives in the container image you point it at, which packages one of the local trainer paths (the groot and cosmos3 providers are directly containerizable). Install with pip install "strands-robots[sagemaker]" (boto3 only; validate works without it).

from strands_robots.training import create_trainer, TrainSpec

trainer = create_trainer(
    "sagemaker",
    image_uri="123456789012.dkr.ecr.us-east-1.amazonaws.com/strands-trainer:latest",
    role_arn="arn:aws:iam::123456789012:role/StrandsSageMakerTraining",
    instance_type="ml.g5.xlarge",
)
spec = TrainSpec(
    dataset_root="s3://my-bucket/datasets/pick",   # -> the "training" input channel
    output_dir="s3://my-bucket/checkpoints/pick",  # -> the job's S3OutputPath
    base_model="lerobot/smolvla_base",
    steps=20000,
    extra={"policy_type": "act"},                  # -> string hyperparameters
)
result = trainer.train(spec)   # validates locally BEFORE submitting
# result.checkpoint_dir -> s3://.../<job name>/output/model.tar.gz on success;
# a failed job is an error result naming the job + FailureReason.

dataset_root and output_dir must be s3:// URIs; every other spec field plus extra travels as string hyperparameters (strings verbatim, everything else JSON-encoded) that the container's entry point decodes back into a TrainSpec with dataset_root=/opt/ml/input/data/training and output_dir=/opt/ml/model. spec.num_nodes maps onto the job's InstanceCount. A job that outlives the local poll budget reports running and stays pollable via trainer.status(job_name).

Required IAM surface - the caller (the identity running train) needs:

• sagemaker:CreateTrainingJob, sagemaker:DescribeTrainingJob
• iam:PassRole on the execution role passed as role_arn

and the execution role itself needs S3 read on the dataset prefix, S3 write on the output prefix, ECR pull on the image, and CloudWatch Logs write (AmazonSageMakerFullAccess is a superset, but the above is the minimal set).

The integration smoke (tests_integ/training/test_sagemaker_smoke.py) skips without AWS credentials; its full submission half additionally reads STRANDS_SAGEMAKER_SMOKE_IMAGE_URI, STRANDS_SAGEMAKER_SMOKE_ROLE_ARN, STRANDS_SAGEMAKER_SMOKE_S3_PREFIX (and optionally STRANDS_SAGEMAKER_SMOKE_INSTANCE_TYPE) - test-only variables, not read by the library.

Simulation (MuJoCo)

Robot("so100") (sim mode) returns a Simulation - a MuJoCo-backed AgentTool exposing 77 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.", a field the schema publishes as a string is refused unless it is one ("Action X: 'Y' must be a string, got 7 (int)"), and vectors/dtypes are validated before MuJoCo sees them - so the agent learns the contract without crashing the process. Every one of those refusals names the field by the spelling you sent, or by the one this schema publishes: the router rewrites a few wire names to their method parameter (torque_vec -> torque) before validating, and a refusal naming the rewritten name would point at a field the schema does not carry.

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

Robot("so100", mesh=True) joins a local Zenoh mesh. Joining is opt-in: a bare Robot() leaves robot.mesh as None unless STRANDS_MESH is true/1/yes. Once joined, peers on the same host find each other with nothing further to configure (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", mesh=True)   # joining is opt-in
b = Robot("so100", mesh=True)   # 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). Opt in per robot with Robot("so100", mesh=True) or process-wide with STRANDS_MESH=true; STRANDS_MESH=false is a hard kill switch that overrides an explicit mesh=True, and Robot("so100", mesh=False) opts a single robot out. 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.

CA Pin Rotation Runbook

The AWS IoT transport pins the SHA-256 of the canonical Amazon Root CA1 PEM, so a network-level attacker (DNS hijack, captive portal, BGP, malicious local proxy) cannot substitute a rogue CA at the download URL. The accepted set is a collection, not a scalar, so old and new pins can both be valid at once - that is what makes a rotation expressible without a flag-day deploy.

When AWS rotates the root, every fleet member refuses the new certificate until a pin covering it is accepted, on both the download path and the on-disk re-use path. Deleting the cached PEM does not help: the re-download fetches the same unpinned bytes and is refused again. Rotation therefore needs an ordered procedure, which is this one.

Recompute the pin of whatever the URL currently serves:

python -c "import hashlib, urllib.request as u; \
print(hashlib.sha256(u.urlopen( \
'https://www.amazontrust.com/repository/AmazonRootCA1.pem' \
).read()).hexdigest())"

Monitor for rotations before they bite: AWS announces root-CA changes in its security bulletins with a deprecation timeline, so a planned rotation can be shipped ahead of the cutover rather than during an outage.

Rotate (planned):

1. Verify the new certificate out of band. A digest computed from the same connection that served the bytes proves nothing. Confirm the certificate against an independent source before it becomes a pin.
2. Ship a release that adds the new pin and keeps the old one. Both stay valid, so peers still on the previous release keep verifying.
3. Wait for fleet uptake. The overlap is bounded by the slowest fleet member, not by the release cadence.
4. Drop the old pin in a follow-up release once uptake is complete.

Emergency (a rotation lands faster than a release can ship): stage the verified new pin in STRANDS_MESH_CA_PINS (comma-separated, 64-char lowercase hex). It is additive - the built-in pin stays accepted and verification stays on - so it buys the grace period a release would have provided. Entries that are not valid hex digests are rejected with a warning and skipped rather than weakening the set. Remove the override once the release carrying the pin is deployed.

STRANDS_MESH_DISABLE_CA_PIN is not part of this procedure. It turns the download-path pin check off rather than widening it, accepts whatever the URL serves, and marks the result as unverified-origin so later runs warn about re-using it. It is a break-glass for a broken pin, never the response to a rotation - a rotation has a verified pin to stage.

Benchmarks

Register declarative benchmarks from file and evaluate policies via the list_benchmarks, register_benchmark_from_file, and evaluate_benchmark simulation actions. strands_robots/simulation/builtin_benchmarks.py ships the canonical locomotion suites; a task-specific suite is a JSON file, not a vendored adapter.

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
└── 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 Rotating the pinned Amazon Root CA1 has an ordered procedure: CA Pin Rotation Runbook.

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