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

# Consumer side: evaluate a card's health indicators from the streams (no transport needed)
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.

Security: identity, roles, audit (spec 8)

ROSP connections use 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 grants each platform a role from an access policy. rosp-sdk ships the certificate tooling and the identity/role/audit primitives; the live session and WebSocket transport that drive them are a separate component.

Issue a local CA and certificates with the CLI (pip install 'rosp-sdk[cli,certs]'):

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.security import AccessPolicy, JsonlAuditLog, TLSConfig

tls = TLSConfig.from_dir("certs", "robot-acme-amr-007")   # SSLContext for the transport layer
access = AccessPolicy.load("certs/access.json")           # map a peer's cert identity to a role
audit = JsonlAuditLog("audit.jsonl")                      # append-only record of who did what

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 refused. These primitives are consumed by the ROSP session/transport, which lives outside this package.

Conformance

A conformance harness exercises any running adapter against 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. Because it needs the live session and transport, it ships separately from this package.

Spec

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

License

Apache 2.0