AutoSynth Python SDK

Typed Python client and CLI for the AutoSynth experiment-check API. The core flow: a chemist registers an observation from the lab against an approved plan, submits a perception check, and gets back the status and a structured result. The SDK also covers review, replay, lineage, datasets, and evaluation.

• Sync and async clients (AutoSynth, AsyncAutoSynth) with pydantic models for every response.
• Retries with exponential backoff on 429/5xx and network errors, honoring Retry-After.
• Writes are only resent when that is safe: artifacts are content-addressed, and check submissions carry an auto-generated Idempotency-Key.
• runs.wait() polls with backoff. Failed runs come back as data with an error code; they don't raise.
• An autosynth command line, plus reproducible synthetic vial images so you can try everything without a camera.

Install

pip install autosynth-sdk        # or: uv add autosynth-sdk

Requires Python 3.10+. The distribution is autosynth-sdk; the import name is autosynth.

Quickstart

from autosynth import AutoSynth, synthetic

with AutoSynth() as client:                      # AUTOSYNTH_API_KEY, AUTOSYNTH_BASE_URL from the environment
    plan = client.plans.list(status="approved")[0].plan_ref
    photo = synthetic.vial(plan, seed=7, scenario="positive").png   # or open("vial.png", "rb").read()
    run = client.checks.run(photo, plan)         # register -> submit -> wait
    print(run.status, run.outcome, run.result.confidence)           # succeeded positive 1.0

The same flow, one step at a time:

artifact = client.artifacts.register("vial.png", "PLN-0001", metadata={"camera": "bay2-cam1"})
run = client.checks.submit("PLN-0001", artifact.artifact_id)        # 201, usually status "queued"
run = client.runs.wait(run, timeout=60)                             # polls until succeeded | failed

if run.succeeded:
    print(run.result.outcome, run.result.confidence, run.result.model, run.result.findings)
    print(run.provenance.input_sha256, run.provenance.config_hash, run.provenance.code.git_sha)
else:
    print(run.error.code, run.error.message)     # e.g. image_quality: observation too blurry to assess

Authentication

Create an API key on the Developers page of the staff app at https://autosynth.bio, then export it:

export AUTOSYNTH_API_KEY=...                       # sent as  Authorization: Bearer <key>

You can also pass the key directly: AutoSynth(api_key=...). Each key is bound to one project, and every id you see (plans, artifacts, runs) is scoped to that project. Ids from other projects look the same as missing ones and return 404. Keys act in staff mode. The customer-portal endpoints (/portal/*) need a customer's browser session, so the SDK doesn't wrap them.

Hosted vs local

export AUTOSYNTH_BASE_URL=http://localhost:8000   # or AutoSynth(base_url="http://localhost:8000")

Results and failures

A check ends with status set to succeeded or failed. A succeeded run has result.outcome of positive, negative or uncertain, plus confidence, findings (for example crystals with bright_pixels, and sharpness), the model name and version, config_hash, and latency_ms. Failed runs are persisted with error.code:

Registering by uri= (a file already in the deployment's store) only accepts your project's upload prefix, s3://<bucket>/uploads/<project>/...; anything else raises PermissionDeniedError with code forbidden_uri. The bytes are copied into content-addressed storage right away. If the source is missing, artifacts.register raises InvalidRequestError with code input_missing, so the problem shows up before any check is submitted.

An uncertain edge result may be escalated to a larger model. The original result stays in run.result (escalated=True), the verdict is appended to run.result_revisions, and run.effective_result.source becomes "escalation". Precedence is: accepted annotation, then the latest result revision, then the original result. run.provenance.route records which backend served the result (mock, hf or remote).

run.outcome is the effective outcome. If a reviewer annotated the run, it is the reviewer's label; otherwise it is the model's. The model's own result stays in run.result.

Errors, retries, timeouts

from autosynth import AutoSynth, ConflictError, NotFoundError, RateLimitError, APIConnectionError

client = AutoSynth(timeout=30, max_retries=3)    # timeout: seconds or an httpx2.Timeout

Retry behaviour:

• Reads, artifact registration, dataset snapshots, and submits/replays (which carry an Idempotency-Key) are retried on connection errors, timeouts, 408, 429 and 5xx.
• Append-only or once-only writes (annotations, releases, training runs, request transitions) are retried only when the server provably did not act: connection refused, 429, or 503.
• Backoff is exponential with full jitter (0.5 s, 1 s, 2 s, ... capped at 8 s). Retry-After and Retry-After-Ms take precedence, capped at 60 s.
• To make a submit safe across processes, pass your own idempotency_key=. The same key with the same body returns the original run; the same key with a different body raises ConflictError.

Async

import asyncio
from autosynth import AsyncAutoSynth, synthetic

async def main():
    async with AsyncAutoSynth() as client:
        frames = [synthetic.vial("PLN-0003", seed) for seed in range(32)]
        runs = await asyncio.gather(*(client.checks.run(f.png, "PLN-0003", priority="bulk") for f in frames))
        print(sum(r.outcome == "positive" for r in runs), "of", len(runs), "crystallized")

asyncio.run(main())

AsyncAutoSynth has the same resources and methods as AutoSynth; every call is awaited, and runs.iter() is an async iterator.

CLI

$ autosynth synth vial.png --scenario positive          # write a synthetic observation (no API call)
$ autosynth check vial.png --plan PLN-0001 --wait
registered art_94e6989b2aff3242 (3249 bytes, sha256 a8fc26cb8bbf…)
submitted  run_8b3c181d9c973e9c on PLN-0001 (online, queued)
run_8b3c181d9c973e9c  succeeded  positive  conf 1.00  vialnet@1.3.0  72 ms
  plan PLN-0001 · artifact art_94e6989b2aff3242 · input sha256 a8fc26cb8bbf… · online
  findings: crystals 1.00 (bright_pixels 155) · sharpness 1.03
  events: queued → running → succeeded
$ autosynth runs list --status failed --limit 10
$ autosynth runs get run_8b3c181d9c973e9c
$ autosynth runs annotate run_8b3c181d9c973e9c uncertain --comment "glare on the vial wall"
$ autosynth runs replay run_8b3c181d9c973e9c --wait
$ autosynth lineage run_8b3c181d9c973e9c
run_8b3c181d9c973e9c run/perception · positive
├─ ← input_of art_94e6989b2aff3242 artifact/observation
├─ → produced art_c0bc71b9b9fa8344 artifact/result
└─ → replay_of run_babceea49588cae6 run/perception · positive
$ autosynth plans
$ autosynth health

Every command accepts --json, --base-url and --api-key, and reads the same environment variables as the client. Exit codes:

• 0: success.
• 1: the check ran but failed (for example image_quality), or the wait timed out.
• 2: an API, network, or usage error.

python -m autosynth works too.

API reference

All methods exist on both AutoSynth and AsyncAutoSynth. Every method returns a pydantic model from autosynth.models.

Notes:

• source is a file path or bytes. Re-registering the same bytes returns the same artifact.
• Models keep fields they don't recognise (extra="allow"), so a newer server doesn't break an older SDK.

Synthetic observations

autosynth.synthetic.vial(plan_ref, seed, scenario) returns a reproducible 64×64 PNG plus capture metadata. It is byte-identical to the frames the platform's robot mock produces. The hosted mock model answers each scenario as follows:

metadata["truth"] holds the later lab-confirmed outcome, which evaluation.get(reference="truth") scores against.

Examples

• examples/quickstart.py: the 10-line version.
• examples/chemist_demo.py: every success, uncertain, negative, and failure case; idempotent resubmits; reviewer correction; replay; lineage; an async bulk batch; datasets and evaluation. Exits non-zero if anything deviates from what's expected.
• notebooks/chemist_workflow.ipynb: the same workflow, executed against a local compose stack.

Compatibility

warp_sdk.WarpClient, the monorepo's pre-0.1 flat client (register_artifact, submit_check, get_run, wait, ...), ships in the same wheel as a thin deprecated wrapper over AutoSynth (available as .client).

Development

cd sdk
uv venv && uv pip install -e . pytest anyio mypy
pytest                                              # unit tests (httpx2.MockTransport); integration tests skip
AUTOSYNTH_INTEGRATION_URL=http://localhost:8000 AUTOSYNTH_INTEGRATION_KEY=... pytest -m integration
mypy --strict autosynth
uv build && uvx twine check dist/*

Releases go out by pushing a tag sdk-vX.Y.Z (the version must match pyproject.toml). The sdk-publish GitHub workflow builds, tests, and publishes to PyPI through trusted publishing. The terminal demo is recorded with docs/demo.sh.

License

MIT

Training lifecycle and experiments (0.2.0)

Both sync and async clients support these helpers. Recording a training run does not launch a GPU job; your training process creates the record, then updates it with measured metrics and a registered checkpoint.

run = client.training_runs.create(
    "ppo", dataset_artifact_ids=[dataset_id], config_hash=config_hash,
    status="running", model_name="pour-policy",
)
run = client.training_runs.get(run.training_run_id)
client.training_runs.update(run.training_run_id, status="succeeded",
                            metrics={"eval_return": measured_return},
                            checkpoint_artifact_id=checkpoint_id)
# Available only when the run has readable recorded job logs:
print(client.training_runs.logs(run.training_run_id, tail=100))
for line in client.training_runs.stream_logs(run.training_run_id):
    print(line)

Log reads use the same project-scoped authentication. Missing/expired jobs return typed API errors. Closing the stream stops following; streaming does not automatically reconnect or replay logs.

sample = client.experiments.create(plan_ref, "SAMPLE-001", location="bay-2")
client.experiments.add_step(sample.experiment_id, "material_check", detail="Materials verified")
check = client.checks.run(image_bytes, plan_ref, experiment_id=sample.experiment_id)
sample = client.experiments.get(sample.experiment_id)

experiments.list() filters by status, finding or review state. Capture retries use checks.submit(..., experiment_id=..., retry_of=failed_run_id). Manual lab steps record events; they do not command a physical robot. Review and customer release remain separate actions.

Chemistry evidence, datasets and simulation learning (0.3.0)

Use experiments.record_report with a typed ScienceReportInput to record precipitation, cloudiness, phase separation, colour change, purity, identity or yield evidence. Attach an artifact first, declare real/synthetic/unknown origin, and supply the method and finding. Purity/yield require a percentage. experiments.reports retrieves records; experiments.review_report appends acceptance/rejection with a comment. Review does not mean independent validation or customer release.

datasets.library lists sources and validation limits. datasets.samples("marco") returns the 80-example manifest and project-scoped preview artifact IDs.

policy_cycles.start queues a simulated PPO cycle; list shows its stages. review_data approves/rejects collected episodes; decide approves/rejects a candidate after fixed simulator gates. rollback restores the previous simulator baseline. No physical robot deployment is exposed. AsyncAutoSynth supports all these methods.

Complete UI and Python walkthroughs

Escalation revisions and forward-compatible models (0.3.1)

When the edge model answers uncertain and the cascade escalates, run.escalation_id is set and the larger model's verdict is appended to run.result_revisions; run.effective_result.source becomes "escalation" (a reviewer annotation still takes precedence). run.result.escalated and run.provenance.route (mock, hf or remote) say how a result was served, and released results carry their basis (model, escalation or review). Enum-like response fields accept values newer servers add, so an unknown status no longer fails validation.