Bulk Enriching a CSV of Companies over REST with Python

The batch endpoints process a file of rows in one run instead of one request per row. This example reads a CSV of company names, uploads it as JSONL, waits for the run, downloads the result file, and writes a CSV with a domain for each company.

What you will build

A script that turns companies.csv (6 rows, one with no name) into companies_with_domains.csv using the four-step batch flow, and reports the credits spent.

companies.csv
    ↓
POST /routines/{id}/run-batch/upload-url     -> { upload_url, file_id }
PUT  <upload_url>                            (JSONL body, no Clay key)
POST /routines/{id}/run-batch/start          -> { routine_run_id }
GET  /routines/run-batch/{id}/results        (poll) -> { status, total, finished, result_url }
GET  <result_url>                            -> one JSON line per row
    ↓
companies_with_domains.csv

companies.csv
    ↓
POST /routines/{id}/run-batch/upload-url     -> { upload_url, file_id }
PUT  <upload_url>                            (JSONL body, no Clay key)
POST /routines/{id}/run-batch/start          -> { routine_run_id }
GET  /routines/run-batch/{id}/results        (poll) -> { status, total, finished, result_url }
GET  <result_url>                            -> one JSON line per row
    ↓
companies_with_domains.csv

AI Prompt

Implement a Python script that enriches a CSV of company names through the Clay Public API batch endpoints.

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.
- Skip rows with an empty company name before uploading. Keep the CSV row number as the row id so results can be joined back, because the result file is not in input order.
- Step 1: POST /routines/{id}/run-batch/upload-url with body {} returns { upload_url, file_id }.
- Step 2: PUT the JSONL to upload_url with content-type application/octet-stream. Each line is { "id": "<string>", "inputs": { ... } }. This request goes to the storage host. Do not send the clay-api-key header to it.
- Step 3: POST /routines/{id}/run-batch/start with { file_id } returns { routine_run_id, status }.
- Step 4: poll GET /routines/run-batch/{routine_run_id}/results until status is not "in_progress". A complete run returns { status, total, finished, result_url }. The rows are not inline.
- Download result_url without the Clay key. It is a separate presigned URL and expires after about 15 minutes. Do not print it.
- Write the domain for each company to a new CSV, leaving the cell empty when a row has no result.
- Read GET /credits/balance before and after, and report the credits spent.
- Run the verification step below before finishing.

Implement a Python script that enriches a CSV of company names through the Clay Public API batch endpoints.

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.
- Skip rows with an empty company name before uploading. Keep the CSV row number as the row id so results can be joined back, because the result file is not in input order.
- Step 1: POST /routines/{id}/run-batch/upload-url with body {} returns { upload_url, file_id }.
- Step 2: PUT the JSONL to upload_url with content-type application/octet-stream. Each line is { "id": "<string>", "inputs": { ... } }. This request goes to the storage host. Do not send the clay-api-key header to it.
- Step 3: POST /routines/{id}/run-batch/start with { file_id } returns { routine_run_id, status }.
- Step 4: poll GET /routines/run-batch/{routine_run_id}/results until status is not "in_progress". A complete run returns { status, total, finished, result_url }. The rows are not inline.
- Download result_url without the Clay key. It is a separate presigned URL and expires after about 15 minutes. Do not print it.
- Write the domain for each company to a new CSV, leaving the cell empty when a row has no result.
- Read GET /credits/balance before and after, and report the credits spent.
- Run the verification step below before finishing.

Prerequisites

Python 3.10+ and pip install requests
A Clay Public API key (clay_scoped_...)
5 credits, one per uploaded company

1. Set up the project

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

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

Create the input file. The row with an empty company_name is a deliberate test row:

printf 'company_name,source\nNotion,web\nFigma,web\n,import\nLinear,web\nVercel,web\nSupabase,web\n' > companies.csv

printf 'company_name,source\nNotion,web\nFigma,web\n,import\nLinear,web\nVercel,web\nSupabase,web\n' > companies.csv

2. Configure credentials

export CLAY_API_KEY=clay_scoped_...

export CLAY_API_KEY=clay_scoped_...

3. Implement the script

Save this shared helper as 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")

Save the script as bulk_csv.py:

import csv
import json
import sys
import time

import requests

from clay import balance, clay

ROUTINE = "function:t_0tjqm9rUjfGY4MunGDM"  # Company Domain, 1 credit per row


def main():
    with open("companies.csv", newline="") as f:
        companies = [row["company_name"].strip() for row in csv.DictReader(f)]
    rows = [{"id": str(i), "inputs": {"Company Name": name}} for i, name in enumerate(companies) if name]
    body = "\n".join(json.dumps(r) for r in rows).encode()
    print(json.dumps({"csv_rows": len(companies), "uploaded_rows": len(rows)}))

    before = balance()

    # 1. Ask for a presigned upload URL.
    upload = clay(f"/routines/{ROUTINE}/run-batch/upload-url", "POST", {})
    # 2. PUT the JSONL to storage. This request goes to the storage host, so it carries no Clay key.
    put = requests.put(upload["upload_url"], data=body, headers={"content-type": "application/octet-stream"}, timeout=60)
    put.raise_for_status()
    # 3. Start the run from the uploaded file.
    run_id = clay(f"/routines/{ROUTINE}/run-batch/start", "POST", {"file_id": upload["file_id"]})["routine_run_id"]
    # 4. Poll until the run is no longer in progress.
    deadline = time.time() + 900
    while time.time() < deadline:
        time.sleep(5)
        status = clay(f"/routines/run-batch/{run_id}/results")
        if status["status"] != "in_progress":
            break
    else:
        raise RuntimeError(f"run {run_id} still in progress after 900 seconds")
    print(json.dumps({"run_id": run_id, "status": status["status"], "total": status["total"], "finished": status["finished"]}))

    # 5. The per-row results are in a second presigned file, not in the poll response.
    lines = requests.get(status["result_url"], timeout=60).text.splitlines()
    by_id = {}
    for line in lines:
        item = json.loads(line)
        by_id[item["id"]] = (item.get("result") or {}).get("Domain", "")

    with open("companies_with_domains.csv", "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["company_name", "domain"])
        for i, name in enumerate(companies):
            writer.writerow([name, by_id.get(str(i), "")])

    time.sleep(5)
    print(json.dumps({"rows_written": len(companies), "credits_spent": before - balance()}))


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

import csv
import json
import sys
import time

import requests

from clay import balance, clay

ROUTINE = "function:t_0tjqm9rUjfGY4MunGDM"  # Company Domain, 1 credit per row


def main():
    with open("companies.csv", newline="") as f:
        companies = [row["company_name"].strip() for row in csv.DictReader(f)]
    rows = [{"id": str(i), "inputs": {"Company Name": name}} for i, name in enumerate(companies) if name]
    body = "\n".join(json.dumps(r) for r in rows).encode()
    print(json.dumps({"csv_rows": len(companies), "uploaded_rows": len(rows)}))

    before = balance()

    # 1. Ask for a presigned upload URL.
    upload = clay(f"/routines/{ROUTINE}/run-batch/upload-url", "POST", {})
    # 2. PUT the JSONL to storage. This request goes to the storage host, so it carries no Clay key.
    put = requests.put(upload["upload_url"], data=body, headers={"content-type": "application/octet-stream"}, timeout=60)
    put.raise_for_status()
    # 3. Start the run from the uploaded file.
    run_id = clay(f"/routines/{ROUTINE}/run-batch/start", "POST", {"file_id": upload["file_id"]})["routine_run_id"]
    # 4. Poll until the run is no longer in progress.
    deadline = time.time() + 900
    while time.time() < deadline:
        time.sleep(5)
        status = clay(f"/routines/run-batch/{run_id}/results")
        if status["status"] != "in_progress":
            break
    else:
        raise RuntimeError(f"run {run_id} still in progress after 900 seconds")
    print(json.dumps({"run_id": run_id, "status": status["status"], "total": status["total"], "finished": status["finished"]}))

    # 5. The per-row results are in a second presigned file, not in the poll response.
    lines = requests.get(status["result_url"], timeout=60).text.splitlines()
    by_id = {}
    for line in lines:
        item = json.loads(line)
        by_id[item["id"]] = (item.get("result") or {}).get("Domain", "")

    with open("companies_with_domains.csv", "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["company_name", "domain"])
        for i, name in enumerate(companies):
            writer.writerow([name, by_id.get(str(i), "")])

    time.sleep(5)
    print(json.dumps({"rows_written": len(companies), "credits_spent": before - balance()}))


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

4. Run

python3 bulk_csv.py

python3 bulk_csv.py

5. Verify the result

Live run on 2026-10-12, workspace 1322578:

{"csv_rows": 6, "uploaded_rows": 5}
{"run_id": "run_0tmtmctBXmNpWN2qeJS", "status": "complete", "total": 5, "finished": 5}
{"rows_written": 6, "credits_spent": 5.0}
company_name,domain
Notion,notion.com
Figma,figma.com
,
Linear,linear.app
Vercel,vercel.com
Supabase,supabase.com

{"csv_rows": 6, "uploaded_rows": 5}
{"run_id": "run_0tmtmctBXmNpWN2qeJS", "status": "complete", "total": 5, "finished": 5}
{"rows_written": 6, "credits_spent": 5.0}
company_name,domain
Notion,notion.com
Figma,figma.com
,
Linear,linear.app
Vercel,vercel.com
Supabase,supabase.com

The check is a set of properties: uploaded_rows is one less than csv_rows because the row with no name was skipped, finished equals total, the output has one line per CSV row with an empty domain for the row with no name, and credits_spent equals uploaded_rows. The domain values are whatever the routine answered on the day of the run.

How it works

The file body is sent to the storage host named in upload_url, not to api.clay.com. start registers the uploaded file against the routine and returns a run id. The poll endpoint reports progress and, once the run is complete, a second presigned URL for the results. Both presigned URLs carry their own authorization, so the script sends the Clay key only to api.clay.com.

Common issues

The result file is a second download

The poll response has total, finished, and result_url, but no rows. A script that reads data from the poll response finds nothing.

Results are not in input order

Each result line carries the id of its input row. The script joins on that id and writes rows in CSV order.

A row with a missing input can reject the file

The CSV has a row with an empty company name, which the script skips. On 2026-10-09 a bulk file that contained a row missing a required input was rejected after upload with validation_failed, and no rows ran. That run used the CLI, and the same routine and file format apply here. Remove rows that lack a required input before building the file.

Inline runs are limited to 100 rows

POST /routines/{id}/run is meant for small lists. The CLI help for the same operation states 1 to 100 items per inline run and points to the bulk path for larger lists.

Next steps

verification:
  status: verified
  tested_at: "2026-10-12"
  product_version: "public/v0"
  command: "python3 bulk_csv.py"
  expected_result: "The row with no name is skipped, the run completes with finished equal to total, companies_with_domains.csv has one row per input row, and credits_spent equals the rows uploaded."

verification:
  status: verified
  tested_at: "2026-10-12"
  product_version: "public/v0"
  command: "python3 bulk_csv.py"
  expected_result: "The row with no name is skipped, the run completes with finished equal to total, companies_with_domains.csv has one row per input row, and credits_spent equals the rows uploaded."