Metadata-Version: 2.5
Name: abstract-client
Version: 0.1.0
Summary: Abstract's robot API from Python: action chunks from served robot models, sync and asyncio.
License-Expression: Apache-2.0
License-File: LICENSE
License-File: NOTICE
Requires-Python: >=3.10
Requires-Dist: httpx2<3,>=2.12
Description-Content-Type: text/markdown

# 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()`.

```sh
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`).

```python
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.

```python
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.

```python
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
```

| DreamClient | atoms.AtomsClient |
|---|---|
| `DreamClient(api_key)` | `AtomsClient(api_key)`, or `AtomsClient(client=...)` |
| `get_models()` | `get_models()`: served models and your fine-tunes, with their rates |
| `policy({"checkpoint": ...})` | `policy({"model": ...})`; `execution_horizon`, `action_hz`, `sync_above_ms`, `timeout_ms` as in DreamClient |
| `set_observation(frames, observation, timestamp=)` | the same: `video.<camera>` frames, `prompt` and `state` (a dict or `state.<name>` keys) |
| `get_action()` returns joint chunks | `get_action()` returns a `Chunk` of end-effector poses: your arm's IK turns them into joints |
| `reset_episode()`, `refill_due()`, `close()` | the same |
| `mode`, `execution_horizon`, `action_hz`, `last_timing`, `checkpoint`, `session_id` | the same, `model` for `checkpoint` |
| `obs_window`, `obs_fps` | none: each answer says how many frames the next observation carries |
| `DreamSlowNetworkWarning` | `atoms.SlowNetworkWarning` |
| one model per client | `policies({name: config})`, `get_actions()`, `robot[name]` |

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:

```python
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`).
