Python client

Use the Python client to submit a run (one task you send to Millwork; the API calls it an execution), follow its progress, and read its receipt. It includes a synchronous Solver and an asynchronous AsyncSolver. Both use the same requests and organization API key as the REST API. The published 0.1.2 wheel can register and test a public output check (a verifier in the API) and manage a protected check through private key entry.

Install

Requires CPython 3.11 to 3.14 and an organization API key.

python -m pip install millwork-solver==0.1.2

The package is millwork-solver on PyPI. Import it as millwork_solver. For output checks, use the generated REST lifecycle reference for operation IDs and exact fields, and the verifier-dock contract for the endpoint request and response. Keep protected endpoint keys on Millwork’s private entry page. For a protected endpoint, use the published Python connection flow. The key is entered only on Millwork’s private page, never in a Python argument.

Make a test run

First configure your organization key in the same terminal. This example submits a test run and reads its receipt without calling a model. Solver() reads SOLVERAPI_API_KEY and SOLVERAPI_BASE_URL from the environment. You can also pass them as api_key and base_url.

export SOLVERAPI_BASE_URL="https://api.getmillwork.dev/v1"

Save this as example.py:

import time
import uuid

from millwork_solver import Solver

FINAL_STATUSES = {"completed", "failed", "cancelled", "expired"}

with Solver() as client:
    submitted = client.request(
        "postExecutions",
        body={
            "mode": "echo",
            "task": {"objective": "Check the Millwork API lifecycle."},
            "policy": {
                "data_classes": ["public"],
                "budget": {"max_cost_usd": 1, "max_runtime_s": 30},
            },
        },
        idempotency_key=str(uuid.uuid4()),
    )
    execution_id = submitted["execution_id"]
    print("Run ID:", execution_id)
    status = submitted["status"]
    deadline = time.monotonic() + 90
    while status not in FINAL_STATUSES:
        if time.monotonic() >= deadline:
            raise TimeoutError(f"Keep run ID {execution_id} and inspect it later.")
        time.sleep(1)
        current = client.request(
            "getExecutionsByExecutionId",
            path_parameters={"executionId": execution_id},
        )
        status = current["status"]

    receipt = client.request(
        "getReceiptsByExecutionId",
        path_parameters={"executionId": execution_id},
    )
    print(status, receipt["mode"])

Run it in the same Python environment where you installed the client:

python example.py

Expected result: completed echo. For another outcome, keep the run ID and follow Errors and retries. To get a model answer, continue with Run a model. AsyncSolver has the same calls: await client.request(...), async for over pages and events, and async with or aclose() to close.

Register and test a public output check

Set VERIFIER_URL to your running endpoint’s HTTPS address. The endpoint must implement the output-check contract. Save this as output_check.py and run it with python output_check.py in the same environment where you installed the pinned version above:

import os
import uuid
from millwork_solver import Solver

with Solver() as client:
    registered = client.request("postVerifiers", body={
        "display_name": "My output check", "version": "1.0.0", "kind": "endpoint",
        "endpoint": {"url": os.environ["VERIFIER_URL"], "auth_ref": ""},
        "input_data_classes": ["public"],
        "scoring": {"correctness": "boolean_anchors", "quality": "scalar_0_1"},
    }, idempotency_key=str(uuid.uuid4()))
    verifier_id = registered["verifier_id"]
    report = client.request("postVerifiersByVerifierIdTest",
        path_parameters={"verifierId": verifier_id},
        idempotency_key=str(uuid.uuid4()))
    print(verifier_id, report)

Keep the returned verifier_id. This test observes endpoint health; it does not authorize a paid run. AsyncSolver accepts the same operation IDs through await client.request(...). The shared connection guide owns deployment, protected key entry, run approval, and receipt inspection. For the endpoint’s exact request and response fields, read the verifier-dock contract. The published 0.1.2 wheel has these verifier operation IDs. The generated REST lifecycle reference may list newer REST operations that this wheel cannot call by request().

What you can do	Python `request()` operation ID
Register	`postVerifiers`
List	`getVerifiers`
Read	`getVerifiersByVerifierId`
Update	`patchVerifiersByVerifierId`
Test	`postVerifiersByVerifierIdTest`
Retire	`deleteVerifiersByVerifierId`

For get, patch, test, and delete, pass path_parameters={"verifierId": verifier_id}.

Test a public check with AsyncSolver

The published wheel also supports asynchronous calls to its public verifier operations. Set VERIFIER_ID to a public check you registered above. Save this as test_check_async.py and run python test_check_async.py:

import asyncio
import os
import uuid
from millwork_solver import AsyncSolver

async def main():
    async with AsyncSolver() as client:
        report = await client.request(
            "postVerifiersByVerifierIdTest",
            path_parameters={"verifierId": os.environ["VERIFIER_ID"]},
            idempotency_key=str(uuid.uuid4()),
        )
        print(report)

asyncio.run(main())

This tests endpoint health without submitting a model run. Protected keys still go through the private entry flow.

Use an output check on a run

First save a model and keep its arm_id. Connect and test a public check above, then keep its verifier_id. A live run has a platform fee when accepted and may incur model usage. Review the exact task, model, limits, check, and charge before authorizing this run; the shared run procedure owns approval and recovery. Set ARM_ID, VERIFIER_ID, MAX_COST_USD, MAX_RUNTIME_S, and a unique RUN_REQUEST_KEY in your terminal. Record that key with the exact request before submitting; reuse it only for an uncertain retry of the same request. Choose positive limits for the task you reviewed; the model budget is separate from the platform fee. Save this as checked_run.py and run python checked_run.py only for the run you have authorized:

import os
import math
from millwork_solver import Solver

arm_id = os.environ["ARM_ID"]
verifier_id = os.environ["VERIFIER_ID"]
request_key = os.environ["RUN_REQUEST_KEY"]
max_cost_usd = float(os.environ["MAX_COST_USD"])
max_runtime_s = int(os.environ["MAX_RUNTIME_S"])
if not math.isfinite(max_cost_usd) or max_cost_usd <= 0 or max_runtime_s <= 0:
    raise ValueError("Choose positive MAX_COST_USD and MAX_RUNTIME_S limits")
with Solver() as client:
    run = client.request("postExecutions", body={
        "mode": "live",
        "task": {"objective": "Reply with the single word: ok"},
        "policy": {
            "data_classes": ["public"],
            "budget": {"max_cost_usd": max_cost_usd, "max_runtime_s": max_runtime_s},
            "on_eval": [],
        },
        "routing": {"required_arm_id": arm_id},
        "verifier_id": verifier_id,
    }, idempotency_key=request_key)
    print(run["execution_id"], run["status"])

Expected result: an execution_id and an initial status. Keep the request key unchanged if you need to recover an uncertain submission. Read the result and receipt by that ID; a completed run can still have a rejected or unavailable check.

Connect, replace, or restore a protected output check

The published 0.1.2 wheel includes verifier_connection. Register your endpoint without a key as shown above and keep its verifier_id.

Create an intent, send only its continue_url to the intended person for private key entry, and continue the connection until the server confirms done. A new intent follows the same path when you replace or restore a key. This connection does not authorize a paid run.

Check the wheel installed in the Python environment you will actually run. Save this as check_capability.py and run python check_capability.py. It uses no API key and makes no network request:

import json
from importlib.metadata import version
from importlib.resources import files
from millwork_solver import Solver

installed = version("millwork-solver")
snapshot = json.loads(files("millwork_solver").joinpath("_contract.json").read_text())
operations = {item["operation_id"] for item in snapshot["operations"]}
with Solver(api_key="capability-probe", base_url="https://api.getmillwork.dev/v1") as client:
    protected = hasattr(client, "verifier_connection")
print("installed:", installed)
print("public verifier test:", "postVerifiersByVerifierIdTest" in operations)
print("protected lifecycle:", protected)

For 0.1.2, the last line prints True. If it prints False, check that this script uses the Python environment where you installed 0.1.2 before attempting a connection. Set VERIFIER_ID to the registered endpoint, MILLWORK_APP_ORIGIN to the trusted origin of your Millwork private-entry app, and STOP_DAYS to the chosen 30, 90, 180, or 365-day Millwork stop period. The origin must come from your deployment configuration, not from the returned URL. For the hosted app:

export VERIFIER_ID="<registered verifier ID>"
export MILLWORK_APP_ORIGIN="https://app.getmillwork.dev"
export STOP_DAYS="30"

This example runs on the intended person’s computer. A server app must hand the URL to that person through its private channel instead of opening its own browser. Save this as connect_check.py and run python connect_check.py:

import os
import uuid
import webbrowser
from urllib.parse import urlsplit

from millwork_solver import Solver

verifier_id = os.environ["VERIFIER_ID"]
stop_days = int(os.environ["STOP_DAYS"])
trusted = urlsplit(os.environ["MILLWORK_APP_ORIGIN"])
if trusted.scheme != "https" or not trusted.netloc or trusted.path not in {"", "/"}:
    raise ValueError("Configure a trusted HTTPS app origin")

with Solver() as client:
    connection = client.verifier_connection
    intent = connection.create_intent(
        verifier_id, {"kind": "preset_days", "days": stop_days},
        idempotency_key=str(uuid.uuid4()),
    )
    private_url = urlsplit(intent["continue_url"])
    if (private_url.scheme, private_url.netloc) != (trusted.scheme, trusted.netloc):
        raise ValueError("Private-entry URL has an unexpected origin")
    operation_key = intent["intent_id"]
    print("Connection operation:", operation_key)
    webbrowser.open(intent["continue_url"])
    for _ in range(5):
        input("Finish private key entry in the browser, then press Enter here: ")
        outcome = connection.continue_connection(
            verifier_id, operation_key=operation_key,
            idempotency_key=str(uuid.uuid4()),
        )
        print("Connection state:", outcome.state, "next:", outcome.next_action)
        if outcome.state == "done":
            break
        if outcome.next_action not in {"enter_key", "resume"}:
            raise RuntimeError("Inspect this operation before another change")
    else:
        raise TimeoutError("Keep the operation key and inspect it later")
    print("Read-backed status:", connection.inspect(
        verifier_id, operation_key=operation_key,
    )["status"])

Keep the operation key for recovery. enter_key means the person has not finished private entry; resume means keep that key and retry with a fresh request key. start_again means correct the endpoint or key and create a new intent. inspect means read the operation before another change. A refusal reports a sanitized status and title; 401 or 403 raises ProblemError. Only read-backed done confirms the connection. Do not log the private URL or endpoint key. After the owner separately confirms disconnection, use a new operation key. This snippet is not part of connect_check.py:

import os
import uuid
from millwork_solver import Solver

if input("Disconnect this check? Type DISCONNECT to confirm: ") != "DISCONNECT":
    raise SystemExit("No change made")
operation_key = str(uuid.uuid4())
print("Disconnect operation:", operation_key)
with Solver() as client:
    outcome = client.verifier_connection.disconnect(
        os.environ["VERIFIER_ID"], operation_key=operation_key,
        idempotency_key=str(uuid.uuid4()), confirmed=True,
    )
    print(outcome.state, outcome.next_action)

If the result asks for resume, keep the same disconnect operation key for the next attempt. Millwork stopping use of a key does not revoke it at your endpoint.

Async protected connection

AsyncSolver.verifier_connection is an AsyncVerifierConnection. Its six coroutine methods have the same arguments and outcomes as the synchronous methods; await each call:

Synchronous `VerifierConnection`	Asynchronous `AsyncVerifierConnection`
`.create_intent()`	`await .create_intent()`
`.continue_connection()`	`await .continue_connection()`
`.test_and_promote()`	`await .test_and_promote()`
`.inspect()`	`await .inspect()`
`.disconnect()`	`await .disconnect()`
`.revoke()`	`await .revoke()`

The normal path uses continue_connection(); it handles the lower-level test_and_promote() call and reads back the result. disconnect() similarly handles revoke(). This runnable example starts an async intent and continues it after private entry. Set the same environment values as above and save it as connect_check_async.py:

import asyncio
import os
import uuid
import webbrowser
from urllib.parse import urlsplit

from millwork_solver import AsyncSolver

async def main():
    verifier_id = os.environ["VERIFIER_ID"]
    stop_days = int(os.environ["STOP_DAYS"])
    trusted = urlsplit(os.environ["MILLWORK_APP_ORIGIN"])
    if trusted.scheme != "https" or not trusted.netloc or trusted.path not in {"", "/"}:
        raise ValueError("Configure a trusted HTTPS app origin")
    async with AsyncSolver() as client:
        connection = client.verifier_connection
        intent = await connection.create_intent(
            verifier_id, {"kind": "preset_days", "days": stop_days},
            idempotency_key=str(uuid.uuid4()),
        )
        private_url = urlsplit(intent["continue_url"])
        if (private_url.scheme, private_url.netloc) != (trusted.scheme, trusted.netloc):
            raise ValueError("Private-entry URL has an unexpected origin")
        operation_key = intent["intent_id"]
        print("Connection operation:", operation_key)
        webbrowser.open(intent["continue_url"])
        for _ in range(5):
            input("Finish private key entry in the browser, then press Enter here: ")
            outcome = await connection.continue_connection(
                verifier_id, operation_key=operation_key,
                idempotency_key=str(uuid.uuid4()),
            )
            print(outcome.state, outcome.next_action)
            if outcome.state == "done":
                break
            if outcome.next_action not in {"enter_key", "resume"}:
                raise RuntimeError("Inspect this operation before another change")
        else:
            raise TimeoutError("Keep the operation key and inspect it later")
        print((await connection.inspect(verifier_id, operation_key=operation_key))["status"])

asyncio.run(main())

For a separately confirmed async disconnect, use the same confirmation boundary and await the result:

import asyncio
import os
import uuid
from millwork_solver import AsyncSolver

async def disconnect():
    if input("Disconnect this check? Type DISCONNECT to confirm: ") != "DISCONNECT":
        return
    operation_key = str(uuid.uuid4())
    print("Disconnect operation:", operation_key)
    async with AsyncSolver() as client:
        outcome = await client.verifier_connection.disconnect(
            os.environ["VERIFIER_ID"], operation_key=operation_key,
            idempotency_key=str(uuid.uuid4()), confirmed=True,
        )
        print(outcome.state, outcome.next_action)

asyncio.run(disconnect())

Keep the same operation key for an uncertain disconnect or a resume result.

What the client does

Both clients expose one request call: name the REST operation by its operation ID from the API reference and pass path parameters, query values, and a body. pages walks a list one page at a time, and lists of your organization’s own items, such as saved models, return the newest first. events follows a run’s events in the order they happened until a final status.

Client call	REST endpoint	What it does
`request("postExecutions", body=..., idempotency_key=...)`	`POST /v1/executions`	Submits a run.
`request("getExecutionsByExecutionId", path_parameters=...)`	`GET /v1/executions/{executionId}`	Reads the run’s status.
`events(path_parameters=...)`	`GET /v1/executions/{executionId}/events`	Yields the run’s events and stops at a final status.
`request("getExecutionsByExecutionIdResult", path_parameters=...)`	`GET /v1/executions/{executionId}/result`	Reads the model output of a completed live run.
`request("getReceiptsByExecutionId", path_parameters=...)`	`GET /v1/receipts/{executionId}`	Reads the receipt.
`request("postExecutionsByExecutionIdCancel", path_parameters=...)`	`POST /v1/executions/{executionId}/cancel`	Cancels a run.
`request("getModelCatalog")`	`GET /v1/model-catalog`	Lists the models your organization can save now.
`request("postArms", body=...)`	`POST /v1/arms`	Saves a model. The API calls a saved model an `arm`.
`pages("getArms", item_field="arms")`	`GET /v1/arms`	Lists saved models, one page per iteration.

A test run (the API calls it Echo, mode: "echo") reaches a final status and returns a receipt. It calls no model, so it has no result content and no platform fee, and reading its result is not part of the test-run flow. You own the request key. Pass it as idempotency_key when you submit a run and the client sends it as the Idempotency-Key header. The client never makes one up for you, and a submit without one is sent once. A read (GET or HEAD) that hits a network failure or a 5xx response is retried, up to two times by default, with a doubling wait that starts at 500 milliseconds. A write with a request key is retried the same way. A write without a request key is never retried by the client. A 429 is not retried either; read retry_after_seconds and wait. Change the defaults with max_retries and retry_backoff_ms on the constructor, or with SOLVERAPI_MAX_RETRIES and SOLVERAPI_RETRY_BACKOFF_MS in the environment.

Errors

Every exception is importable from millwork_solver.

Exception	When	Fields
`ProblemError`	Millwork answered with an error.	`code` (the error code, such as `unauthenticated`, `insufficient_credit`, `idempotency_conflict`, `rate_limited`, `upstream_unavailable`, `not_found`), `status`, `title`, `detail`, `instance`, `problem_type`, `retry_after_seconds`, `field_errors` (each with `field` and `message`), `retryable` (true for `5xx`), and the full `problem` object.
`NetworkError`	The request never produced an HTTP response: an unreachable host or a TLS failure. A malformed base URL raises `ValueError` when you construct the client.	`cause`, the underlying exception.
`ProtocolError`	The response was not the JSON or error shape the client expects.	Message only.
`OperationBoundaryError`	The operation ID is not one the client offers, or its path parameters do not match.	Message only.
`ClientClosedError`	A call after `close()` or `aclose()`.	Message only.

An empty list is a normal answer, not an error. For what each error code means and which are safe to retry, read Errors and retries.

API basics

Authentication, request keys, and the response shapes the client returns.

Make your first API call

Send a free test run with curl and read its receipt.

Get started

Provide models

Build

Build safely

SDK and MCP

Install

Make a test run

Register and test a public output check

Test a public check with AsyncSolver

Use an output check on a run

Connect, replace, or restore a protected output check

Async protected connection

What the client does

Errors

API basics

Make your first API call

​Install

​Make a test run

​Register and test a public output check

​Test a public check with AsyncSolver

​Use an output check on a run

​Connect, replace, or restore a protected output check

​Async protected connection

​What the client does

​Errors

API basics

Make your first API call

Install

Make a test run

Register and test a public output check

Test a public check with AsyncSolver

Use an output check on a run

Connect, replace, or restore a protected output check

Async protected connection

What the client does

Errors