ACT Harness

ACT Harness: the robot station as a set of tool calls for a coding agent.

MuJoCo is included with the SDK. serve listens on a local Unix socket until Ctrl+C, reconnecting after each station job or connection failure. Every other entry point is one action, one JSON reply::

python -m se3labs.sdk.examples.act_harness serve --station <id> --address <host:port>
python -m se3labs.sdk.examples.act_harness wait
python -m se3labs.sdk.examples.act_harness send move left 0.05 0 0
python -m se3labs.sdk.examples.act_harness run "move left 0 0 -0.03" "gripper left close"
python -m se3labs.sdk.examples.act_harness cameras --out ./frames
python -m se3labs.sdk.examples.act_harness end-episode
python -m se3labs.sdk.examples.act_harness skill --install .claude/skills
python -m se3labs.sdk.examples.act_harness tools --style openai

end-episode ends the rollout normally: station scoring, policy reset, then the station's configured home walk. Its reply acknowledges the request, not arrival home. goto <arm> 0 0 0 only returns to the episode start pose. SDK and worker must both speak protocol v2.

Three layers, thinnest on top:

• tools: Harness, a client with one method per action; TOOLS and tool_definitions describing them for an LLM; call_tool to execute a model's tool call. SKILL.md next to this file teaches an agent the shell workflow and can be installed as a Claude Code skill.
• commands: the line protocol over the socket, its server and client.
• policy: the SDK Policy that owns the arms: motion (a trapezoidal profile per axis group), kinematics (YAM IK), keyboard (a human can drive the same session), and camera snapshots to a RAM-backed directory.

serve repeatedly runs station jobs with this policy; the one-job plain client works too (--request se3labs.sdk.examples.act_harness:ACT_HARNESS_REQUEST) with the bundled ten-minute request.

Two input paths feed the same motion model, and both may be used at once.

Keyboard (the arm's base frame: X forward, Y left, Z up; roll, pitch and yaw turn about base X, Y and Z at the flange)::

left   1 forward  2 backward  3 left  4 right  5 up  6 down  7 open  8 close
       u roll+    i roll-     j pitch+  k pitch-  n yaw+  m yaw-
right  a forward  b backward  c left  d right  e up  f down  g open  h close
       o roll+    p roll-     l pitch+  ; pitch-  . yaw+  / yaw-

A key drives its axis for key_hold_s after the keystroke; holding the key down keeps it moving through the terminal's auto-repeat, releasing it stops the arm within key_hold_s. The open and close keys set the gripper's goal, which it ramps to.

Command socket, for an agent or a script. From serve startup until the process exits, the policy listens on a Unix socket (socket_path, default $SE3LABS_ACT_HARNESS_SOCKET or /tmp/se3labs-act-harness.sock). One command per line; a motion command blocks until it completes, and every command answers one JSON line carrying phase, the rollout's elapsed_s and budget_s, and the state of every arm. Outside a live rollout (before the station starts an episode, while the arms walk to the start pose, during scoring) motion commands are refused at once with the phase, never queued. Status and help remain available between jobs. send help lists the commands: translations are metres, rotations degrees.

Cameras. While a rollout runs the policy writes the latest frame of every camera as a JPEG into snapshot_dir (default /dev/shm/se3labs-act-harness, RAM-backed so the SSD is untouched), refreshed every snapshot_period_s. cameras returns their paths, and the cameras subcommand copies them wherever an agent wants to look.

Motion. Every axis follows a trapezoidal profile: it accelerates at accel_m_s2 (rot_accel_deg_s2) to at most speed_m_s (rot_speed_deg_s) and decelerates to stop exactly at the goal. Each tick advances the profile, solves damped least-squares IK on the bundled YAM model, and sends a chunk holding the predicted trajectory LEAD_S ahead at LEAD_RATE; the next tick replaces it. An idle arm gets a hold chunk IDLE_HOLD_S long, refreshed at half that interval, so the station always has a valid target. X/Y offsets and upward Z travel are limited to reach_m from the start position; there is no lower-Z bound. A goal outside those limits, or a step IK cannot reach, is refused: the arm stops and a socket command answers ok: false. These are kinematic checks only, not the daemon's safety layer.

Keys come from the controlling terminal (/dev/tty) on a reader thread that is only engaged while infer is being called, so the operator's prompts between episodes keep their normal line editing. Without a terminal the keyboard is skipped with a notice and the socket is the only input.

ActHarnessPolicy

ActHarnessPolicy(
    *,
    bindings: dict[str, ArmKeys] | None = None,
    speed_m_s: float = 0.04,
    accel_m_s2: float = 0.25,
    rot_speed_deg_s: float = 30.0,
    rot_accel_deg_s2: float = 120.0,
    key_hold_s: float = 0.5,
    gripper_rate: float = 2.0,
    reach_m: float = 0.3,
    keys: KeySource | None = None,
    keyboard: bool = True,
    socket_path: str | None = DEFAULT_SOCKET,
    snapshot_dir: str | None = DEFAULT_SNAPSHOT_DIR,
    snapshot_period_s: float = 0.2,
    start_q: tuple[float, ...] = START_Q,
)

Drive each arm's flange from the keyboard or a command socket; hold when idle.

Attributes

START_Q = (0.0, 1.5, 1.5, 0.0, 0.0, 0.0)

MAX_SPEED_M_S = 0.15

MAX_ROT_SPEED_DEG_S = 90.0

LIVE_S = 0.5

A rollout is live while infer has run within this long.

LEAD_S = 0.5

Every motion chunk carries the predicted trajectory this far ahead, at this row rate.

LEAD_RATE = 20.0

IDLE_HOLD_S = 5.0

An idle arm's hold chunk reaches this far ahead and is refreshed at half that interval.

bindings

limits

key_hold_s

reach_m

keys

keyboard

socket_path

snapshot_dir

snapshot_period_s

start_q

kinematics: YamKinematics | None

source: KeySource | None

server: CommandServer | None

commands: queue.Queue[Command]

waiting: list[Command]

arms: dict[str, ArmState]

last_timestamp: float | None

last_observation: Station_Observation | None

phase

end_requested

budget_s

first_timestamp: float | None

live_until

hold_sent: float | None

snapshot_at: float | None

snapshots: dict[str, dict[str, Any]]

Methods

start_server() -> None

Expose the command socket even while waiting for a station or episode.

initialize(station: Station_Spec, task: Task_Spec) -> Station_JointPose

print_legend(notes: list[str]) -> None

live() -> bool

refusal() -> dict[str, Any] | None

Why a command cannot run now, or None while a rollout is live. Called from server threads.

reset() -> None

close() -> None

Release the command socket. reset keeps it so a session spans episodes.

infer(observation: Station_Observation) -> Station_ActionChunk | None

apply_keys(keys: list[str], now: float) -> None

submit(command: Command) -> None

Queue a command for the next tick; CommandServer does this from its threads.

apply_commands(now: float) -> None

apply_motion(command: Command) -> None

hold_chunk(timestamp: float) -> Station_ActionChunk

Every arm at its current target, IDLE_HOLD_S ahead.

motion_chunk(timestamp: float, previous: dict[str, MotorizedArm_JointPose]) -> Station_ActionChunk

Row 0 is the previous target at timestamp; the rest follow each arm's profile forward at LEAD_RATE for LEAD_S.

write_snapshots(observation: Station_Observation) -> None

Every camera's latest views side by side as one JPEG, replaced atomically.

finish_settled() -> None

finish_arm(name: str, *, ok: bool, message: str) -> None

finish_all(message: str) -> None

finish_command(command: Command, *, ok: bool, message: str) -> None

reply(ok: bool, message: str) -> dict[str, Any]

ACT_HARNESS_TASK

ACT_HARNESS_TASK = Task_Spec(name='act_harness', description="Drive both arms' end effectors and grippers from the keyboard or the command socket.", reset_instruction="Clear the arms' reach and keep hands away; the keyboard or an agent drives them.", max_rollout_time=600.0, scoring=Task_ScoreSheet(success=Task_ScoreItem('responsive', 'Did the arms follow the commands and hold when idle?', Task_ScoreKind.Binary)))

ACT_HARNESS_REQUEST

ACT_HARNESS_REQUEST = JobRequest(task=ACT_HARNESS_TASK, policy=ActHarnessPolicy(), episodes=1)

Ten minutes of the harness on a bimanual station, once.

Harness

Harness(socket_path: str = DEFAULT_SOCKET, timeout_s: float = 90.0)

A client for one running ACT Harness session.

Motion methods block until the arm has stopped at its goal, or was refused, and return the reply: ok, message, phase, elapsed_s, budget_s, cameras and every arm's state.

Attributes

socket_path

timeout_s

Methods

send(line: str) -> dict[str, Any]

status() -> dict[str, Any]

The phase of the session and the pose, velocity, goal and gripper of every arm.

move(arm: str, dx: float = 0.0, dy: float = 0.0, dz: float = 0.0, roll: float = 0.0, pitch: float = 0.0, yaw: float = 0.0) -> dict[str, Any]

Translate and rotate the arm's flange relative to where it is now; blocks until it stops.

goto(arm: str, x: float, y: float, z: float, roll: float = 0.0, pitch: float = 0.0, yaw: float = 0.0) -> dict[str, Any]

Bring the flange to an offset and orientation relative to its start pose; blocks until it stops.

rotate(arm: str, roll: float = 0.0, pitch: float = 0.0, yaw: float = 0.0) -> dict[str, Any]

Rotate the flange in place; blocks until it stops.

gripper(arm: str, position: float | str) -> dict[str, Any]

Open fraction 0 (closed) to 1 (open), or 'open' / 'close'; blocks until reached.

speed(linear_m_s: float, angular_deg_s: float | None = None) -> dict[str, Any]

Set the speeds later motions cruise at.

hold(arm: str | None = None) -> dict[str, Any]

Stop one arm, or all arms, and cancel their goals.

end_episode() -> dict[str, Any]

Request normal episode completion: station scoring, reset, then homing.

Acknowledges the request, not completion of the station's home walk. Operator scoring may need to finish before homing begins.

cameras(out_dir: str | None = None) -> dict[str, Any]

Write the latest frame of every camera as a JPEG and return their paths; out_dir copies them there.

wait_for_rollout(timeout_s: float = 600.0, poll_s: float = 0.5) -> dict[str, Any]

Block until the session is in a live rollout (the arms accept motion), or timeout_s passes.

run(steps: list[str]) -> dict[str, Any]

Send command lines one after another, stopping at the first that is not ok.

Returns ok (every step succeeded), completed (how many ran) and results (their replies, in order).

send

send(line: str, path: str = DEFAULT_SOCKET, timeout_s: float = 90.0) -> dict[str, Any]

Send one command line to a running policy and return its JSON reply.

call_tool

call_tool(
    harness: Harness,
    name: str,
    arguments: dict[str, Any] | None = None,
) -> dict[str, Any]

Execute one tool call by name. Unknown names and bad arguments answer ok: false rather than raising.

tool_definitions

tool_definitions(style: str = 'anthropic') -> list[dict[str, Any]]

TOOLS in the shape an API expects: anthropic (input_schema) or openai (function calling).