Refreshing the Stalest Contacts in a CSV Within a Credit Budget with Python

A contact list cannot be re-checked all at once without a large credit bill, so a recurring job refreshes the oldest records first and stops when its budget is gone. This Python script sorts a CSV by last verification date, re-runs Work Email for as many contacts as the budget pays for, writes the result and a new verification date back to the file, and leaves the rest for the next run. Two consecutive runs refreshed different contacts.

What you will build

A script that reads contacts.csv, picks the stalest rows that fit in BUDGET_CREDITS, runs them, records found_email, a check_result of unchanged, changed, or not_found, and today's date as last_verified, then rewrites the file. Because the dates move, running it again picks the next group.

contacts.csv  (last_verified)
    ↓ sort ascending, take floor(BUDGET / 3) rows
POST /routines/{Work Email}/run   -> poll /routines/run/{id}/results
    ↓
compare found vs stored  ->  unchanged | changed | not_found
    ↓
rewrite contacts.csv with new last_verified for the rows that completed

AI Prompt

Write a Python script that refreshes the stalest contacts in a CSV within a credit budget, using the Clay Public API.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- contacts.csv has the columns id, full_name, company_name, company_domain, stored_email, last_verified (an ISO date).
- Use the "Work Email" routine, function:t_0tjqmaaroo9mQFCBA5C. Required inputs: "Full Name", "Company Domain", "Company Name". The listed estimate is 3 credits per run, read once with `clay routines get`.
- BUDGET_CREDITS (default 6) limits the run: rows to refresh = floor(min(budget, balance) / 3), taken from the oldest last_verified date. If that is zero rows, exit with code 1 and send nothing.
- Run the rows with POST /routines/{id}/run and body { items: [{ id, inputs }] }, poll GET /routines/run/{routine_run_id}/results for up to 10 minutes, and match rows by id.
- For each row with status "complete", set found_email to result["Work Email"] or an empty string, and check_result to "unchanged" (same address ignoring case), "changed", or "not_found". Set last_verified to the run date. A row that did not complete keeps its old date so that the next run picks it up again.
- The measured spend is not guaranteed to equal the estimate. Report the balance difference around the run.
- Print masked addresses (***@domain) only. The CSV may hold full addresses.
- Run the verification step below before finishing.

Write a Python script that refreshes the stalest contacts in a CSV within a credit budget, using the Clay Public API.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- contacts.csv has the columns id, full_name, company_name, company_domain, stored_email, last_verified (an ISO date).
- Use the "Work Email" routine, function:t_0tjqmaaroo9mQFCBA5C. Required inputs: "Full Name", "Company Domain", "Company Name". The listed estimate is 3 credits per run, read once with `clay routines get`.
- BUDGET_CREDITS (default 6) limits the run: rows to refresh = floor(min(budget, balance) / 3), taken from the oldest last_verified date. If that is zero rows, exit with code 1 and send nothing.
- Run the rows with POST /routines/{id}/run and body { items: [{ id, inputs }] }, poll GET /routines/run/{routine_run_id}/results for up to 10 minutes, and match rows by id.
- For each row with status "complete", set found_email to result["Work Email"] or an empty string, and check_result to "unchanged" (same address ignoring case), "changed", or "not_found". Set last_verified to the run date. A row that did not complete keeps its old date so that the next run picks it up again.
- The measured spend is not guaranteed to equal the estimate. Report the balance difference around the run.
- Print masked addresses (***@domain) only. The CSV may hold full addresses.
- Run the verification step below before finishing.

Prerequisites

Python 3.10+ with requests
A Clay Public API key
At least 30 credits for the two sample runs, which spent 19 together

1. Set up the project

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

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

Create the contact file. The stored addresses are first-name guesses, which is what a list filled by pattern looks like:

id,full_name,company_name,company_domain,stored_email,last_verified
c1,Aaron Levie,Box,box.com,[email protected],2025-01-15
c2,Dylan Field,Figma,figma.com,[email protected],2025-06-14
c3,Daniel Ek,Spotify,spotify.com,[email protected],2025-08-30
c4,Patrick Collison,Stripe,stripe.com,[email protected],2025-11-02
c5,Tobi Lutke,Shopify,shopify.com,[email protected],2026-09-20

id,full_name,company_name,company_domain,stored_email,last_verified
c1,Aaron Levie,Box,box.com,[email protected],2025-01-15
c2,Dylan Field,Figma,figma.com,[email protected],2025-06-14
c3,Daniel Ek,Spotify,spotify.com,[email protected],2025-08-30
c4,Patrick Collison,Stripe,stripe.com,[email protected],2025-11-02
c5,Tobi Lutke,Shopify,shopify.com,[email protected],2026-09-20

2. Implement the script

The shared helper, clay.py, polls for up to 10 minutes by default:

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 refresh script, refresh_stalest.py:

import csv
import json
import os
import sys
import time
from datetime import date

from clay import balance, run_routine

WORK_EMAIL = "function:t_0tjqmaaroo9mQFCBA5C"
PER_ROW = 3  # listed estimate for Work Email, from `clay routines get`
BUDGET = float(os.environ.get("BUDGET_CREDITS", "6"))
TODAY = date(2026, 10, 12).isoformat()
FIELDS = ["id", "full_name", "company_name", "company_domain", "stored_email", "last_verified", "found_email", "check_result"]


def mask(email):
    return "***@" + email.split("@")[1] if email else None


def main():
    with open("contacts.csv", newline="") as f:
        rows = list(csv.DictReader(f))
    for r in rows:
        r.setdefault("found_email", "")
        r.setdefault("check_result", "")

    # Oldest verification date first, limited to what the budget pays for.
    slots = int(min(BUDGET, balance()) // PER_ROW)
    chosen = sorted(rows, key=lambda r: r["last_verified"])[:slots]
    print(json.dumps({"contacts": len(rows), "budget": BUDGET, "slots": slots, "chosen": [r["id"] for r in chosen]}))
    if not chosen:
        print("Budget too small for one contact", file=sys.stderr)
        sys.exit(1)

    before = balance()
    results = run_routine(
        WORK_EMAIL,
        [{"id": r["id"], "inputs": {"Full Name": r["full_name"], "Company Domain": r["company_domain"], "Company Name": r["company_name"]}} for r 
    )
    time.sleep(5)
    after = balance()

    by_id = {item["id"]: item for item in results.get("data", [])}
    for r in chosen:
        item = by_id.get(r["id"])
        found = ((item or {}).get("result") or {}).get("Work Email", "")
        if item is None or item["status"] != "complete":
            continue  # leave last_verified alone so the row is picked again next time
        r["found_email"] = found
        r["check_result"] = "not_found" if not found else ("unchanged" if found.lower() == r["stored_email"].lower() else
        r["last_verified"] = TODAY
        print(json.dumps({"id": r["id"], "result": r["check_result"], "stored": mask(r["stored_email"]), "found": mask(found)}))

    with open("contacts.csv", "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        writer.writerows(rows)
    print(json.dumps({"rows_refreshed": sum(1 for r in chosen if r["last_verified"] == TODAY), "credits_spent"


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

import csv
import json
import os
import sys
import time
from datetime import date

from clay import balance, run_routine

WORK_EMAIL = "function:t_0tjqmaaroo9mQFCBA5C"
PER_ROW = 3  # listed estimate for Work Email, from `clay routines get`
BUDGET = float(os.environ.get("BUDGET_CREDITS", "6"))
TODAY = date(2026, 10, 12).isoformat()
FIELDS = ["id", "full_name", "company_name", "company_domain", "stored_email", "last_verified", "found_email", "check_result"]


def mask(email):
    return "***@" + email.split("@")[1] if email else None


def main():
    with open("contacts.csv", newline="") as f:
        rows = list(csv.DictReader(f))
    for r in rows:
        r.setdefault("found_email", "")
        r.setdefault("check_result", "")

    # Oldest verification date first, limited to what the budget pays for.
    slots = int(min(BUDGET, balance()) // PER_ROW)
    chosen = sorted(rows, key=lambda r: r["last_verified"])[:slots]
    print(json.dumps({"contacts": len(rows), "budget": BUDGET, "slots": slots, "chosen": [r["id"] for r in chosen]}))
    if not chosen:
        print("Budget too small for one contact", file=sys.stderr)
        sys.exit(1)

    before = balance()
    results = run_routine(
        WORK_EMAIL,
        [{"id": r["id"], "inputs": {"Full Name": r["full_name"], "Company Domain": r["company_domain"], "Company Name": r["company_name"]}} for r in chosen],
    )
    time.sleep(5)
    after = balance()

    by_id = {item["id"]: item for item in results.get("data", [])}
    for r in chosen:
        item = by_id.get(r["id"])
        found = ((item or {}).get("result") or {}).get("Work Email", "")
        if item is None or item["status"] != "complete":
            continue  # leave last_verified alone so the row is picked again next time
        r["found_email"] = found
        r["check_result"] = "not_found" if not found else ("unchanged" if found.lower() == r["stored_email"].lower() else "changed")
        r["last_verified"] = TODAY
        print(json.dumps({"id": r["id"], "result": r["check_result"], "stored": mask(r["stored_email"]), "found": mask(found)}))

    with open("contacts.csv", "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        writer.writerows(rows)
    print(json.dumps({"rows_refreshed": sum(1 for r in chosen if r["last_verified"] == TODAY), "credits_spent": before - after}))


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

3. Run it twice

python3 refresh_stalest.py
python3 refresh_stalest.py
cut -d, -f1,6,8 contacts.csv

python3 refresh_stalest.py
python3 refresh_stalest.py
cut -d, -f1,6,8 contacts.csv

4. Check the output

Output on 2026-10-12:

## first run
{"contacts": 5, "budget": 6.0, "slots": 2, "chosen": ["c1", "c2"]}
{"id": "c1", "result": "unchanged", "stored": "***@box.com", "found": "***@box.com"}
{"id": "c2", "result": "changed", "stored": "***@figma.com", "found": "***@figma.com"}
{"rows_refreshed": 2, "credits_spent": 13}
## second run
{"contacts": 5, "budget": 6.0, "slots": 2, "chosen": ["c3", "c4"]}
{"id": "c3", "result": "changed", "stored": "***@spotify.com", "found": "***@spotify.com"}
{"id": "c4", "result": "changed", "stored": "***@stripe.com", "found": "***@stripe.com"}
{"rows_refreshed": 2, "credits_spent": 6}
## last_verified after both runs
c1 2026-10-12 unchanged
c2 2026-10-12 changed
c3 2026-10-12 changed
c4 2026-10-12 changed
c5 2026-09-20 -

## first run
{"contacts": 5, "budget": 6.0, "slots": 2, "chosen": ["c1", "c2"]}
{"id": "c1", "result": "unchanged", "stored": "***@box.com", "found": "***@box.com"}
{"id": "c2", "result": "changed", "stored": "***@figma.com", "found": "***@figma.com"}
{"rows_refreshed": 2, "credits_spent": 13}
## second run
{"contacts": 5, "budget": 6.0, "slots": 2, "chosen": ["c3", "c4"]}
{"id": "c3", "result": "changed", "stored": "***@spotify.com", "found": "***@spotify.com"}
{"id": "c4", "result": "changed", "stored": "***@stripe.com", "found": "***@stripe.com"}
{"rows_refreshed": 2, "credits_spent": 6}
## last_verified after both runs
c1 2026-10-12 unchanged
c2 2026-10-12 changed
c3 2026-10-12 changed
c4 2026-10-12 changed
c5 2026-09-20 -

What to check is a set of properties. Each run chooses slots rows, which is 2 for a budget of 6. The first run chooses the two oldest dates and the second run chooses the next two, because the first run moved its rows to today's date. The fifth contact, verified on 2026-09-20, is never chosen in these two runs. Each chosen row ends with exactly one check_result. Which addresses come back, and what each run costs, can differ from run to run.

How it works

Sorting by last_verified and writing the date back turns the file into a queue. Nothing else tracks progress, and a row that failed to complete keeps its old date, so it stays at the front.

slots divides the smaller of the budget and the balance by the per-row estimate. That decides how many rows to send. The measured difference of two balance reads reports what the run really cost.

Common issues

Budgets above 20 rows

A budget that pays for more than 20 contacts returns more than one page of results. The helper requests limit=100 and follows the cursor, so rows 21 and later are not left with their old dates by mistake.

The estimate is not a ceiling

The same two-row run cost 13 credits the first time and 6 the second, against an estimate of 3 per row. A budget of 6 therefore bounded the number of rows and not the spend. Treat the figure as a planning number and set the budget with room for it, or lower the slots after seeing real costs.

A changed address is a candidate, not a correction

changed means the routine returned a different address from the stored one. The script writes it to found_email and leaves stored_email alone, so someone can review before the CRM changes. The stored guesses in this example are the kind of values that differ from a verified address.

The run date is a constant in the example

TODAY is set to 2026-10-12 so the example repeats. Use date.today() in a real job.

A `not_found` row still moves to the back of the queue

Every row that completes gets today's date, including rows with no address. That prevents a contact with no available address from being re-sent, and re-billed, on every run. If you want a retry sooner, store a separate last_attempted date.

Next steps

verification:
  status: verified
  tested_at: "2026-10-12"
  product_version: "public/v0"
  command: "python3 refresh_stalest.py && python3 refresh_stalest.py"
  expected_result: "The first run refreshes the two oldest contacts, the second run refreshes the next two, and each refreshed row has one check_result and a new last_verified."

verification:
  status: verified
  tested_at: "2026-10-12"
  product_version: "public/v0"
  command: "python3 refresh_stalest.py && python3 refresh_stalest.py"
  expected_result: "The first run refreshes the two oldest contacts, the second run refreshes the next two, and each refreshed row has one check_result and a new last_verified."