Resuming an Interrupted Batch from a State File over REST with Python

state.json  ->  rows with no entry yet
    ↓
chunk of 2  ->  POST /routines/{id}/run  ->  poll results
    ↓
write state.json (atomic rename)
    ↓
next chunk ... (process may be killed here)

Implement a Python script that processes a list through a Clay routine in chunks and can resume after being killed.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- Use the "Company Domain" routine, function:t_0tjqm9rUjfGY4MunGDM. Its input is "Company Name" and its result is { "Domain": "<domain>" }. It costs 1 credit per row.
- Use the row's position in the list as its id, so ids stay stable between runs.
- Run each chunk with POST /routines/{id}/run and body { items: [{ id, inputs }] }, poll GET /routines/run/{routine_run_id}/results until status is not "in_progress", and match rows by id.
- Record a row as done only when its status is "complete" and the result has a Domain. Write the state file after every chunk, using a temporary file and an atomic rename.
- On start, skip rows that are already in the state file.
- For the test, an environment variable CRASH_AFTER_CHUNKS stops the process with os._exit(137) after that many chunks, to simulate a kill.
- Read GET /credits/balance before and after each invocation and report the credits spent.
- Run the verification step below before finishing.

mkdir clay-resume && cd clay-resume
python3 -m venv .venv && source .venv/bin/activate
pip install requests

export CLAY_API_KEY=clay_scoped_...

import os
import time

import requests

BASE = "https://api.clay.com/public/v0"

session = requests.Session()
session.headers.update({"clay-api-key": os.environ["CLAY_API_KEY"], "content-type": "application/json"})


def clay(path, method="GET", body=None):
    res = session.request(method, f"{BASE}{path}", json=body, timeout=60)
    if not res.ok:
        raise RuntimeError(f"{res.status_code} {path}: {res.text}")
    return res.json()


def balance():
    return clay("/credits/balance")["balance"]


def all_results(run_id):
    """Fetch every row. The endpoint returns 20 rows per page unless limit is set, and adds a cursor when more remain."""
    page = clay(f"/routines/run/{run_id}/results?limit=100")
    rows = list(page.get("data", []))
    while page.get("cursor"):
        page = clay(f"/routines/run/{run_id}/results?limit=100&cursor={page['cursor']}")
        rows.extend(page.get("data", []))
    page["data"] = rows
    return page


def run_routine(routine_id, rows, timeout_s=600):
    started = clay(f"/routines/{routine_id}/run", "POST", {"items": rows})
    run_id = started["routine_run_id"]
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        time.sleep(5)
        results = all_results(run_id)
        if results["status"] != "in_progress":
            return results
    raise RuntimeError(f"run {run_id} still in progress after {timeout_s} seconds")

import json
import os
import sys
import time
from pathlib import Path

from clay import balance, run_routine

COMPANY_DOMAIN = "function:t_0tjqm9rUjfGY4MunGDM"
COMPANIES = ["Notion", "Figma", "Linear", "Vercel", "Supabase"]
CHUNK = 2
STATE = Path("state.json")


def load_state():
    return json.loads(STATE.read_text()) if STATE.exists() else {"done": {}}


def save_state(state):
    tmp = STATE.with_suffix(".tmp")
    tmp.write_text(json.dumps(state, indent=2))
    tmp.replace(STATE)  # atomic on POSIX, so a crash never leaves half a file


def main():
    crash_after = int(os.environ.get("CRASH_AFTER_CHUNKS", "0"))
    state = load_state()
    todo = [i for i in range(len(COMPANIES)) if str(i) not in state["done"]]
    print(json.dumps({"rows": len(COMPANIES), "already_done": len(state["done"]), "to_run": len(todo)}))

    before = balance()
    chunks_run = 0
    for start in range(0, len(todo), CHUNK):
        ids = todo[start:start + CHUNK]
        results = run_routine(COMPANY_DOMAIN, [{"id": str(i), "inputs": {"Company Name": COMPANIES[i]}} for i in ids])
        for item in results.get("data", []):
            domain = (item.get("result") or {}).get("Domain")
            if item["status"] == "complete" and domain:
                state["done"][item["id"]] = domain
        save_state(state)
        chunks_run += 1
        print(json.dumps({"chunk": ids, "done_total": len(state["done"])}))
        if crash_after and chunks_run >= crash_after:
            sys.stdout.flush()  # os._exit skips the normal flush, so print first
            os._exit(137)  # simulates the process being killed between chunks

    time.sleep(5)
    print(json.dumps({"finished": len(state["done"]) == len(COMPANIES), "credits_spent": before - balance()}))


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as err:
        print(err, file=sys.stderr)
        sys.exit(1)

rm -f state.json
CRASH_AFTER_CHUNKS=1 python3 resumable.py; echo "exit=$?"
python3 resumable.py; echo "exit=$?"

{"rows": 5, "already_done": 0, "to_run": 5}
{"chunk": [0, 1], "done_total": 2}
exit=137
{
  "done": {
    "1": "figma.com",
    "0": "notion.com"
  }
}{"rows": 5, "already_done": 2, "to_run": 3}
{"chunk": [2, 3], "done_total": 4}
{"chunk": [4], "done_total": 5}
{"finished": true, "credits_spent": 3.0}
exit=0

verification:
  status: verified
  tested_at: "2026-10-12"
  product_version: "public/v0"
  command: "rm -f state.json; CRASH_AFTER_CHUNKS=1 python3 resumable.py; python3 resumable.py"
  expected_result: "The first invocation exits 137 with 2 rows saved; the second runs only the 3 missing rows and reports finished true."