RB Router Python SDK

The SDK connects robot fleets and customer-controlled training infrastructure to the RB Router control plane. It intentionally does not replace RLlib, TorchRL, Isaac Lab, Slurm, Kubernetes, SageMaker, or Vertex AI. It gives those systems a versioned experience, execution, evaluation, and policy-deployment contract.

Install from this repository

python -m pip install ./sdk/python

npm run sdk:build creates the installable wheel under sdk/python/dist/. The Python SDK GitHub workflow tests and builds the package on every SDK change, attaches the wheel as a CI artifact, and publishes to PyPI through OIDC Trusted Publishing only when a GitHub Release is published and the repository's pypi environment has been authorized. No long-lived PyPI token is stored in the repository.

Upload canonical robot experience

from rbrouter import Client, EdgeAgent, Episode, Transition, TransitionShard

client = Client("https://rbrouter.com", api_key="rbr_...")
edge = EdgeAgent(client, fleet_id="fleet_...", robot_id="robot_...")

episode = Episode(
    episode_id="insertion-000184",
    skill="connector_insertion",
    policy_version="policy-v18",
    embodiment_id="franka_panda",
    observation_schema_version="franka.obs.v2",
    action_schema_version="eef.delta_pose.v1",
    reward_schema_version="connector.reward.v3",
    control_frequency_hz=20,
    started_at="2026-08-24T20:10:00Z",
    outcome="failure",
    transitions=[
        Transition(
            step=0,
            timestamp_ns=0,
            observation={"joint_position": [0.0] * 7},
            action={"delta_position": [0.0, 0.0, 0.01]},
            reward=-1.0,
            cost=0.2,
            terminated=True,
            truncated=False,
        )
    ],
    failure={"type": "lateral_misalignment", "taskStage": "insert", "severity": "high"},
)
edge.upload([episode])

For large episodes, use TransitionShard.write_jsonl() and upload the immutable shard to R2/S3/GCS/Azure before sending its manifest. Inline transitions are limited to 500 steps by the control-plane API.

For a process-level integration, atomically rename canonical episode JSON files into /var/lib/rbrouter/outbox and run:

export RBROUTER_API_KEY='rbr_edge_...'
rbrouter edge \
  --base-url https://rbrouter.com \
  --fleet-id fleet_... \
  --robot-id robot_... \
  --state-directory /var/lib/rbrouter

Delivered inputs are retained under delivered/. Signed policies are staged under policies/ after checksum and ECDSA verification. A protected-metric rollback request is durably written to rollback-request.json and stops the process so the local safety supervisor can switch to its cached approved policy.

Run an execution agent

The runner maps server-selected algorithms to local, operator-approved commands. The control plane never sends arbitrary shell commands.

export RBROUTER_API_KEY='rbr_...'
rbrouter runner \
  --base-url https://rbrouter.com \
  --runner-id warehouse-gpu-01 \
  --adapter-config /etc/rbrouter/runner.json

Example /etc/rbrouter/runner.json:

{
  "ppo": ["python", "/opt/train/ppo.py", "--manifest", "{manifest_file}"],
  "sac": ["python", "/opt/train/sac.py", "--manifest", "{manifest_file}"],
  "iql": ["python", "/opt/train/iql.py", "--manifest", "{manifest_file}"],
  "cql": ["python", "/opt/train/cql.py", "--manifest", "{manifest_file}"],
  "evaluation": ["python", "/opt/eval/offline.py", "--manifest", "{manifest_file}"],
  "simulation": ["python", "/opt/eval/isaac_lab.py", "--manifest", "{manifest_file}"]
}

Each command must write a JSON result to the path provided in the RBROUTER_RESULT_PATH environment variable. Training results include checkpointUri, metrics, and optional artifacts. Evaluation results include the frozen-suite result fields expected by the RB Router evaluation API.

The repository includes an allowlist template at examples/runner.adapters.json and a hardened systemd unit at examples/rbrouter-runner.service. The referenced training programs are deployment-owned adapters: install the organization's reviewed RLlib, TorchRL, or Isaac Lab entry points at those paths. RB Router never downloads executable commands from a task manifest.

An Edge systemd unit is provided at examples/rbrouter-edge.service. Replace the fleet and robot IDs, install the Edge API key in a root-owned environment file, and grant the service account write access only to its state directory.

Pull and verify an approved policy

assignment = edge.assignments()[0]
checkpoint = edge.download_assignment(assignment, "/var/lib/rbrouter/policies/candidate.pt")

# Re-fetch after verification, then atomically switch the local policy symlink
# or supervisor-owned deployment slot inside this callback.
verified = next(item for item in edge.assignments() if item["id"] == assignment["id"])
edge.activate_assignment(verified, activate=lambda item: activate_local_policy(checkpoint, item))

download_assignment streams to a temporary file, fsyncs and atomically renames it, verifies the SHA-256 digest and P-256 signature, and only then marks the assignment verified. Activation remains an explicit caller-owned operation because robot supervisors have embodiment-specific safe-stop requirements.