MoveIt2¶

MoveIt2Policy is a thin ZMQ + msgpack client for a sidecar ROS 2 node running moveit_py. Like cuRobo it is a non-VLA, collision-aware motion planner: it reads its goal from **kwargs (target_pose / target_joints), ignores camera frames (requires_images = False), and never parses the instruction string for control.

Unlike cuRobo's in-process CUDA library, MoveIt2 runs out-of-process: the ROS 2 stack and moveit_py live entirely in a sidecar, so the Python venv running strands_robots stays free of ROS 2 deps. The only client-side requirements are pyzmq + msgpack (the [moveit2] extra).

Install¶

pip install 'strands-robots[moveit2]'   # client side: pyzmq + msgpack only

The ROS 2 / moveit_py deps stay out of pyproject.toml - they live in the sidecar. Bring the sidecar up via the docker-compose recipe or natively:

# Option 1 - docker compose (pinned, reproducible)
cd strands_robots/policies/moveit2/server
docker compose up

# Option 2 - native ROS 2 (dev loop)
source /opt/ros/jazzy/setup.bash         # or your distro
pip install pyzmq msgpack                # the only non-ROS deps
python -m strands_robots.policies.moveit2.server.zmq_node \
    --port 5556 --planning-group arm

See policies/moveit2/server/README.md for the sidecar deployment and forking guidance.

Quickstart¶

from strands_robots.policies import create_policy

policy = create_policy(
    "moveit2",                     # alias: "moveit"
    host="127.0.0.1",
    port=5556,
    planning_group="arm",
)

actions = policy.get_actions_sync(
    observation_dict={"observation.state": [0.0] * 6},
    instruction="reach for the red block",   # ignored by planners
    target_pose=[0.3, 0.0, 0.4, 1.0, 0.0, 0.0, 0.0],   # [x,y,z, qw,qx,qy,qz]
)

The smart-string resolver also works: create_policy("zmq://127.0.0.1:5556", planning_group="arm").

Parameters¶

MoveIt2Policy(
    host="127.0.0.1",      # sidecar hostname (loopback only by default)
    port=5556,             # sidecar port
    planning_group="arm",  # default MoveIt2 planning group; per-call override allowed
    timeout_ms=15000,      # ZMQ send + recv timeout in milliseconds
    api_token=None,        # included in every request; falls back to MOVEIT2_API_TOKEN
)

api_token falls back to the MOVEIT2_API_TOKEN environment variable when not passed. The client emits a plaintext-over-TCP warning when host is a non- loopback address (the token travels unencrypted; terminate TLS at a proxy or keep the sidecar on loopback).

Goal kwargs¶

MoveIt2Policy shares the non-VLA goal vocabulary with the rest of the planner family (see cuRobo), so a goal can flow across providers without coupling to a backend:

Key	Type	Meaning
`target_pose`	`list[float]`	Cartesian goal `[x, y, z, qw, qx, qy, qz]` in the base frame
`target_joints`	`dict[str, float]`	Joint-space goal keyed by joint name (rad / m)
`world_update`	`dict \\| None`	Per-call collision-scene refresh, forwarded to the sidecar
`planning_group`	`str`	Override the policy's default planning group for this call

Provide at least one of target_pose / target_joints. If both are set, target_joints wins (mirrors MoveIt2's setJointValueTarget). When neither is given, get_actions(...) raises ValueError rather than returning a zero-action.

Input validation¶

Goals are validated up-front before they reach the wire (defence-in-depth for LLM-agent inputs):

target_pose must be exactly 7 finite floats (NaN / inf rejected).
target_joints keys must match ^[A-Za-z][A-Za-z0-9_-]*$ - the same allowlist mesh.security.validate_command applies, so a value the mesh accepts flows end-to-end without a second mismatch. Values must be finite.
planning_group must be a plain identifier (no shell metacharacters or path traversal).

Wire protocol¶

request  = {"endpoint": "plan",
            "data": {"joint_state": list[float] | None,
                     "planning_group": str,
                     "target_pose": [x, y, z, qw, qx, qy, qz] | None,
                     "target_joints": dict[str, float] | None,
                     "world_update": dict | None}}
response = {"trajectory": list[list[float]],
            "success": bool,
            "status": str}

Trajectory rows are [time_from_start_seconds, q0, q1, ..., qN]. The client drops the leading time column when packing per-step action dicts and maps the remaining columns onto the keys from set_robot_state_keys(...). A success=False response raises RuntimeError. The sidecar also exposes ping (health check) and reset (per-episode hook).

In simulation¶

from strands_robots import Robot

sim = Robot("panda")              # sim-by-default
sim.run_policy(
    robot_name="panda",
    instruction="",               # ignored by the planner
    policy_provider="moveit2",
    policy_config={"host": "127.0.0.1", "port": 5556, "planning_group": "arm"},
    target_pose=[0.3, 0.0, 0.4, 1.0, 0.0, 0.0, 0.0],
    duration=10.0,
    control_frequency=50.0,
)

The LLM-agent demo path (Robot.start_task(..., policy_provider="moveit2", target_pose=[...])) flows the same target_pose / target_joints kwargs through start_task's **policy_kwargs, so agents share one goal vocabulary across VLA and planner providers.