Get your first actions

Select a checkpoint and request one action chunk using synthetic observations.

Install the reference client

New to Reactor? How the API works explains the session, camera streams, commands, and action messages used below. You need Python 3.12, uv, a Reactor API key, and Reactor’s public cookbook. The client code is public; your Reactor account still needs access to the selected hosted model. No local model weights or GPU are required for this example.

git clone https://github.com/reactor-team/reactor-cookbook.git
cd reactor-cookbook
cd robotics/sim/notebooks
uv sync --frozen --python 3.12
export REACTOR_API_KEY='your-api-key'

Run the example below from robotics/sim/notebooks/. This environment installs the Python SDK (reactor-sdk, imported as reactor_sdk) and the reactor_robotics helper directory in Reactor’s public cookbook. The helpers provide camera publishing and message queues; they are separate from the SDK. Keep your API key in the environment. On Linux, the published glibc wheel requires glibc 2.34 or newer. The default endpoint is https://api.reactor.inc; REACTOR_API_URL overrides it. Examples select a model by slug, not a frozen release; check its release notes when the deployed contract changes. If setup or connection fails, see first-run troubleshooting.

Request a chunk

Run the cookbook example from the current robotics/sim/notebooks/ directory:

uv run --frozen python first_flux_actions.py

import asyncio
import json

import numpy as np
from reactor_robotics.session import ReactorSession

TRACKS = ("wrist_view", "exterior_view_1", "exterior_view_2")


async def main():
    session = ReactorSession("reactor/flux3-action-droid", fps=15, frame_size=(360, 640))
    try:
        await session.connect(
            TRACKS, subscribe=("checkpoint_selected", "action_prediction", "command_error")
        )
        await session.send("get_checkpoint", {})
        status = await session.next_message("checkpoint_selected", timeout_s=20)
        selected = status["checkpoint"]
        if selected not in status["available"]:
            raise RuntimeError("Default checkpoint unavailable; select a listed alternative")
        await session.send("select_checkpoint", {"checkpoint": selected})
        confirmed = await session.next_message("checkpoint_selected", timeout_s=20)
        assert confirmed["locked"] and confirmed["checkpoint"] == selected
        await session.send(
            "set_task_description", {"task_description": "put the marker in the cup"}
        )
        rng = np.random.default_rng(0)
        session.set_frames({
            name: rng.integers(0, 256, (360, 640, 3), dtype=np.uint8)
            for name in TRACKS
        })
        request = {
            "proprio": [0.0, -0.6, 0.0, -2.2, 0.0, 1.6, 0.8, 0.0],
            "chunk_id": 1,
            "seed": 1,
        }
        await session.send("set_state_json", {"state_json": json.dumps(request)})
        reply = await session.next_message("action_prediction", timeout_s=90)
        actions = np.asarray(reply["actions"], dtype=np.float64)
        assert reply["step"] == request["chunk_id"]
        assert actions.shape == (32, 8) and np.isfinite(actions).all()
        assert reply["checkpoint"] == selected
        print("checkpoint:", selected, "actions:", actions.shape)
    finally:
        await session.close()


asyncio.run(main())

Expect a checkpoint name and actions: (32, 8). The repeating publisher supplies post-request frames; it does not simulate a moving robot. To compare checkpoints, run separate sessions and select each desired value from available before sending state. Next, build a simulator adapter or use the integration guide with your robot client.

Get started

Models

Get your first actions

Install the reference client

Request a chunk

​Install the reference client

​Request a chunk

Install the reference client

Request a chunk