Running a Nightly Clay Enrichment from Cron with Python

This example runs a Python script that calls the Clay Public API from a shell entry point, under env -i with only /usr/bin:/bin on the PATH, which approximates the environment cron provides. A lock file stopped a second copy that started while the first was running, and the first copy wrote a dated results file and one log line.

What you will build

A job directory with nightly_enrich.py, a run.sh entry point, an accounts.jsonl input, and a cron entry. The script takes a non-blocking lock, runs Company Domain over the input, writes out/<date>.jsonl, and appends a JSON line to job.log.

cron -> run.sh (loads key, calls the venv's python by absolute path)
          ↓
nightly_enrich.py
   flock(job.lock, non-blocking)  -> already held? log skipped, exit 0
   GET  /credits/balance
   POST /routines/{id}/run  ->  poll /routines/run/{id}/results (up to 10 minutes)
   write out/<date>.jsonl
   log finished { rows, failed, credits_spent }

cron -> run.sh (loads key, calls the venv's python by absolute path)
          ↓
nightly_enrich.py
   flock(job.lock, non-blocking)  -> already held? log skipped, exit 0
   GET  /credits/balance
   POST /routines/{id}/run  ->  poll /routines/run/{id}/results (up to 10 minutes)
   write out/<date>.jsonl
   log finished { rows, failed, credits_spent }

AI Prompt

Implement a Python script, meant to be started by cron, that runs a Clay routine over a JSONL file.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key", read from the CLAY_API_KEY environment variable.
- cron starts jobs with a minimal environment: PATH=/usr/bin:/bin and no virtualenv. The entry point must source a key file and call the virtualenv's python by absolute path. The system python in /usr/bin may not have the requests package.
- Take an exclusive, non-blocking lock with fcntl.flock on job.lock. If it is already held, append a "skipped" line to job.log and exit 0, so cron does not send mail for a harmless overlap.
- 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. Each line of accounts.jsonl is { "id": "<string>", "inputs": { "Company Name": "<name>" } }.
- Run it with POST /routines/{id}/run and body { items: [...] }, then poll GET /routines/run/{routine_run_id}/results every 5 seconds for up to 10 minutes, until status is not "in_progress".
- The results endpoint returns 20 rows per page unless `?limit=` is set (1 to 100), and adds a "cursor" to the response when more rows remain. Request `?limit=100` and follow the cursor with `&cursor=<value>` until the response has none, so that a list of any size is written in full.
- Write each result item to out/<UTC date>.jsonl. Results are not in input order, so keep the id on each line.
- Append one JSON line to job.log for every outcome: finished (with run_id, rows, failed ids, credits_spent), skipped, timeout, or error. Exit 1 on timeout, error, or failed rows.
- Test it with env -i HOME=<empty dir> PATH=/usr/bin:/bin so the run does not depend on the developer's shell.
- Run the verification step below before finishing.

Implement a Python script, meant to be started by cron, that runs a Clay routine over a JSONL file.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key", read from the CLAY_API_KEY environment variable.
- cron starts jobs with a minimal environment: PATH=/usr/bin:/bin and no virtualenv. The entry point must source a key file and call the virtualenv's python by absolute path. The system python in /usr/bin may not have the requests package.
- Take an exclusive, non-blocking lock with fcntl.flock on job.lock. If it is already held, append a "skipped" line to job.log and exit 0, so cron does not send mail for a harmless overlap.
- 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. Each line of accounts.jsonl is { "id": "<string>", "inputs": { "Company Name": "<name>" } }.
- Run it with POST /routines/{id}/run and body { items: [...] }, then poll GET /routines/run/{routine_run_id}/results every 5 seconds for up to 10 minutes, until status is not "in_progress".
- The results endpoint returns 20 rows per page unless `?limit=` is set (1 to 100), and adds a "cursor" to the response when more rows remain. Request `?limit=100` and follow the cursor with `&cursor=<value>` until the response has none, so that a list of any size is written in full.
- Write each result item to out/<UTC date>.jsonl. Results are not in input order, so keep the id on each line.
- Append one JSON line to job.log for every outcome: finished (with run_id, rows, failed ids, credits_spent), skipped, timeout, or error. Exit 1 on timeout, error, or failed rows.
- Test it with env -i HOME=<empty dir> PATH=/usr/bin:/bin so the run does not depend on the developer's shell.
- Run the verification step below before finishing.

Prerequisites

Python 3.10+
A Clay Public API key (clay_scoped_...)
A workspace with at least 3 credits

1. Create the job directory

mkdir -p ~/clay-cron && cd ~/clay-cron
python3 -m venv .venv
.venv/bin/pip install requests
printf '%s\n' \
  '{"id":"a","inputs":{"Company Name":"Notion"}}' \
  '{"id":"b","inputs":{"Company Name":"Figma"}}' \
  '{"id":"c","inputs":{"Company Name":"Linear"}}' > accounts.jsonl

mkdir -p ~/clay-cron && cd ~/clay-cron
python3 -m venv .venv
.venv/bin/pip install requests
printf '%s\n' \
  '{"id":"a","inputs":{"Company Name":"Notion"}}' \
  '{"id":"b","inputs":{"Company Name":"Figma"}}' \
  '{"id":"c","inputs":{"Company Name":"Linear"}}' > accounts.jsonl

2. Store the key

umask 077
printf 'export CLAY_API_KEY=%s\n' "<your key>" > ~/.clay-cron.env

umask 077
printf 'export CLAY_API_KEY=%s\n' "<your key>" > ~/.clay-cron.env

umask 077 makes the file readable only by you.

3. Write the script and the entry point

Save this as ~/clay-cron/nightly_enrich.py:

"""Nightly Clay enrichment meant to be started by cron.

Reads accounts.jsonl, runs Company Domain over it, writes out/<date>.jsonl, and appends one line to job.log.
A lock file stops a second copy from running while the first is still active.
"""
import fcntl
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

import requests

HERE = Path(__file__).resolve().parent
BASE = "https://api.clay.com/public/v0"
ROUTINE = "function:t_0tjqm9rUjfGY4MunGDM"
HEADERS = {"clay-api-key": os.environ["CLAY_API_KEY"], "content-type": "application/json"}


def log(**fields):
    fields["at"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with open(HERE / "job.log", "a") as f:
        f.write(json.dumps(fields) + "\n")


def call(path, method="GET", body=None):
    res = requests.request(method, BASE + path, headers=HEADERS, json=body, timeout=30)
    res.raise_for_status()
    return res.json()


def main():
    lock = open(HERE / "job.lock", "w")
    try:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        log(event="skipped", reason="previous run still active")
        return 0

    rows = [json.loads(line) for line in open(HERE / "accounts.jsonl")]
    before = call("/credits/balance")["balance"]
    run_id = call(f"/routines/{ROUTINE}/run", "POST", {"items": rows})["routine_run_id"]
    for _ in range(120):
        time.sleep(5)
        results = call(f"/routines/run/{run_id}/results?limit=100")
        if results["status"] != "in_progress":
            break
    else:
        log(event="timeout", run_id=run_id)
        return 1

    # The first page holds up to 100 rows. Follow the cursor for the rest of a larger list.
    while results.get("cursor"):
        more = call(f"/routines/run/{run_id}/results?limit=100&cursor={results['cursor']}")
        results["data"] = results.get("data", []) + more.get("data", [])
        results["cursor"] = more.get("cursor")

    out = HERE / "out"
    out.mkdir(exist_ok=True)
    stamp = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    with open(out / f"{stamp}.jsonl", "w") as f:
        for item in results.get("data", []):
            f.write(json.dumps(item) + "\n")

    failed = [r["id"] for r in results.get("data", []) if r["status"] != "complete"]
    time.sleep(5)
    after = call("/credits/balance")["balance"]
    log(event="finished", run_id=run_id, rows=len(rows), failed=failed, credits_spent=before - after)
    return 1 if failed else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as err:
        log(event="error", error=str(err))
        sys.exit(1)

"""Nightly Clay enrichment meant to be started by cron.

Reads accounts.jsonl, runs Company Domain over it, writes out/<date>.jsonl, and appends one line to job.log.
A lock file stops a second copy from running while the first is still active.
"""
import fcntl
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

import requests

HERE = Path(__file__).resolve().parent
BASE = "https://api.clay.com/public/v0"
ROUTINE = "function:t_0tjqm9rUjfGY4MunGDM"
HEADERS = {"clay-api-key": os.environ["CLAY_API_KEY"], "content-type": "application/json"}


def log(**fields):
    fields["at"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with open(HERE / "job.log", "a") as f:
        f.write(json.dumps(fields) + "\n")


def call(path, method="GET", body=None):
    res = requests.request(method, BASE + path, headers=HEADERS, json=body, timeout=30)
    res.raise_for_status()
    return res.json()


def main():
    lock = open(HERE / "job.lock", "w")
    try:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        log(event="skipped", reason="previous run still active")
        return 0

    rows = [json.loads(line) for line in open(HERE / "accounts.jsonl")]
    before = call("/credits/balance")["balance"]
    run_id = call(f"/routines/{ROUTINE}/run", "POST", {"items": rows})["routine_run_id"]
    for _ in range(120):
        time.sleep(5)
        results = call(f"/routines/run/{run_id}/results?limit=100")
        if results["status"] != "in_progress":
            break
    else:
        log(event="timeout", run_id=run_id)
        return 1

    # The first page holds up to 100 rows. Follow the cursor for the rest of a larger list.
    while results.get("cursor"):
        more = call(f"/routines/run/{run_id}/results?limit=100&cursor={results['cursor']}")
        results["data"] = results.get("data", []) + more.get("data", [])
        results["cursor"] = more.get("cursor")

    out = HERE / "out"
    out.mkdir(exist_ok=True)
    stamp = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    with open(out / f"{stamp}.jsonl", "w") as f:
        for item in results.get("data", []):
            f.write(json.dumps(item) + "\n")

    failed = [r["id"] for r in results.get("data", []) if r["status"] != "complete"]
    time.sleep(5)
    after = call("/credits/balance")["balance"]
    log(event="finished", run_id=run_id, rows=len(rows), failed=failed, credits_spent=before - after)
    return 1 if failed else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as err:
        log(event="error", error=str(err))
        sys.exit(1)

Create ~/clay-cron/run.sh with this content and make it executable:

#!/bin/sh
# Entry point for cron. cron starts jobs with a minimal environment, so load the key explicitly
# and call the virtualenv's interpreter by absolute path.
. "$HOME/.clay-cron.env"
exec "$HOME/clay-cron/.venv/bin/python" -I "$HOME/clay-cron/nightly_enrich.py"

#!/bin/sh
# Entry point for cron. cron starts jobs with a minimal environment, so load the key explicitly
# and call the virtualenv's interpreter by absolute path.
. "$HOME/.clay-cron.env"
exec "$HOME/clay-cron/.venv/bin/python" -I "$HOME/clay-cron/nightly_enrich.py"

4. Test it with cron's environment

Copy the job into a throwaway home directory and start two copies four seconds apart, with only the PATH cron would provide:

T=$(mktemp -d)
mkdir "$T/clay-cron" && cp -R ~/clay-cron/. "$T/clay-cron/" && cp ~/.clay-cron.env "$T/"
rm -f "$T/clay-cron/job.log"
cd "$T"
env -i HOME="$T" PATH=/usr/bin:/bin /bin/sh -c "$T/clay-cron/run.sh" &
sleep 4
env -i HOME="$T" PATH=/usr/bin:/bin /bin/sh -c "$T/clay-cron/run.sh"; echo "second copy exit=$?
wait; echo "first copy exit=$?"
cat "$T/clay-cron/job.log"
ls "$T/clay-cron/out"; cat "$T"/clay-cron/out/*.jsonl

T=$(mktemp -d)
mkdir "$T/clay-cron" && cp -R ~/clay-cron/. "$T/clay-cron/" && cp ~/.clay-cron.env "$T/"
rm -f "$T/clay-cron/job.log"
cd "$T"
env -i HOME="$T" PATH=/usr/bin:/bin /bin/sh -c "$T/clay-cron/run.sh" &
sleep 4
env -i HOME="$T" PATH=/usr/bin:/bin /bin/sh -c "$T/clay-cron/run.sh"; echo "second copy exit=$?"
wait; echo "first copy exit=$?"
cat "$T/clay-cron/job.log"
ls "$T/clay-cron/out"; cat "$T"/clay-cron/out/*.jsonl

5. Add the cron entry

crontab -e

crontab -e

0 6 * * * /bin/sh $HOME/clay-cron/run.sh

0 6 * * * /bin/sh $HOME/clay-cron/run.sh

This line starts the job at 06:00 daily. No crontab was edited for this test, so the 06:00 trigger itself was not observed, and the step 4 run stands in for it.

6. Verify the result

Printed by the step 4 commands on 2026-10-12:

second copy exit=0
first copy exit=0
--- job.log
{"event": "skipped", "reason": "previous run still active", "at": "2026-10-14T01:02:11+00:00"}
{"event": "finished", "run_id": "run_0tmvhjjQ9ZEBUc58GzV", "rows": 3, "failed": [], "credits_spent": -10.0, "at": "2026-10-14T01:02:23+00:00"}
--- out/
2026-10-14.jsonl
{"id": "b", "status": "complete", "result": {"Domain": "figma.com"}}
{"id": "a", "status": "complete", "result": {"Domain": "notion.com"}}
{"id": "c", "status": "complete", "result": {"Domain": "linear.app"}}

second copy exit=0
first copy exit=0
--- job.log
{"event": "skipped", "reason": "previous run still active", "at": "2026-10-14T01:02:11+00:00"}
{"event": "finished", "run_id": "run_0tmvhjjQ9ZEBUc58GzV", "rows": 3, "failed": [], "credits_spent": -10.0, "at": "2026-10-14T01:02:23+00:00"}
--- out/
2026-10-14.jsonl
{"id": "b", "status": "complete", "result": {"Domain": "figma.com"}}
{"id": "a", "status": "complete", "result": {"Domain": "notion.com"}}
{"id": "c", "status": "complete", "result": {"Domain": "linear.app"}}

The check is a set of properties: the second copy exits 0 and logs skipped, the first copy exits 0 and logs finished with rows of 3 and an empty failed list, and the dated results file has one line for each of the 3 accounts. The credits_spent value is a balance difference, and in the output above it is negative because credits were added to the workspace by something else while the job ran, as described below.

6b. Check a list longer than one page

The results endpoint returns 20 rows per page by default. A job that reads only the first page writes 20 lines for a 25-row list and still reports success. The script requests ?limit=100 and follows cursor for anything beyond that. To confirm, replace accounts.jsonl with 25 lines, run the job again from the throwaway home, and count the output lines. The output below is from that run:

exit=0
{"event": "finished", "run_id": "run_0tmvhjzNJFhwwypy5Sd", "rows": 25, "failed": [], "credits_spent": 39.0, "at": "2026-10-14T01:02:45+00:00"}
      25 clay-cron/out/2026-10-14.jsonl

exit=0
{"event": "finished", "run_id": "run_0tmvhjzNJFhwwypy5Sd", "rows": 25, "failed": [], "credits_spent": 39.0, "at": "2026-10-14T01:02:45+00:00"}
      25 clay-cron/out/2026-10-14.jsonl

The check is that rows is 25, failed is empty, and the dated file has 25 lines.

How it works

run.sh loads the key file and then replaces itself with the virtualenv's interpreter, so the script does not depend on the PATH. The -I flag runs Python in isolated mode, which ignores PYTHON* environment variables and the current directory on the import path.

fcntl.flock with LOCK_NB fails immediately when another process holds the lock, which is how the second copy detected the first. An flock lock belongs to the open file descriptor, so it is released when the process exits.

Common issues

The system python may not have `requests`

On the machine used here, /usr/bin/python3 is Python 3.9 and import requests failed with ModuleNotFoundError. A virtualenv python at an absolute path avoids this.

Overlap exits 0 on purpose

A skipped run exits 0 so cron does not treat a long previous run as a failure. If you want an alert on overlap, exit with a non-zero code in the BlockingIOError branch.

Polling gives up after 10 minutes

After 10 minutes the script logs a timeout line that carries the run_id and exits 1. Results for that run can still be fetched by id afterward.

Results come in pages of 20

The default page size is 20 rows, and a cursor field in the response points to the next page. Without limit and the cursor loop the script would drop every row after the first 20. With ?limit=100 the whole 25-row list arrived in one page, and following cursor with ?limit=10 returned the same 25 rows in 3 pages.

The credit delta can include other jobs

credits_spent is a difference of two balance reads, so anything else that changes the balance during the run is included in it. The first run above reports a negative figure, which means the balance rose during the run for a reason outside the job, and the 25-row run reports 39 for 25 rows.

Next steps

verification:
  status: verified
  tested_at: "2026-10-12"
  product_version: "public/v0"
  command: "env -i HOME=$T PATH=/usr/bin:/bin /bin/sh -c $T/clay-cron/run.sh"
  expected_result: "A second copy started during the first logs skipped and exits 0; the first logs finished with rows 3 and no failures and writes a dated results file with 3 lines; a 25-row list produces 25 lines."

verification:
  status: verified
  tested_at: "2026-10-12"
  product_version: "public/v0"
  command: "env -i HOME=$T PATH=/usr/bin:/bin /bin/sh -c $T/clay-cron/run.sh"
  expected_result: "A second copy started during the first logs skipped and exits 0; the first logs finished with rows 3 and no failures and writes a dated results file with 3 lines; a 25-row list produces 25 lines."