Connecting a policy

Your model runs on your own machine. The station streams observations to it over one gRPC stream and executes the action chunks it returns. You need the job id the operator gives you and the address of the relay or station.

Install

The client is the se3labs package on PyPI. It needs only grpcio and protobuf and imports nothing that touches hardware. Python 3.10 or newer.

pip install se3labs

This installs the se3labs command and two modules: se3labs.eval.policy, the client, and se3labs.eval.msg, which holds every protocol message type.

Try a reference policy

Hold-still connects, answers every observation with the current joint positions, and prints a report when the station closes the session:

se3labs eval run --job job_… --address relay.example.com:7443 --policy hold

Option	Default	Meaning
--job	required	Job id from SE3 Labs
--address	required	Relay or station host:port
--policy	hold	hold, or module:Attr naming your own policy (below)
--horizon	4	Actions per chunk, for the reference policy
--image	none	Request a camera stream, CAMERA[:WxH], repeatable
--secure	off	TLS to the address

Exit status is 0 when the station closed the session normally, 1 if your policy raised during any step, and 2 if the handshake was rejected.

While a job runs, poll its progress:

se3labs eval status --job job_… --address relay.example.com:7443 [--json]

python -m se3labs.eval.policy … is an alias of se3labs eval run ….

Write your own

A policy is any object with three methods:

from collections.abc import Sequence
from se3labs.eval import msg, policy


class MyPolicy:
    def manifest(self) -> msg.ClientManifest:
        return policy.default_manifest(
            model="my-model", version="2026-09-01",
            images=[("cam_chest", 640, 480), ("cam_wrist_l", 320, 240)],
            max_horizon=8,
            stateful=True,
        )

    def reset(self, reset: msg.Reset) -> None:
        # Clear per-episode state. reset.task carries the instruction.
        self.history = []

    def act(self, observe: msg.Observe) -> Sequence[msg.Action] | msg.EndEpisode:
        obs = observe.observation
        if self.task_complete(obs):
            # Ends the episode; the operator then scores it.
            return msg.EndEpisode(reason="goal reached")
        targets = {}
        for arm in obs.arms:
            q = list(arm.joint_pos_rad)          # driver joint order, radians
            g = arm.gripper_pos if arm.HasField("gripper_pos") else None
            targets[arm.name] = (q, g)
        # 1 to negotiated.max_horizon actions, one per dt.
        return [policy.joint_action(targets) for _ in range(8)]


report = policy.serve_policy("job_…", "relay.example.com:7443", MyPolicy())
print(report.close_reason, report.episodes, report.steps, report.errors)

msg exposes every message type of the protocol under its proto name, for example msg.Observe, msg.Action, and the enumeration values such as msg.PROPRIO_FIELD_JOINT_POS. policy exposes the client: serve_policy, get_status, joint_action, default_manifest, the Policy protocol, SessionReport, SessionRejected, and the reference policies.

If the class needs no constructor arguments you can skip the last two lines and run it from the command line. With the file above saved as my_policy.py in the current directory:

se3labs eval run --job job_… --address relay.example.com:7443 --policy my_policy:MyPolicy

--policy module:Attr imports the module, then calls Attr if it is a class or uses it as-is if it is already an instance.

act returns either a sequence of actions or msg.EndEpisode. Returning msg.EndEpisode(reason=...) ends the running episode: the station holds the arms and asks the operator to score the episode as it stands. The client fills in the episode id; the optional reason is recorded in the episode manifest. An episode that is not ended this way runs until the station's step limit, its timeout, or the operator ends it.

serve_policy blocks for the whole session and returns a SessionReport with the Welcome message, episode and step counts, the number of episodes the policy ended, the close reason, and any exceptions your act raised. An exception in act sends an empty chunk for that step and is recorded, not fatal. A rejected handshake raises SessionRejected with the code and message.

policy.FunctionPolicy(fn) wraps a bare fn(observe) -> actions with a default manifest if you would rather not write a class.

What to expect

1. Your manifest is bounded by the station's caps: image sizes are limited to the eval config's per-camera ceilings, the horizon to its maximum, and the action schema must be one the station offers. Anything the station cannot serve rejects with PROFILE_UNSUPPORTED.
2. Preflight. The station sends a Reset with synthetic=true followed by five Observe messages carrying real observations. The returned chunks are not executed. Each chunk is validated for structure: the echoed identifiers, an action count between 1 and max_horizon, one JointPositionAction per commanded arm with the correct joint count, and finite values throughout. Round-trip time is measured for each step. Preflight fails with PREFLIGHT_FAILED if any chunk is missing, late, or structurally invalid, or if the p95 round-trip time exceeds deadline_ms. The safety profile (joint step, speed, workspace, and collision limits) is not evaluated during preflight; it applies to chunks submitted for execution.
3. Each episode is a Reset, which you must acknowledge, then Observe messages at the negotiated rate. Answer each with one Chunk, or with EndEpisode to finish the episode.
4. Close ends the session.

Read the protocol for every message and rule, and observations and actions for what is in them.

Timing

The station enforces the deadline on its own clock. A chunk that arrives late is discarded. Too many consecutive misses invalidate the episode, so aim to answer well inside deadline_ms (250 ms by default) and return several actions per chunk so the buffer covers a slow step. A newer chunk supersedes the unexecuted tail of the previous one, so re-planning every step is expected.