Add a New Robot#

This guide walks through what you need to write to plug a new physical / simulated robot into RPent’s LLM-in-the-loop runner. Use robots/libero/ as the worked reference.

Integration guidelines#

  • Reuse RPent abstractions. Prefer existing Env, VLA, runtime, and memory components such as BaseEnvClient, BaseEnvFacade, BaseVLAClient, BaseVLAFacade, and MemoryManager.

  • Prefer RLinf Env or VLA implementations. If RLinf already supports the required Env or VLA, keep RPent as a thin adapter where possible.

  • Stay consistent with existing robot integrations where possible. Follow the existing RobotSpec, Prompt, Toolkit, and runtime patterns instead of introducing a new shared mechanism for one robot.

  • Explain when reuse is not possible. If an existing RPent or RLinf Env / VLA cannot be reused, explain why in the PR description or open an issue so the existing abstraction can be improved.

Integration steps#

For the overall process layout, service responsibilities, and communication model, see System Design. This guide focuses on the extension points required to add a robot. Complete them in the following order:

  1. Register the RobotSpec and toolkit factory in the entry point.

  2. Implement env_client and env_server. To integrate a VLA service and model client, see Add a VLA (or other model-based primitive).

  3. Define the prompts.

  4. Implement the toolkit and primitives.

  5. Register robot arguments and build RunConfig.

  6. Implement the runtime hook. The same hook starts the complete runtime for normal CLI runs or a selected component subset for the Dashboard.

  7. Add tests for the environment, its components, and the complete policy chain.

Entry point#

For a new robot named myrobot, use the following directory layout:

robots/myrobot/
    __init__.py            # package entry point; re-exports the factories
    robot_spec.py          # RobotSpec, factories, Dashboard spec, runtime hooks
    env_client.py          # MyEnvClient — agent-side RPC stub (§1)
    prompt_bundle.py       # system()/user() prompt factories         (§2)
    toolkit.py             # MyRobotToolkit + primitives + tool definitions (§3)
    env_server.py          # server-side facade + RPC server (§1)
    vla_server.py          # (optional) VLA model server

__init__.py is the robot package’s entry point. Keep it small and re-export the factories implemented in robot_spec.py. The registry in rpent/robots/base.py lazily imports robots.<name> on demand and calls these two functions:

# robots/myrobot/__init__.py
from robots.myrobot.robot_spec import get_robot_spec, get_toolkit

# robots/myrobot/robot_spec.py
from rpent.dashboard.events import DashboardEventSink
from rpent.memory import MemoryManager
from rpent.robots.robot_spec import RobotSpec, RunConfig
from rpent.robots.prompt_bundle import PromptBundle
from rpent.utils.config import get_memory_dir
from robots.myrobot.prompt_bundle import system_prompt, user_prompt

MYROBOT_DASHBOARD_SPEC = {...}

def get_robot_spec() -> RobotSpec:
    return RobotSpec(
        name="myrobot",
        prompts=PromptBundle(system=system_prompt, user=user_prompt),
        add_cli_args=_add_cli_args,
        parse_config=_parse_config,
        init_runtime=_init_runtime,
        dashboard=MYROBOT_DASHBOARD_SPEC,
    )

def get_toolkit(
    *,
    primitives_kwargs,
    dashboard_events: DashboardEventSink,
    config: RunConfig,
):
    from robots.myrobot.toolkit import MyRobotToolkit
    return MyRobotToolkit(
        primitives_kwargs=primitives_kwargs,
        dashboard_events=dashboard_events,
        memory=MemoryManager(
            root=config.prompt_vars.get("memory_dir") or get_memory_dir("myrobot"),
        ),
    )

def _add_cli_args(parser, use_dashboard) -> None:
    """Register robot flags on the shared parser. See §4."""
    ...

def _parse_config(args) -> RunConfig:
    """Validate final `args`, return a RunConfig. See §4."""
    ...

def _init_runtime(
    args,
    output_dir,
    dashboard_events: DashboardEventSink,
    components: set[str] | None,
):
    """Initialize all runtime components, or only the selected subset.

    Returns (daemons, primitives_kwargs). See §5.
    """
    ...

dashboard is optional. Leave it as None if the environment does not support Dashboard control. Otherwise, define the spec in the robot package: its task section describes the command, validated fields, display template, and output slug; robot-specific Session settings remain normal CLI arguments; runtime_components describes service rows; frame_channels maps camera names to canonical image artifacts; and primitives is the ordered allowlist of Toolkit actions displayed and executable as Dashboard controls. Keep task suggestions in the spec so importing the robot does not require simulator packages. See robots/libero/robot_spec.py for the reference shape.

That’s the entire registration step — _resolve_robot(name) does an importlib.import_module(f"robots.{name}"), so dropping the package under robots/ on disk is enough. No central list to update.

The sections below describe what each referenced module must contain. _add_cli_args / _parse_config are covered in §4 and the runtime hook in §5. The Dashboard spec is consumed only by the Dashboard runner.

1. env_client.py + env_server.py#

These files connect the agent process to env_server. The client converts method calls into RPC requests, and env_server handles those requests.

1.1 Env client (agent side)#

Subclass rpent.robots.components.env_client_base.BaseEnvClient. It already validates env.get_env_meta at startup, performs the initial reset, caches last_obs, and implements the common reset, step, chunk_step, render_camera, get_camera_meta, and get_task_language RPCs. Add only environment-specific methods, and extend the timeout table when an extension needs its own timeout. Keep RPC names stable — the server-side facade registers each name explicitly.

from rpent.robots.components.env_client_base import BaseEnvClient

class MyEnvClient(BaseEnvClient):
    _TIMEOUT_S = {
        **BaseEnvClient._TIMEOUT_S,
        "env.custom_method": 30.0,
    }

    def custom_method(self, arg):
        return self._client.call(
            "env.custom_method",
            args=(arg,),
            timeout_s=self._TIMEOUT_S["env.custom_method"],
        )

env = MyEnvClient(rpc_client, expected_meta=expected_meta)

1.2 Env server (server side)#

Mirror the client’s API in a facade class on the server side (e.g. MyEnvFacade). Subclass rpent.robots.components.env_facade_base.BaseEnvFacade; it provides the common RPC routes and read/write dispatch locking. Implement the common environment methods and extend _register_rpc for environment-specific routes. Methods take the same positional / keyword arguments the client sends and return transport-supported Python / NumPy values (not torch — the agent side does not import torch).

from rpent.robots.components.env_facade_base import BaseEnvFacade

class MyEnvFacade(BaseEnvFacade):
    def __init__(self, env, meta):
        self._env = env
        self._meta = meta
        super().__init__()

    def _register_rpc(self):
        super()._register_rpc()
        # Custom methods must be registered explicitly
        self._rpc["env.custom_method"] = self.custom_method

    # Abstract methods required by BaseEnvFacade
    def get_env_meta(self): ...
    def reset(self): ...
    def step(self, action): ...
    def chunk_step(self, actions, **kwargs): ...
    def get_camera_meta(self, camera_name, **kwargs): ...
    def render_camera(self, camera_name, **kwargs): ...
    def get_task_language(self): ...

    def custom_method(self, arg): ...

facade = MyEnvFacade(env, meta)
facade.serve(transport="http", host=host, port=port)

BaseEnvFacade registers common routes through _register_rpc and serializes state-changing calls with its read/write lock. Add a route to _readonly_methods only when it is genuinely safe to run concurrently with other reads. The inherited RpcFacade.serve handles transport binding (HTTP or socket), healthz / shutdown, parent-death detection, and clean teardown.

2. prompt_bundle.py#

Define two prompt factories, system_prompt() and user_prompt(), and build a PromptBundle(system=system_prompt, user=user_prompt) in the robot’s robot_spec.py (see the entry point above). Each factory returns an ordered dict[str, PromptNode] of titled sections; PromptBundle.render assembles and fills them. One prompt serves every planner (API loop, Claude Code, Codex): refer to tools by their bare names (move_to, …) and note once that the Claude Code and Codex SDKs show them as mcp__rpent__<name>. Do not maintain separate prompt copies for CLI and API planners.

# robots/myrobot/prompt_bundle.py
from robots.myrobot.prompts import system as system_parts
from robots.myrobot.prompts import user as user_parts
from rpent.prompt.utils import PromptNode

def system_prompt() -> PromptNode:
    return {
        "INTRO": system_parts.PREAMBLE,
        "GOAL": system_parts.GOAL,
        "RULES": system_parts.RULES,
        "WORKFLOW": system_parts.WORKFLOW,
        "ENVIRONMENT": system_parts.ENVIRONMENT,
        "OUTPUT": system_parts.OUTPUT,
    }

def user_prompt() -> PromptNode:
    return {
        "TASK": user_parts.TASK,
        "BEGIN": user_parts.BEGIN,
    }

Keep the prompt content under the robot package, for example in robots/myrobot/prompts/system.py and user.py. Section bodies are plain strings (or BulletList / Numbered) with {{suite}} / {{task}} / {{seed}} / {{output_dir}} / {{recipe_tag}} placeholders filled at render time.

3. toolkit.py#

This module owns everything the LLM can call: the tool schemas, the primitives, the per-step state dump, and the MCP allowlist. (In the LIBERO robot these are split between tools.py and toolkit.py for historical reasons; for a new robot it is fine to keep them all in toolkit.py.)

A toolkit module typically contains four pieces:

Primitives class (e.g. MyRobotPrimitives) — a Python object owned by the toolkit. It holds the EnvClient, the VLA model client, and any state needed for the current run. It exposes one method per primitive tool (move_to, pi0_pick, release, …), with each method returning a dict log.

Tool definitions and handlers — a module-level TOOLS_SPEC list of Anthropic-style tool definitions (name, description, input_schema), plus any module-level functions referenced by the toolkit (e.g. view_env_state, back_project, finish).

Per-step state dump — dump_state(driver, env_state, log) opens env_state.record_step(...) and receives the allocated step index; the StepRecord is appended and committed immediately. Save large observations through env_state.save(...) — inside a record_step block the step argument may be omitted (it defaults to the new step), pass an explicit step=<int> to target a different step, and step=None for run-level artifacts. EnvState adds every successfully saved base name to the step’s flat artifacts set automatically. Readers use the canonical artifact filenames rather than maintaining a parallel observation index.

Toolkit class — subclass rpent.tools.toolkit.Toolkit:

  • forward memory (a MemoryManager) and state to super().__init__(...). Configure memory_access and inbox_cell_tag on the MemoryManager; eval uses read-only access by default.

  • build the primitives in __init__ through a custom initialization helper (named init_primitives in LIBERO; it calls EnvState.reset(), constructs the primitives, and dumps step 0),

  • register each tool with self.add_tool(name, spec, handler) — stateless readers (view_env_state, finish, …) bind directly to module-level functions; primitive tools route through _step(name, **kwargs) which calls getattr(self._primitives, name)(**kwargs) and re-renders state,

  • override close() to save remaining agent-side artifacts through EnvState (for example state.save("episode.mp4", frames, step=None)).

primitives_kwargs (forwarded from robot_spec.py:get_toolkit) is the dict the toolkit passes verbatim to your primitives’ __init__ — typically {"env": MyEnvClient(...), "model": VLAClient(...), ...}.

Conventions worth keeping#

  • output_dir is the working directory that the runner creates for each run. Environment observations are owned by EnvState; callers use logical base names and never construct storage paths. Transcripts and other run-management outputs share the same run directory.

  • Tool definitions use the Anthropic format (name / description / input_schema). Every tool registered with self.add_tool(...) is exposed to all planners.

  • Server-side return values must be picklable and torch-free.

  • Each primitive tool dumps a fresh state snapshot after running so the next view_env_state call reflects the post-action world.

  • Treat dump_state as the source of truth for what the agent sees — any new modality (e.g. tactile, force) goes through it.

4. _add_cli_args + _parse_config (runner hooks)#

Robot-specific CLI arguments enter rpent/cli/main.py through two hooks and participate in the final argparse pass:

``_add_cli_args(parser, use_dashboard) -> None``. Register the robot’s arguments on the shared parser created by main.py. use_dashboard determines whether normally required arguments remain optional. For each Dashboard TaskRun, the robot’s Dashboard task command supplies the fields declared by its spec before parse_config runs. main.py calls this hook before parser.parse_args(), so argparse’s usage and error output includes the robot arguments.

``_parse_config(args) -> RunConfig``. In normal CLI mode, this is called after parser.parse_args(). In Dashboard mode, it is called for each TaskRun after the requested fields have been copied to the task arguments. It validates those fields and returns a RunConfig:

  • recipe_tag — robot’s per-run tag, used in transcript filenames / recipe path (LIBERO: f"{suite.replace('libero_', '')}_t{task}_s{seed}").

  • output_dir — path to the working directory for this run (main.py then calls init_output_dir to create it and configure logging).

  • prompt_vars — dict passed to PromptBundle.render (typically the run identifiers plus anything else the prompts reference).

  • task_desc — robot-specific dict of task-identifying fields, written into the transcript JSON record verbatim (LIBERO: {"suite": ..., "task": ..., "seed": ...}).

def _add_cli_args(parser, use_dashboard) -> None:
    required = not use_dashboard
    parser.add_argument("--suite", default=None, required=required)
    parser.add_argument("--task", type=int, default=None, required=required)
    # ... other robot-specific flags ...

def _parse_config(args) -> RunConfig:
    if not args.suite: raise ValueError("--suite is required")
    # ... derive recipe_tag, output_dir, and prompt_vars ...
    return RunConfig(
        recipe_tag=recipe_tag,
        output_dir=output_dir,
        prompt_vars=prompt_vars,
        task_desc={"suite": args.suite, "task": args.task, "seed": args.seed},
    )

5. Runtime initialization hook#

init_runtime returns (owned_daemons, primitives_kwargs):

  • owned_daemons: list[ProcessDaemon] contains only subprocesses started by this process. The active runner stops them during cleanup. A client for an external endpoint must not add that external service to this list.

  • primitives_kwargs: dict is passed to the toolkit constructor, which forwards it to the primitives’ __init__. A complete set commonly contains {"env": MyEnvClient(...), "model": VLAClient(...)} plus any supporting clients.

The fourth argument, components, selects which named services to initialize. None means all services and is what the normal CLI passes. The Dashboard derives two subsets from dashboard.runtime_components. Every component declares either scope: "shared" or scope: "unique" explicitly. The Dashboard initializes shared components once, then initializes unique components for every fresh environment instance. It calls this same hook for both subsets and merges the returned primitives_kwargs dictionaries. For LIBERO, the subsets are {"vla", "sam3"} and {"env"}.

An implementation should reject unknown component names before starting anything. When several selected local services are expensive to initialize, start them all before waiting for readiness so their initialization can overlap. See robots/libero/robot_spec.py for the ordered component registry used by the reference implementation.

Endpoint parsing (--env-endpoint, --vla-endpoint, and LIBERO’s --sam3-endpoint) and robot-specific server commands belong in the hook that owns the corresponding service. Wrap those spawners with rpent.robots.runtime.try_spawn_server and try_wait_server so status events, readiness failures, and owned-daemon cleanup stay consistent across robots. The runners do not handle these environment details. See robots/libero/robot_spec.py and robots/robocasa/robot_spec.py for the reference pattern.

Optional run-result finalizer#

RobotSpec.finalize_run is a universal, robot-agnostic end-of-run hook. RoboCasa is its current consumer and uses it to record per-cell evaluation results for later statistics and aggregation. Any robot that publishes machine-readable evaluation artifacts may register the hook. The default is None and leaves the runner unchanged. When the hook is present, the normal terminal runner captures toolkit.solved() before closing the toolkit, then passes a structured RunFinalizationContext to the hook after runtime cleanup. The hook owns the artifact schema and filename; RPent only defines the lifecycle boundary.

Use write_json_atomic when the artifact is JSON so an interrupted write cannot leave a partial result:

from rpent.evaluation import RunFinalizationContext, write_json_atomic

def _finalize_run(context: RunFinalizationContext):
    return write_json_atomic(
        context.output_dir / "result.json",
        {
            "robot": context.robot_name,
            "task": dict(context.task_desc),
            "success": context.environment_success,
        },
    )

Register the callback as RobotSpec(..., finalize_run=_finalize_run). This hook is currently limited to normal terminal runs; the Dashboard does not call it. Keep benchmark manifests, robot-specific runtime fields, and aggregation logic in the robot package rather than the shared CLI.

6. Tests to add#

When adding a robot, test each runtime component it uses separately, then run one complete policy chain. For example, a robot using Env, Pi0.5, and SAM3 needs an Env test, a Pi0.5 inference test, a SAM3 segmentation test, and a policy-chain test. Each component test must make a real request and check the result, beyond importing the module or checking server health.

Test locations#

Use myrobot below for the new robot’s package name:

  • tests/unit_tests/robots/myrobot/: offline tests for configuration, client argument handling, tool dispatch, and runtime startup/cleanup logic. Use fakes for simulators and models so these tests run on CPU.

  • tests/e2e_tests/myrobot/test_components.py: one named test per real component, such as test_environment_component, test_pi05_component, and test_sam3_component. Include only the components the robot uses.

  • tests/e2e_tests/myrobot/test_policy_chain.py: one test connecting the planner, toolkit, model, and environment.

  • Keep fixtures in the same directory’s conftest.py and reusable scenario setup/calls in scenario.py. Reuse lifecycle and assertion helpers from tests/e2e_tests/common.py. See tests/e2e_tests/libero/ for an example.

What each component test should check#

  • Env: start the real environment, reset a fixed task/seed, and read an observation. Check required camera images and state fields, including their shapes and dtypes. Execute at least one valid action and check the next observation and termination/success information.

  • VLA or other action model: load a real checkpoint and perform one prediction using an observation and instruction in the robot’s input format. Check that the returned actions are nonempty, finite, and have the expected dimensions for this environment.

  • Perception or other services: call each service with a known input and validate its output. For example, ask SAM3 to segment a known object and check that the mask matches the image dimensions and contains foreground.

Start each component through the supported runtime interface, and verify that its owned daemons exit after the test. Existing tests may cover a reused component; identify that coverage and test any new input/output adaptation.

Complete policy-chain test#

After component tests pass, use run_scripted_policy_chain from tests/e2e_tests/common.py to run the public CLI with a fixed task/seed and bounded action count. Its local OfflinePlannerServer requests a real action primitive followed by finish, without an external LLM API. Keep the environment and model services real; do not monkeypatch CLI or runtime internals.

Check that at least one environment action was executed, finish was recorded in the transcript, states.json contains the action without errors, expected observation artifacts exist, and owned daemons have exited. Task success is not required for this bounded integration check.

Run offline tests with pytest tests/unit_tests/robots/myrobot -v. After installing the robot extra and preparing its GPU/checkpoint/assets resources, run pytest tests/e2e_tests/myrobot -v. See tests/README.md and tests/e2e_tests/run_gpu_suite.sh for running GPU suites in clean environments.