rosp-sdk

Python SDK for the Robot Open Specification Protocol (ROSP) — base classes, data models, plugin registry, and tools for building robot adapters.

Install

Pick the transports your project actually needs; the base install stays lean.

# Base install — protocol models + ROSPAdapter base class + registry.
# Small, no transport-specific deps.
pip install rosp-sdk

# With one or more transport adapters (see "Transports" below).
pip install rosp-sdk[ros2]        # + rosp-adapter-ros2  (needs system ROS 2)
pip install rosp-sdk[grpc]        # + rosp-adapter-grpc  (Spot, Viam, custom)
pip install rosp-sdk[vda5050]     # + rosp-adapter-vda5050 (MQTT AGVs)
pip install rosp-sdk[all]         # every adapter above

# With CLI tools:
pip install rosp-sdk[cli]

The plugin model

Adapters are ordinary Python packages that register themselves under the rosp.adapters entry-point group. rosp-sdk discovers them at runtime via importlib.metadata — no imports until you ask for one, no heavyweight transitive dependencies unless a transport is installed.

from rosp_sdk import available_adapters, load_adapter

available_adapters()
# {'ros2': 'rosp_adapter_ros2:ROS2Adapter',
#  'grpc': 'rosp_adapter_grpc:GRPCAdapter',
#  'vda5050': 'rosp_adapter_vda5050.rosp_adapter:VDA5050ROSPAdapter'}

Adapter = load_adapter("ros2")           # returns the class
adapter = Adapter(config={"robot_id": "urn:rosp:robot:..."})
async with adapter:
    card = await adapter.describe()
    async for msg in adapter.stream(["/sensor/lidar/scan"]):
        print(msg.data)

If an adapter isn't installed, load_adapter raises ROSPAdapterNotInstalled with the exact pip install command needed to fix it.

Transports

Quick Start

Build an Adapter

from rosp_sdk import ROSPAdapter, RobotCard, StreamMessage, StreamQoS

class MyRobotAdapter(ROSPAdapter):
    async def describe(self, depth="full") -> RobotCard:
        return RobotCard(
            rosp_version="0.1",
            id="urn:rosp:robot:mycompany:myrobot:001",
            identity={"manufacturer": "MyCompany", "model": "MyRobot", "type": "mobile_robot"},
            hardware={"weight_kg": 25.0, "payload_kg": 10.0},
            sensors=[{"sensor_id": "lidar", "type": "lidar_2d"}],
            actuators=[{"actuator_id": "drive", "type": "wheel"}],
            capabilities=[{"capability_id": "navigation", "type": "navigation"}],
            integration={"supported_protocols": [{"protocol": "ros2"}]},
            safety={"e_stop_available": True},
        )

    async def stream(self, topics, qos=None):
        # Subscribe to robot sensors and yield StreamMessage objects
        ...

    async def discover(self):
        # Return robot identity for network discovery
        ...

    async def health_check(self):
        # Return adapter and robot health
        ...

    async def connect(self):
        # Connect to robot's native interface
        ...

    async def disconnect(self):
        # Clean shutdown
        ...

Validate a Robot Card

from rosp_sdk import validate_robot_card

errors = validate_robot_card(card_dict)
if not errors:
    print("Valid!")

CLI

rosp validate my-robot.json        # Validate against schema
rosp export-card my-robot.json     # Pretty-print
rosp inspect my-robot.json         # Show sensors, actuators, integration
rosp version                       # Show SDK and spec version

New here? Connect a robot walks you from your robot's API to a conformant ROSP robot in about 30 minutes (tested: our suite follows it literally).

ROSP 0.3: components, schema catalog, events, health

Spec: docs/spec/ROSP-v0.3-stream-contract.md. Every stream names the part it belongs to, carries its payload definition, and parts declare what "unhealthy" means; the robot pushes events and health.

from rosp_sdk import Event, HealthStatus, AdapterStatus, RobotStatus, streams

# In describe(): streams name their component and a catalog schema (29 shipped: pose3d, twist,
# geolocation, battery, esc_report, joint_report, flight_state, imu_status, gps_status, ...).
card = RobotCard(..., actuators=[{"actuator_id": "rotor-4", "type": "propeller", "health": {"indicators": [
            {"id": "rpm_deviation", "field": "rpm", "compare": "siblings_pct",
             "warn": {"op": "gt", "value": 8}, "critical": {"op": "gt", "value": 15}, "for_s": 2}]}}],
       components=[{"component_id": "battery", "type": "battery"}],
       streams=[{"stream": "actuators/rotor-4/esc", "schema": "esc_report", "component_id": "rotor-4", "rate_hz": 2},
                {"stream": "robot/battery", "schema": "battery", "component_id": "battery", "rate_hz": 1}])

# Anywhere in the adapter: report what happened. Platforms that negotiated "event" get it at once.
self.emit(Event("motor.failure", "critical", "ESC 4 failure flags 0x04", component_id="rotor-4", data={"flags": 4}))

# health_check() is called by the session (every 10 s, and sent immediately when it changes).
HealthStatus(AdapterStatus.HEALTHY, RobotStatus.ERROR, latency_ms=12, uptime_seconds=300,
             components={"rotor-4": {"status": "error", "message": "ESC failure flags 0x04"}})

# Platform side
session = await ws.connect_to_adapter(url, platform_id="brain", on_event=print, on_health=print)
from rosp_sdk.health import HealthEvaluator      # the card's indicators, evaluated from the streams
ev = HealthEvaluator(card)
for change in ev.observe("actuators/rotor-4/esc", {"rpm": 5400}):
    print(change.message())                       # "rotor-4: rpm deviation from siblings 9.2% (ok -> warning)"

rosp_sdk.streams gives consumers the catalog (catalog(), schema_def(), fields()), the parts of a card in one shape (components()), the resolved stream declarations (resolve_card()), stream naming (stream_name()) and payload validation. validate_robot_card() now treats unknown part types as warnings (check_robot_card() returns both errors and warnings) and reports dangling component or indicator references as errors. rosp check tests event-kind streams, payloads against their definitions, and the health push.

Security: mutual TLS, roles, audit (spec 8)

Every ROSP connection uses mutual TLS 1.3. Each side's certificate carries its identity as a SAN URI — rosp://robot/<robot-id> or rosp://platform/<platform-id> — and the robot gives each platform a role from an access policy. Set it up with the CLI (pip install 'rosp-sdk[cli,certs,ws]'):

rosp certs init --dir certs                          # local CA: ca.crt (+ ca.key, keep private)
rosp certs issue robot acme-amr-007 --dir certs      # robot-id as derived from the card (spec 4.2)
rosp certs issue platform my-fleet-manager --dir certs
echo '{"platform/my-fleet-manager": "controller", "*": "observer"}' > certs/access.json

from rosp_sdk import ws
from rosp_sdk.security import AccessPolicy, JsonlAuditLog, TLSConfig

# Robot side (here dialing out to the platform; serve_adapter(..., ssl=tls.server()) to listen instead)
tls = TLSConfig.from_dir("certs", "robot-acme-amr-007")
await ws.run_reverse(adapter, "wss://platform.example:8791/rosp/adapter", ssl=tls.client(),
                     access=AccessPolicy.load("certs/access.json"), audit=JsonlAuditLog("audit.jsonl"))

# Platform side
tls = TLSConfig.from_dir("certs", "platform-my-fleet-manager")
await ws.serve_platform(on_session, port=8791, ssl=tls.server(), platform_id="my-fleet-manager")

A platform the policy doesn't list, or whose platform_id differs from its certificate, is refused (4001); a robot announcing a robot-id other than its certificate's is dropped by the platform. Plain ws:// needs insecure=True and logs a warning — local development only.

Check your adapter: rosp check

Point the conformance kit at any running adapter. It acts as a platform and uses only what the robot declares (its Robot Card): handshake and TLS identity, card validity, the robot-id rule, every declared stream (data matches its schema), refusals (unknown topic 2006, unknown action 3003, bad params 2005, expired 5002) and the command lifecycle.

rosp check wss://robot:8765/rosp --certs certs --as rosp-check        # the robot listens
rosp check --serve 8792 --certs certs --as rosp-check                 # the robot dials in
rosp check ... --motion --watchdog       # also command it — it WILL move a few metres

--motion adds goto lifecycle, cancel, preemption (aircraft take off to 2 m first and land after); --watchdog silences the platform to check the robot ends the command with 5005 and applies its link-loss behaviour. The platform you run it as (--as) needs a role on the robot (access policy). Exit code 0 means conformant; --json for CI.

Spec

• ROSP v0.1 Specification
• Adapter SDK Specification
• Getting Started Guide

License

Apache 2.0