Subsystem SDK API Guide

The subsystem SDK API is the subsystem-author API for Python and C++.

Use this page when writing code that subclasses or wraps a Librux subsystem. Do not use this API from browser clients or outside-runtime tools; those clients should use the Host Control REST API or Host Control WebSocket API.

This page explains the subsystem-author flow. For the table-oriented public class and method catalog, see Subsystem SDK Reference.

Scope

The SDK API owns the following.

subsystem construction and lifecycle hooks
Event publish/subscribe registration and publishing
Control endpoint registration and timed exchange calls
Procedure registration/calls
Operation registration/start/cancel/query
typed payload refs and endpoint timing policy

API Contract, component, and message definitions live under Spec. This page shows how subsystem code uses those contracts.

Construction

Use the role-specific subclass that matches subsystem.role in subsystem.yaml.

Role	Python/C++ subclass
`gateway`	`GatewaySubsystem`
`component`	`ComponentSubsystem`
`compound`	`CompoundSubsystem`
`app`	`AppSubsystem`

Subsystem remains the common base class. In Python, the role-specific classes check the loaded manifest role during construction.

Python example.

ComponentSubsystem(
    name="controller",
    log_level="INFO",
    user_data_path="./userdata",
    manifest_path="./subsystem.yaml",
    heartbeat_port=40000,
)

C++ example.

librux::wrapper::SubsystemOptions options;
options.name = "controller";
options.log_level = "INFO";
options.user_data_path = "./userdata";
options.manifest_path = "./subsystem.yaml";
options.heartbeat_port = 40000;

MyComponentSubsystem subsystem(std::move(options));

When a subsystem is started with librux launch run or lbx deploy up, the package launcher also sets LIBRUX_SUBSYSTEM_MANIFEST and LIBRUX_BINDINGS. That lets C++ subsystems report the same manifest, interfaces, and bindings status shape as Python subsystems.

Lifecycle

A subsystem implements required startup hooks and may implement runtime lifecycle hooks. Runtime lifecycle commands are requests; user code does not set the subsystem state directly.

Python example.

from librux.wrapper import ComponentSubsystem


class MySubsystem(ComponentSubsystem):
    def on_initialize(self) -> bool:
        return True

    def on_start(self) -> bool:
        return True

    def on_terminate(self) -> bool:
        return True

    def on_reset(self) -> bool:
        return True

    def on_pause(self) -> bool:
        return True

    def on_resume(self) -> bool:
        return True

    def on_fault(self, reason: str | None = None) -> bool:
        return True

C++ example.

class MySubsystem final : public librux::wrapper::ComponentSubsystem {
protected:
    bool on_initialize() override { return true; }
    bool on_start() override { return true; }
    bool on_terminate() override { return true; }
    bool on_reset() override { return true; }
    bool on_pause() override { return true; }
    bool on_resume() override { return true; }
    bool on_fault(const std::string& reason) override {
        (void)reason;
        return true;
    }
};

Canonical subsystem lifecycle states are documented in Subsystem Lifecycle. The common operator commands are these.

lbx subsystem pause <subsystem>
lbx subsystem resume <subsystem>
lbx subsystem reset <subsystem>
lbx subsystem fault <subsystem> --reason "fault text"

Event API

For Event methods, topic means the Event channel key.

Python methods.

register_event_publish(...)
register_event_subscribe(...)
publish_event(...)
start_event_transport()
stop_event_transport()

C++ methods use the same conceptual operations with EventPublishSpec, EventSubscribeSpec, and EventPayload.

register_event_publish accepts an optional Event queue_depth in Python, and EventPublishSpec::queue_depth in C++. A value of 0 or omission uses the SDK default heuristic. Use this for Event streaming/telemetry workloads where short bursts should be absorbed before samples are dropped.

For the method catalog, see Subsystem SDK Reference.

Control API

Control can be raw, typed, or typed/timed. Use the control method names in subsystem code. Low-level wire protocol details are reserved for internal protocol documentation.

The SDK argument is still named topic for wire compatibility. In user-facing terms, treat this value as the Control endpoint key, not as an Event channel.

Python example.

target = self.binding_target("gateway")
response = self.transact_control(
    target=target,
    topic="timed.request",
    type="bytes",
    value={"value": 1},
    request_type="msg.core.common.primitive.v1/Int32Value",
    response_type="msg.core.common.primitive.v1/Int32Value",
    timing={"execute_after_ns": 500_000},
    period_ms=0.0,
    deadline_ms=1000.0,
)

For the method catalog, see Subsystem SDK Reference.

C++ example.

librux::wrapper::ControlTransactOptions options;
options.request_type = "msg.core.common.primitive.v1/Int32Value";
options.response_type = "msg.core.common.primitive.v1/Int32Value";
options.timing = librux::wrapper::ControlTimingOptions{};
options.timing->execute_after_ns = 500'000;

const auto response = transact_control(
    "gateway",
    "timed.request",
    "{\"value\":1}",
    options
);

Procedure API

Basic procedure calls remain available. Typed/timed calls add request and response message refs.

Python example.

response = self.call_bound_procedure(
    binding="io_gateway",
    procedure="set_output",
    params={"index": 1, "value": True},
    request_type="msg.core.common.primitive.v1/IndexedBoolValue",
    response_type="msg.core.common.primitive.v1/BoolValue",
    timing={"execute_after_ns": 500_000},
)

C++ example.

librux::wrapper::ProcedureCallOptions options;
options.request_type = "msg.core.common.primitive.v1/IndexedBoolValue";
options.response_type = "msg.core.common.primitive.v1/BoolValue";
options.timing = librux::wrapper::ProcedureTimingOptions{};
options.timing->execute_after_ns = 500'000;

auto response = call_procedure(
    "io_gateway",
    "set_output",
    "{\"index\":1,\"value\":true}",
    options
);

For the method catalog, see Subsystem SDK Reference.

Operation API

Operation uses the control exchange path for start/cancel/query and Event for state/feedback/result.

Python example.

operation = self.start_bound_operation(
    binding="arm",
    operation="move_j",
    params={"positions": [0.0, 0.1, 0.2]},
    request_type="msg.motion.articulated.v1/JointWayPoint",
    timing={"execute_after_ns": 1_000_000},
)

C++ example.

librux::wrapper::OperationStartOptions options;
options.request_type = "msg.motion.articulated.v1/JointWayPoint";
options.timing = librux::wrapper::OperationTimingOptions{};
options.timing->execute_after_ns = 1'000'000;

auto response = start_operation("arm", "move_j", "{...}", options);

For the method catalog, see Subsystem SDK Reference.

Endpoint Timing Policy

Endpoint registration accepts the following.

timing_support = "none"
timing_support = "optional"
timing_support = "required"

This applies to Control endpoints, Procedures, and Operations.

For the option fields used by Python and C++, see Subsystem SDK Reference.

Managed Resource Handles

Packages launched through Librux receive LIBRUX_RESOURCE_LEASE_ID and can ask librux-resourced for approved file descriptors. Hardware-facing handles use the normal broker helpers such as open_udp_socket, open_tcp_socket, open_can_socket, and open_device.

Managed package frontend backends use a narrower helper. If package.yaml declares frontend.context.api or frontend.context.ws, launch resolves that context into a loopback listener grant. The package process opens that listener through the broker.

from librux.core import ResourceBrokerClient

api_listener = ResourceBrokerClient().open_frontend_tcp_listener(kind="api").socket
ws_listener = ResourceBrokerClient().open_frontend_tcp_listener(kind="ws").socket

The browser still connects to the control backend same-origin proxy. The package backend serves the accepted listener socket; it must not bind a separate frontend port directly under brokered network sandboxing.

For resource helper methods, see Subsystem SDK Reference.