abstract-client

Abstract's robot API from Python. A Client holds your key and its connections and is safe to share between threads; each Policy is one episode of one served robot model doing one task, one action chunk per predict().

uv add abstract-client

This is all a robot with its own stack needs. The abstract command and the robot station are abstract-robots, built on it, and abstract-atoms installs both (apps/atoms).

import atoms

client = atoms.Client()  # the key: api_key=, else ABSTRACT_API_KEY, else `abstract login`'s file

with client.connect("abstract/dreamzero-astra", "fold the towel") as policy:
    frames = {"overhead": top_jpeg, "wrist_l": left_jpeg, "wrist_r": right_jpeg}
    while not done:
        state = {
            "left_eef_xyz": [x, y, z],  # metres, in the left arm's base frame
            "left_eef_rot6": rot6,  # the rotation matrix's first two columns
            "left_grip": [g],  # 0 closed, 1 open
            "right_eef_xyz": [x, y, z],
            "right_eef_rot6": rot6,
            "right_grip": [g],
        }
        chunk = policy.predict(frames, state)
        # Row k is a pose for t + k * chunk.dt; your arm's inverse kinematics turns each into
        # joint angles. chunk.elapsed_rows were already due when the answer arrived.
        for left, right in zip(chunk.left[chunk.elapsed_rows :], chunk.right[chunk.elapsed_rows :]):
            robot.command(left, right)
        # The next call sends policy.frames_needed JPEGs per camera, oldest first.
        frames = latest_frames(policy.frames_needed)
# Leaving the block ends the episode now; the server would after 60 s idle.

A stateful model (DreamZero) takes the video since its last observation: after the first call, send policy.frames_needed frames per camera.

atoms.AtomsClient: Dream Labs' DreamClient on this API

atoms.AtomsClient keeps that window for you and paces the requests, with the calls of Dream Labs' DreamClient (Apache-2.0; see NOTICE): bind the key, choose the model, push every camera tick, pull a chunk when it says so.

from concurrent.futures import ThreadPoolExecutor

robot = atoms.AtomsClient()  # the key, as atoms.Client finds it
robot.policy({"model": "abstract/dreamzero-astra"})
with robot, ThreadPoolExecutor(1) as worker:  # the request runs while the robot moves
    chunk, row, pending = None, 0, None
    while not done:
        frames = {"video.overhead": top, "video.wrist_l": left, "video.wrist_r": right}
        robot.set_observation(frames, {"prompt": "fold the towel", "state": read_state()})
        rows_left = len(chunk.left) - row if chunk else 0
        if pending is None and robot.refill_due(row, rows_left):
            pending = worker.submit(robot.get_action)
        if pending is not None and pending.done():
            chunk, pending = pending.result(), None
            row = chunk.elapsed_rows
        if chunk and row < len(chunk.left):
            robot.command(chunk.left[row], chunk.right[row])
            row += 1
        wait_for_next_tick()

Several models on one robot

DreamClient runs one model. AtomsClient runs as many as you open on the same camera ticks, each with its own episode, frames (DreamZero takes four per camera, MolmoAct2 one), pacing and timing. One drives the arms; the others answer the same observations, to compare them.

robot.policies(
    {"dz": {"model": "abstract/dreamzero-astra"}, "molmo": {"model": "allenai/molmoact2"}}
)
robot.set_observation(frames, observation)  # one tick, every model
answers = robot.get_actions()  # every model at once: {name: Chunk, or the error it raised}
drive = robot["dz"]  # one model: get_action(), refill_due(), last, mode, last_timing

Not here yet: DreamClient's seam smoothing (commanded), datasets and fine-tuning. Embodiments, ik and Live View have no counterpart on this API.

atoms.AsyncClient is native asyncio with the same arguments, rules and calls (await policy.predict(...)). Cancelling a call cancels its request, but the server may already be running it, so the next predict() asks for that answer first, as after a lost connection.

Connections are HTTPX2's (Pydantic's maintained HTTPX, as the OpenAI and Anthropic SDKs use). Proxies come from HTTPS_PROXY and NO_PROXY, and TLS is checked against the operating system's trust store, so a CA your IT installed works as is. For a private CA, a client certificate or a proxy in code, configure HTTPX2 and pass it in; the client you pass stays yours to close:

import ssl

import httpx2

tls = ssl.create_default_context(cafile="lab-ca.pem")
tls.load_cert_chain("robot.pem", "robot.key")
client = atoms.Client(http_client=httpx2.Client(verify=tls, proxy="http://proxy.lab:3128"))

connect(model, task, scaling={"seed": 7}) (or scaling in an AtomsClient config) sends the API's scaling with every observation, so a seedable model replays its noise; the engine's reports (scaling, cache, context) come back in chunk.raw.

Each chunk is billed and can wake a GPU. A call that gets no answer is sent again with the same bytes and observation_id, so the server answers it from its record and never bills it twice. Refusals the server marks retryable are sent again after its Retry-After, up to max_retries; timeout bounds a whole call, resends included, and connect_timeout opening a connection.

Every error is an atoms.AbstractError. A refusal is an atoms.APIStatusError subclass named for its status (AuthenticationError, PaymentRequiredError, RateLimitError, ...) with the server's code, retryable, retry_after and request_id. APIConnectionError and APITimeoutError mean no answer came: the next predict() asks for it first. The atoms logger reports resends; it never logs the key.

Python 3.10 or later; its one dependency is httpx2. Tests: pytest apps/atoms (the contract tests need Bun and the engine's packages: uv run --project apps/engine pytest apps/atoms).