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             # Runtime, MyRobotToolkit, observations (§3)
    tools.py               # Native tool declarations and handlers (§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(
    *,
    runtime_kwargs,
    dashboard_events: DashboardEventSink,
    config: RunConfig,
):
    from robots.myrobot.toolkit import MyRobotToolkit
    return MyRobotToolkit(
        runtime_kwargs=runtime_kwargs,
        output_dir=config.output_dir,
        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, runtime_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; and primitives is the ordered allowlist of Toolkit actions displayed and executable as Dashboard controls. Camera tabs are discovered from the PNG artifacts in each recorded step. 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#

Split robot behavior between tools.py and toolkit.py, following robots/libero/, robots/robocasa/, or robots/robotwin/:

Runtime — a per-session object such as MyRobotRuntime holds the env and model clients, cached observations, and task progress. Construct it from runtime_kwargs returned by init_runtime. Tools access it through ctx.robot.

Native tools — define typed functions in tools.py with @tool and a required keyword-only ctx: ToolContext[MyRobotRuntime]. Return ToolResult and collect the declarations in a tuple such as MYROBOT_TOOLS. Tool schemas come from signatures and Google-style docstrings; planner adapters handle SDK and MCP serialization. See Add an Action Primitive for an example.

Observations — dump_state(runtime, env_state, log) opens env_state.record_step(...) to allocate and commit a StepRecord. Save arrays, images, and metadata through env_state.save(...): inside the block, omitting step targets the new step; an explicit integer targets another step and step=None creates a run-level artifact. EnvState adds each saved logical name to the record’s artifacts set. build_observation(state, record) returns JSON data and ordered PNG bytes for both action responses and view_env_state.

Toolkit — subclass Toolkit[MyRobotRuntime]:

  • Construct the runtime and a fresh EnvState(output_dir); pass state, memory, robot=runtime, output_dir, tools=MYROBOT_TOOLS, and dashboard_events to super().__init__. Configure memory_access and inbox_cell_tag on MemoryManager; evaluation defaults to read-only memory.

  • Initialize the environment and record step 0. Publish its StepRecordEvent so the Dashboard can display the initial observation.

  • Implement _capture_observation(*, command, result, elapsed_s) using dump_state and build_observation. Save the original result.to_dict() in the step log. Return an observation containing the step, artifact names, state, and action log, together with its PNG images. The base executor calls this after non-readonly robot handlers, including failed handlers, except finish. Common tools also skip capture.

  • Implement solved() using the environment’s success criterion. Provide a robot-specific finish tool returning status and summary; the executor stores accepted results in toolkit.finish_result.

Tools submit environment-step RGB frames via ctx.record_frame(rgb). The base toolkit owns the frame buffer, Dashboard action clips, episode video, recipe export, and cancellation lifecycle. Use the inherited close() for recording cleanup; if a robot needs additional cleanup, preserve the base call.

Conventions worth keeping#

  • output_dir is the runner-created working directory. Environment artifacts are managed by EnvState through logical names; transcripts share the same run directory.

  • @readonly below @tool enables shared execution and skips automatic observation capture. Keep actions and finish exclusive and call ctx.check_cancelled() at safe boundaries in long loops.

  • Server-side values use transport-supported Python / NumPy types and remain torch-free.

  • Add new observation modalities in dump_state and expose them through build_observation. view_env_state reads a saved step without stepping or re-rendering the environment.

See Core interfaces for the full tool execution and lifecycle contract.

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, runtime_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.

  • runtime_kwargs: dict is passed to the toolkit constructor, which uses it to construct the robot runtime. 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 runtime_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.