Pure-RTPS ROS 2 integration¶

ROS 2 runs over DDS, and DDS speaks RTPS on the wire. use_rtps lets a strands agent join a ROS 2 graph as a first-class DDS participant using only the pip-installable cyclonedds binding - no rclpy, no sourced ROS 2 distro, no ros2 CLI. Because RTPS is stable across ROS 2 distros, one implementation interoperates with Humble, Jazzy, Rolling, and beyond.

use_ros vs use_rtps¶

	`use_ros`	`use_rtps`
Role	client / observer	participant / robot
Backend	in-process `rclpy`	`cyclonedds` (pip wheel)
Needs sourced ROS 2	yes	no
Type coverage	any installed interface	curated IDL bundle
Runs on macOS / CI bare	no (needs ROS)	yes

Use use_ros when you have ROS 2 sourced and need full type coverage or services. Use use_rtps when you want zero-install interop or to act as a robot - publishing topics a real ROS 2 stack (rviz, nav2, a teleop node) will consume, indistinguishable from hardware on the wire.

pip install 'strands-robots[ros2]'   # cyclonedds - a self-contained wheel

Actions¶

Action	Required args	Returns
`status`	-	Whether the cyclonedds backend is available
`types`	-	The ROS 2 message types in the local IDL bundle
`advertise`	`topic`, `type`	Creates a publisher (appear on the graph)
`publish`	`topic`, `type`	Publishes N messages built from `fields`
`subscribe`	`topic`, `type`	Creates a subscription
`echo`	`topic`, `type`	Returns the next N samples as JSON

Scope (v1) is topics only; services and actions need the ROS 2 request/reply- over-DDS protocol and are a focused follow-up.

Type coverage¶

To publish a message you must own its type definition locally, so use_rtps ships a curated IDL bundle (strands_robots.rtps.idl) of the common ROS 2 messages, registered under their ROS 2 type strings: the geometry_msgs primitives (Twist/Pose/...) plus the sensor_msgs JointState and Image (with their std_msgs/Header + builtin_interfaces/Time chain) that the rclpy-free hardware bridge publishes. List them with use_rtps(action="types"). Arbitrary custom messages are out of scope until cyclonedds-python's dynamic (XTypes) support matures - use use_ros (rclpy) for those.

ROS 2 names are mangled to their DDS form automatically: a topic /turtle1/cmd_vel becomes rt/turtle1/cmd_vel, and a type geometry_msgs/msg/Twist becomes geometry_msgs::msg::dds_::Twist_ - the conventions that make a bare DDS participant interoperable with real ROS 2 nodes.

Examples¶

from strands_robots.tools import use_rtps

use_rtps(action="status")
use_rtps(action="types")

# Act as a robot: advertise then drive a cmd_vel topic a real node consumes.
use_rtps(action="advertise", topic="/turtle1/cmd_vel", type="geometry_msgs/msg/Twist")
use_rtps(action="publish", topic="/turtle1/cmd_vel",
         type="geometry_msgs/msg/Twist",
         fields={"linear": {"x": 2.0}, "angular": {"z": 1.5}},
         count=15, rate=10.0)

RtpsRobot: a ROS 2 robot over pure RTPS¶

RtpsRobot is the pure-RTPS sibling of RosBridgedRobot. It forwards to use_rtps, so it drives a ROS 2 mobile base with nothing but a pip wheel - and because it publishes real DDS samples, it can act as the robot itself.

from strands import Agent
from strands_robots.mesh import RtpsRobot

turtle = RtpsRobot.from_rtps(
    node_name="turtlesim",
    cmd_vel_topic="/turtle1/cmd_vel",
)

turtle.advertise()                       # appear on the graph
turtle.drive(linear=1.0, duration=1.5)   # publish Twist over RTPS for 1.5 s
turtle.stop()

agent = Agent(tools=turtle.tools)        # drive_turtlesim, stop_turtlesim
agent("drive forward for two seconds")

See examples/ros2/rtps_turtle_demo.py for an end-to-end script, and tests_integ/tools/test_use_rtps_live.py for the gated live test that drives a real turtlesim from a bare participant (RTPS_LIVE=1 pytest -m rtps).

For a fully reproducible, self-contained cross-process proof (real turtlesim node + our publisher, one command), see examples/ros2/rtps_proof/:

cd examples/ros2/rtps_proof
docker compose run --build --rm proof   # exits 0 iff the turtle moved

Hardware bridge over pure RTPS (no rclpy)¶

Robot(ros2_bridge=True) defaults to the rclpy backend (ros2_transport="rclpy", full sensor_msgs fidelity, needs a sourced ROS 2 distro). Pass ros2_transport="rtps" to run the same bridge over pure cyclonedds instead - a single pip wheel, no rclpy and no sourced distro:

from strands_robots import Robot

# rclpy-free: publishes /so101/joint_states (+ camera image_raw) and subscribes
# /so101/joint_command -> send_action, all over cyclonedds RTPS.
arm = Robot("so101", mode="real", ros2_bridge=True, ros2_transport="rtps")

The two transports emit byte-identical topics, so a real ROS 2 node (or ros2 topic echo / ros2 topic pub) cannot tell them apart on the wire:

ros2 topic echo /so101/joint_states     # decodes the cyclonedds-published JointState
ros2 topic pub --once /so101/joint_command sensor_msgs/msg/JointState \
  '{name: ["shoulder_pan.pos"], position: [0.1]}'   # drives the arm

The trade-off is the same as use_rtps: type coverage is bounded by the IDL bundle (joint_states + image_raw are in; anything else needs the rclpy backend). The bridge is implemented by strands_robots.hardware_rtps_bridge.HardwareRtpsBridge, the rclpy-free sibling of HardwareRosBridge. Both derive from strands_robots.ros_telemetry.RosTelemetryBase, which owns the topic names and the inbound joint_command parsing, so the two transports are byte-identical on the wire by construction; they present the identical publish_joint_states / publish_image / inbound-joint_command surface.

Securing the inbound command surface¶

The inbound /<robot>/joint_command subscription lets any participant on the DDS domain drive the physical arm. Two layers harden it, both threaded through the Robot() constructor.

DDS Security gate (RTPS only)¶

When the command surface is enabled (ros2_bridge=True, ros2_commands=True, ros2_transport="rtps"), HardwareRtpsBridge refuses to start unless one of the following is true:

a dds_security_config dict is supplied, or
the operator sets STRANDS_ROS2_BRIDGE_I_KNOW_THIS_IS_INSECURE=1 (truthy: 1 / true / yes) to explicitly accept an unsecured graph.

A telemetry-only bridge (ros2_commands=False) is publish-only and is not gated. dds_security_config requires the following keys (each a path or a file: / data: URI per the OMG DDS-Security spec); permissions_ca is optional:

from strands_robots import Robot

arm = Robot(
    "so101",
    mode="real",
    ros2_bridge=True,
    ros2_transport="rtps",
    dds_security_config={
        "identity_ca":  "file:/etc/dds/identity_ca.pem",   # identity CA
        "certificate":  "file:/etc/dds/participant.pem",   # participant cert
        "private_key":  "file:/etc/dds/participant_key.pem",
        "governance":   "file:/etc/dds/governance.p7s",    # signed governance
        "permissions":  "file:/etc/dds/permissions.p7s",   # signed permissions
        # "permissions_ca": "file:/etc/dds/permissions_ca.pem",  # optional
    },
)

The credentials are wired into the cyclonedds DomainParticipant QoS together with the builtin DDS-Security plugins, so both the outbound telemetry and the inbound command surface ride an authenticated, access-controlled graph. A half-filled config is rejected at construction.

dds_security_config is RTPS-specific: passing it with ros2_transport="rclpy" raises, because the rclpy backend gets its DDS Security from the ROS 2 RMW keystore/env (ROS_SECURITY_* / sros2), not from a config dict.

Joint position bounds¶

joint_limits={motor: (min, max)} (threaded into either transport) range-checks every inbound command. If any commanded joint falls outside its declared range, the entire command is rejected - never partially applied - so one out-of-range joint can never drive part of the arm while the rest holds. Joints without a declared bound are unconstrained; to leave a joint unbounded, omit it rather than declaring an infinite bound. Every bound must be a finite number - a non-finite one declares a range that admits nothing, and the bridge refuses it at construction rather than dropping every command for that joint mid-run.

arm = Robot(
    "so101",
    mode="real",
    ros2_bridge=True,
    ros2_transport="rtps",
    dds_security_config={...},
    joint_limits={"shoulder_pan.pos": (-3.14, 3.14), "elbow.pos": (-1.57, 1.57)},
)

Safety¶

Agent-supplied topic and type names are validated against an allowlist before mangling (absolute /-rooted alnum/_ names; pkg/msg/Name types). The tool never constructs a shell command or generates source, so there is no command-injection or eval surface. Backend, type-resolution, and field errors are returned as structured {"status": "error"} results rather than raised.

The numeric options are checked in the same place, ahead of the backend probe, so a refusal happens before a writer joins the graph and reports identically whether or not cyclonedds is installed. count (publish, echo) must be a positive integer, and rate (publish) and timeout (echo) must be positive finite numbers - the same accepted domain use_ros enforces, so a value publishable through one transport is publishable through the other. An option the requested action never reads is not second-guessed.