Collecting Results for Every Run in a Job Log with Python

A scheduled job that starts Clay runs with the CLI leaves a trail of run ids in its log, and nothing else keeps the results together. This Python script reads such a log, fetches each distinct run over REST whether it was inline or bulk, and merges every row into one file. In the test, 5 log lines pointed at 4 distinct runs, 3 were readable, and 6 rows were collected.

What you will build

A script that scans job.log for finished lines, removes repeated run ids, picks the right results endpoint for each run, downloads bulk result files, and writes collected.jsonl with the run id on every row.

job.log  (JSON lines, some lines are not JSON)
    ↓ keep event == "finished", dedupe the run ids
for each run id:
    GET /routines/run/{id}/results
        404 "is a batch run"  ->  GET /routines/run-batch/{id}/results -> result_url -> download
        404 otherwise         ->  record as not_found
    ↓
collected.jsonl  (run, id, status, result)

job.log  (JSON lines, some lines are not JSON)
    ↓ keep event == "finished", dedupe the run ids
for each run id:
    GET /routines/run/{id}/results
        404 "is a batch run"  ->  GET /routines/run-batch/{id}/results -> result_url -> download
        404 otherwise         ->  record as not_found
    ↓
collected.jsonl  (run, id, status, result)

AI Prompt

Write a Python script that collects the results of every Clay run listed in a job log.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key". Reads only, so it spends no credits.
- The log has one JSON object per line, and lines that do not parse must be skipped. A line with "event": "finished" carries the run id in "run". The same run id can appear more than once, so fetch each distinct id once.
- The log does not say whether a run was inline or bulk. Request GET /routines/run/{id}/results first. If it returns 404 with a message containing "is a batch run", request GET /routines/run-batch/{id}/results instead. Any other 404, for example "Routine run '<id>' not found", means the id is unknown, so record it as not_found and keep going.
- Inline results are in data. The endpoint returns 20 rows per page unless `?limit=` is given (1 to 100) and adds a "cursor" when more remain, so request `?limit=100` and follow `&cursor=<value>` until it is absent. A bulk response has status, total, finished and result_url, with no rows. Download result_url without the Clay key, and never print it. Each downloaded line is { id, status, result }.
- A run whose status is still "in_progress" is recorded as in_progress and skipped.
- Write every row to collected.jsonl with the run id added. Print one summary line for each run and one total line.
- Run the verification step below before finishing.

Write a Python script that collects the results of every Clay run listed in a job log.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key". Reads only, so it spends no credits.
- The log has one JSON object per line, and lines that do not parse must be skipped. A line with "event": "finished" carries the run id in "run". The same run id can appear more than once, so fetch each distinct id once.
- The log does not say whether a run was inline or bulk. Request GET /routines/run/{id}/results first. If it returns 404 with a message containing "is a batch run", request GET /routines/run-batch/{id}/results instead. Any other 404, for example "Routine run '<id>' not found", means the id is unknown, so record it as not_found and keep going.
- Inline results are in data. The endpoint returns 20 rows per page unless `?limit=` is given (1 to 100) and adds a "cursor" when more remain, so request `?limit=100` and follow `&cursor=<value>` until it is absent. A bulk response has status, total, finished and result_url, with no rows. Download result_url without the Clay key, and never print it. Each downloaded line is { id, status, result }.
- A run whose status is still "in_progress" is recorded as in_progress and skipped.
- Write every row to collected.jsonl with the run id added. Print one summary line for each run and one total line.
- Run the verification step below before finishing.

Prerequisites

Python 3.10+ with requests
A Clay Public API key
A job log with run ids. The sample log below uses runs created by the CLI earlier in this batch.

1. Set up the project

mkdir clay-collect-runs && cd clay-collect-runs
python3 -m venv .venv && source .venv/bin/activate
pip install requests
export CLAY_API_KEY=clay_scoped_...

mkdir clay-collect-runs && cd clay-collect-runs
python3 -m venv .venv && source .venv/bin/activate
pip install requests
export CLAY_API_KEY=clay_scoped_...

Create job.log. It holds a repeated id, a stray non-JSON line, a skipped line, and one id that does not exist:

{"at":"2026-10-13T00:48:49Z","event":"finished","run":"run_0tmtmc48DekZmqUQW7w","status":"complete","total":2,"finished":2}
{"at":"2026-10-13T00:49:02Z","event":"skipped","reason":"balance 12 is below 100"}
{"at":"2026-10-13T02:00:27Z","event":"finished","run":"run_0tmtmbrDa76RUSPoyUG","status":"complete","total":1,"finished":1}
not json, a stray line from another program
{"at":"2026-10-13T02:00:30Z","event":"finished","run":"run_0tmtmbrDa76RUSPoyUG","status":"complete","total":1,"finished":1}
{"at":"2026-10-13T02:01:10Z","event":"finished","run":"run_0tmtpkfyYgu7xpiSwWf","status":"complete","total":3,"finished":3}
{"at":"2026-10-13T02:02:00Z","event":"finished","run":"run_does_not_exist","status":"complete","total":1,"finished":1}

{"at":"2026-10-13T00:48:49Z","event":"finished","run":"run_0tmtmc48DekZmqUQW7w","status":"complete","total":2,"finished":2}
{"at":"2026-10-13T00:49:02Z","event":"skipped","reason":"balance 12 is below 100"}
{"at":"2026-10-13T02:00:27Z","event":"finished","run":"run_0tmtmbrDa76RUSPoyUG","status":"complete","total":1,"finished":1}
not json, a stray line from another program
{"at":"2026-10-13T02:00:30Z","event":"finished","run":"run_0tmtmbrDa76RUSPoyUG","status":"complete","total":1,"finished":1}
{"at":"2026-10-13T02:01:10Z","event":"finished","run":"run_0tmtpkfyYgu7xpiSwWf","status":"complete","total":3,"finished":3}
{"at":"2026-10-13T02:02:00Z","event":"finished","run":"run_does_not_exist","status":"complete","total":1,"finished":1}

2. Implement the script

The shared helper, clay.py:

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 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")

The collector, collect_runs.py:

import json
import sys

import requests

from clay import session, BASE

LOG = sys.argv[1] if len(sys.argv) > 1 else "job.log"


def fetch(run_id):
    """Return (rows, kind) for one run, trying the inline endpoint first and the batch endpoint when told to."""
    res = session.get(f"{BASE}/routines/run/{run_id}/results?limit=100", timeout=60)
    kind = "inline"
    if res.status_code == 404 and "is a batch run" in res.json().get("message", ""):
        res = session.get(f"{BASE}/routines/run-batch/{run_id}/results", timeout=60)
        kind = "bulk"
    if res.status_code == 404:
        return None, "not_found"
    res.raise_for_status()
    page = res.json()
    if page["status"] == "in_progress":
        return None, "in_progress"
    if kind == "bulk":
        text = requests.get(page["result_url"], timeout=60).text  # presigned, so no Clay key
        return [json.loads(line) for line in text.splitlines() if line], kind
    rows = page.get("data", [])
    while page.get("cursor"):
        page = session.get(f"{BASE}/routines/run/{run_id}/results?limit=100&cursor={page['cursor']}", timeout=60).json()
        rows += page.get("data", [])
    return rows, kind


def main():
    run_ids = []
    for line in open(LOG):
        try:
            entry = json.loads(line)
        except json.JSONDecodeError:
            continue  # a log can hold other text
        if entry.get("event") == "finished" and entry.get("run"):
            run_ids.append(entry["run"])
    print(json.dumps({"log_lines_with_runs": len(run_ids)}))

    merged, summary = [], []
    for rid in dict.fromkeys(run_ids):  # drop repeats, keep order
        rows, kind = fetch(rid)
        summary.append({"run": rid, "kind": kind, "rows": None if rows is None else len(rows)})
        for r in rows or []:
            merged.append({"run": rid, "id": r["id"], "status": r["status"], "result": r.get("result")})
    for s in summary:
        print(json.dumps(s))
    with open("collected.jsonl", "w") as f:
        for m in merged:
            f.write(json.dumps(m) + "\n")
    print(json.dumps({"runs": len(summary), "rows_collected": len(merged), "unreadable": sum(1 for s in summary if


if __name__ == "__main__":
    main()

import json
import sys

import requests

from clay import session, BASE

LOG = sys.argv[1] if len(sys.argv) > 1 else "job.log"


def fetch(run_id):
    """Return (rows, kind) for one run, trying the inline endpoint first and the batch endpoint when told to."""
    res = session.get(f"{BASE}/routines/run/{run_id}/results?limit=100", timeout=60)
    kind = "inline"
    if res.status_code == 404 and "is a batch run" in res.json().get("message", ""):
        res = session.get(f"{BASE}/routines/run-batch/{run_id}/results", timeout=60)
        kind = "bulk"
    if res.status_code == 404:
        return None, "not_found"
    res.raise_for_status()
    page = res.json()
    if page["status"] == "in_progress":
        return None, "in_progress"
    if kind == "bulk":
        text = requests.get(page["result_url"], timeout=60).text  # presigned, so no Clay key
        return [json.loads(line) for line in text.splitlines() if line], kind
    rows = page.get("data", [])
    while page.get("cursor"):
        page = session.get(f"{BASE}/routines/run/{run_id}/results?limit=100&cursor={page['cursor']}", timeout=60).json()
        rows += page.get("data", [])
    return rows, kind


def main():
    run_ids = []
    for line in open(LOG):
        try:
            entry = json.loads(line)
        except json.JSONDecodeError:
            continue  # a log can hold other text
        if entry.get("event") == "finished" and entry.get("run"):
            run_ids.append(entry["run"])
    print(json.dumps({"log_lines_with_runs": len(run_ids)}))

    merged, summary = [], []
    for rid in dict.fromkeys(run_ids):  # drop repeats, keep order
        rows, kind = fetch(rid)
        summary.append({"run": rid, "kind": kind, "rows": None if rows is None else len(rows)})
        for r in rows or []:
            merged.append({"run": rid, "id": r["id"], "status": r["status"], "result": r.get("result")})
    for s in summary:
        print(json.dumps(s))
    with open("collected.jsonl", "w") as f:
        for m in merged:
            f.write(json.dumps(m) + "\n")
    print(json.dumps({"runs": len(summary), "rows_collected": len(merged), "unreadable": sum(1 for s in summary if s["rows"] is None)}))


if __name__ == "__main__":
    main()

3. Run

python3 collect_runs.py job.log
wc -l collected.jsonl

python3 collect_runs.py job.log
wc -l collected.jsonl

4. Check the output

Output on 2026-10-12:

{"log_lines_with_runs": 5}
{"run": "run_0tmtmc48DekZmqUQW7w", "kind": "bulk", "rows": 2}
{"run": "run_0tmtmbrDa76RUSPoyUG", "kind": "inline", "rows": 1}
{"run": "run_0tmtpkfyYgu7xpiSwWf", "kind": "bulk", "rows": 3}
{"run": "run_does_not_exist", "kind": "not_found", "rows": null}
{"runs": 4, "rows_collected": 6, "unreadable": 1}

{"log_lines_with_runs": 5}
{"run": "run_0tmtmc48DekZmqUQW7w", "kind": "bulk", "rows": 2}
{"run": "run_0tmtmbrDa76RUSPoyUG", "kind": "inline", "rows": 1}
{"run": "run_0tmtpkfyYgu7xpiSwWf", "kind": "bulk", "rows": 3}
{"run": "run_does_not_exist", "kind": "not_found", "rows": null}
{"runs": 4, "rows_collected": 6, "unreadable": 1}

What to check is a set of properties. log_lines_with_runs counts the finished lines, 5 here, while runs counts distinct ids, 4. Each readable run reports inline or bulk and a row count. The unknown id is reported as not_found and does not stop the script. rows_collected equals the sum of the row counts and the number of lines in collected.jsonl. The domains inside the rows depend on the routine's answers that day.

How it works

The two result endpoints answer for different kinds of run, and each one rejects the other's ids with a 404. The inline endpoint's rejection message names the batch endpoint, so the script treats that one message as a signal to switch. A different 404 message means the id is unknown.

Bulk rows are a second request, to a presigned URL in result_url. The request carries no Clay key, because the URL has its own authorization. Deduplicating ids first matters for bulk runs, since each repeat would otherwise download the file again.

Common issues

Presigned result URLs are short-lived

The result_url expires after roughly 15 minutes, so the script requests it fresh for each run and downloads it straight away. A stored result_url from an earlier collection will not work.

The log is only as complete as the job that wrote it

The sample line with run_does_not_exist shows what a stale or mistyped id looks like. A job that crashed before writing its finished line leaves a run that this script never sees. Log the run id when the run starts, as well as when it finishes.

Inline runs are paged

The inline endpoint returns 20 rows per page by default. A 25-row run read without limit loses 5 rows, so the collector asks for limit=100 and follows the cursor. It collected all 25 rows of a test run this way, where the log pointed at that run.

Lines that are not JSON are common in real logs

The stray text line is skipped rather than crashing the script. Counting skipped lines and reporting them would be a reasonable extension.

A run can still be in progress

A run read while in_progress has no rows yet, and the script records it as such so that a later pass can pick it up.

Next steps

verification:
  status: verified
  tested_at: "2026-10-12"
  product_version: "public/v0"
  command: "python3 collect_runs.py job.log"
  expected_result: "Five finished lines give four distinct runs, an inline and two bulk runs are read, the unknown id is reported as not_found, and rows_collected equals the lines in collected.jsonl."

verification:
  status: verified
  tested_at: "2026-10-12"
  product_version: "public/v0"
  command: "python3 collect_runs.py job.log"
  expected_result: "Five finished lines give four distinct runs, an inline and two bulk runs are read, the unknown id is reported as not_found, and rows_collected equals the lines in collected.jsonl."