Building a Dry-Run Lookup Page for Teammates with Python

A small Flask page lets someone who does not write code paste company names, see how many rows and credits a lookup would use, and press a button to run it. The preview step makes no run request, and this example checks that with real HTTP requests against the running page, including two presses of the run button at the same moment.

What you will build

A Flask app, app.py, with three pages: a form, a preview that spends nothing, and a results table. The preview cleans the pasted list (trims, collapses spaces, removes blanks and case-insensitive repeats), shows the row count, an estimate and the workspace balance, and gives the page a one-time token. The run step takes only that token, so the list cannot change between preview and run. A /stats route, added for testing, reports how many run requests the process has made.

GET  /          -> form
POST /preview   -> clean list, rows, estimate, balance (GET /credits/balance only)
POST /run       -> token -> POST /routines/{id}/run -> poll -> results table
GET  /stats     -> run_requests_sent (test aid)

AI Prompt

Implement a Flask app that lets a non-technical teammate look up company domains with a dry-run step first.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- Use the "Company Domain" routine, function:t_0tjqm9rUjfGY4MunGDM. Input "Company Name", result { "Domain": "<domain>" }. Use 1 credit per row as a planning figure.
- GET / shows a textarea. POST /preview normalizes the list: collapse whitespace, trim, drop blank lines, drop repeats that differ only by case or spacing, keep the first spelling. Reject an empty list and a list of more than 10 names with a 400 and a message on the page.
- The preview page shows the number of rows, how many lines were removed, the estimated credits (rows times 1), and GET /credits/balance. It must not call POST /routines/{id}/run. Store the cleaned names on the server under a random token and put only the token in the run form.
- POST /run takes the token. An unknown token returns 400. The first press runs the rows with POST /routines/{id}/run and { items: [{ id, inputs: { "Company Name": ... } }] }, polls GET /routines/run/{routine_run_id}/results?limit=100 until the status is not "in_progress", matches rows by id, and shows a table of company, status and domain. A second press of the same token shows the saved results with a notice and makes no new run request. Two presses at the same moment must also start one run.
- Escape everything shown on the page, including pasted names.
- A /stats route returns { run_requests_sent }, counting POST /routines/{id}/run calls in the HTTP helper.
- Verify with real HTTP requests against the running app, including that preview and every rejection leave run_requests_sent at 0.
- Run the verification step below before finishing.

Implement a Flask app that lets a non-technical teammate look up company domains with a dry-run step first.

Prerequisites

Python 3.10+
pip install flask requests
A Clay Public API key (clay_scoped_...)
About 5 credits for the verification run and the two-press test

1. Set up the project

mkdir clay-dry-run-page && cd clay-dry-run-page
python3 -m venv .venv && source .venv/bin/activate
pip

mkdir clay-dry-run-page && cd clay-dry-run-page
python3 -m venv .venv && source .venv/bin/activate
pip install flask requests

2. Discover the real schema

The page makes two kinds of call: a balance read and a run request with polling. A run request with an empty inputs object is rejected with 400 and the message Invalid inputs for item probe: Company Name: Missing required field: Company Name, which is how the input name Company Name was confirmed without spending credits. The results endpoint returns status, total, finished and data rows shaped { id, status, result: { Domain } }, and it is polled until status is no longer in_progress.

3. Configure credentials

export CLAY_API_KEY=clay_scoped_...

export CLAY_API_KEY=clay_scoped_...

4. Implement the app

Save the 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[

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"})
run_requests_sent = 0  # POST /routines/{id}/run calls made by this process


def clay(path, method="GET", body=None, raw=False):
    global run_requests_sent
    if method == "POST" and path.endswith("/run"):
        run_requests_sent += 1
    res = session.request(method, f"{BASE}{path}", json=body, timeout=60)
    if raw:
        return res
    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):
    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(4)
        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 app as app.py:

import re
import threading
import uuid

from flask import Flask, render_template_string, request

import clay

COMPANY_DOMAIN = "function:t_0tjqm9rUjfGY4MunGDM"
PER_ROW = 1      # credits per row observed for Company Domain; a planning figure
MAX_ROWS = 10    # the page refuses anything larger

app = Flask(__name__)
PREVIEWS = {}    # token -> {"names": [...], "result": None | list}

PAGE = """<!doctype html>
<meta charset="utf-8">
<title>Company domain lookup</title>
<style>body{font:16px system-ui;max-width:640px;margin:2rem auto;padding:0 1rem}
textarea{width:100%;height:9rem}td,th{border-bottom:1px solid #ddd;padding:.3rem .6rem;text-align:left}
.warn{color:#9a3412}button{padding:.5rem 1rem;font-size:1rem}</style>
<h1>Company domain lookup</h1>
{% if error %}<p class="warn" id="error">{{ error }}</p>{% endif %}
{% if preview is none and result is none %}
<form method="post" action="/preview">
  <p>Paste company names, one per line (up to {{ max_rows }}).</p>
  <textarea name="names" required></textarea>
  <p><button type="submit">Preview (no credits used)</button></p>
</form>
{% elif result is none %}
<p>Rows to look up: <b id="rows">{{ preview.names|length }}</b>
   {% if preview.removed %}(<span id="removed">{{ preview.removed }}</span> blank or repeated lines removed){% endif %}</p>
<p>Estimated credits: <b id="estimate">{{ preview.estimate }}</b> &middot; Workspace balance: <span id="balance">{{ preview.balance }}</span></p>
<ul>{% for n in preview.names %}<li>{{ n }}</li>{% endfor %}</ul>
<p>No credits have been used. Pressing the button below starts the lookup.</p>
<form method="post" action="/run">
  <input type="hidden" name="token" value="{{ preview.token }}">
  <button type="submit" id="run">Run for {{ preview.estimate }} credits</button>
</form>
<p><a href="/">Start over</a></p>
{% else %}
{% if repeated %}<p id="repeated">This lookup already ran; showing the saved results. No new credits were used.</p>{% endif %}
<table id="results"><tr><th>Company</th><th>Status</th><th>Domain</th></tr>
{% for r in result %}<tr><td>{{ r.name }}</td><td>{{ r.status }}</td><td>{{ r.domain or "" }}</td></tr>{% endfor %}</table>
<p><a href="/">New lookup</a></p>
{% endif %}"""


def normalize(raw):
    names, seen, removed = [], set(), 0
    for line in raw.splitlines():
        name = re.sub(r"\s+", " ", line).strip()
        key = name.casefold()
        if not name or key in seen:
            removed += 1
            continue
        seen.add(key)
        names.append(name)
    return names, removed


def page(status=200, **ctx):
    ctx.setdefault("error", None); ctx.setdefault("preview", None); ctx.setdefault("result", None)
    ctx.setdefault("repeated", False)
    return render_template_string(PAGE, max_rows=MAX_ROWS, **ctx), status


@app.get("/")
def index():
    return page()


@app.post("/preview")
def preview():
    names, removed = normalize(request.form.get("names", ""))
    if not names:
        return page(400, error="Paste at least one company name.")
    if len(names) > MAX_ROWS:
        return page(400, error=f"{len(names)} names is more than the limit of {MAX_ROWS}.")
    token = uuid.uuid4().hex
    PREVIEWS[token] = {"names": names, "result": None, "lock": threading.Lock()}
    info = {"token": token, "names": names, "removed": removed, "estimate": len(names) * PER_ROW, "balance": clay.balance()}
    return page(preview=info)


@app.post("/run")
def run():
    entry = PREVIEWS.get(request.form.get("token", ""))
    if entry is None:
        return page(400, error="This preview has expired. Start over.")
    with entry["lock"]:  # two presses at once must not start two runs
        if entry["result"] is not None:
            return page(result=entry["result"], repeated=True)
        rows = [{"id": str(i), "inputs": {"Company Name": n}} for i, n in enumerate(entry["names"])]
        results = clay.run_routine(COMPANY_DOMAIN, rows)
        by_id = {r["id"]: r for r in results["data"]}
        entry["result"] = [{"name": n, "status": by_id.get(str(i), {}).get("status", "missing"),
                            "domain": (by_id.get(str(i), {}).get("result") or {}).get("Domain")} for i, n in enumerate(entry["names"])]
        return page(result=entry["result"])


@app.get("/stats")
def stats():
    """Test aid: how many POST /routines/{id}/run calls this process has made."""
    return {"run_requests_sent": clay.run_requests_sent}


if __name__ == "__main__":
    app.run(port=5057)

import re
import threading
import uuid

from flask import Flask, render_template_string, request

import clay

COMPANY_DOMAIN = "function:t_0tjqm9rUjfGY4MunGDM"
PER_ROW = 1      # credits per row observed for Company Domain; a planning figure
MAX_ROWS = 10    # the page refuses anything larger

app = Flask(__name__)
PREVIEWS = {}    # token -> {"names": [...], "result": None | list}

PAGE = """<!doctype html>
<meta charset="utf-8">
<title>Company domain lookup</title>
<style>body{font:16px system-ui;max-width:640px;margin:2rem auto;padding:0 1rem}
textarea{width:100%;height:9rem}td,th{border-bottom:1px solid #ddd;padding:.3rem .6rem;text-align:left}
.warn{color:#9a3412}button{padding:.5rem 1rem;font-size:1rem}</style>
<h1>Company domain lookup</h1>
{% if error %}<p class="warn" id="error">{{ error }}</p>{% endif %}
{% if preview is none and result is none %}
<form method="post" action="/preview">
  <p>Paste company names, one per line (up to {{ max_rows }}).</p>
  <textarea name="names" required></textarea>
  <p><button type="submit">Preview (no credits used)</button></p>
</form>
{% elif result is none %}
<p>Rows to look up: <b id="rows">{{ preview.names|length }}</b>
   {% if preview.removed %}(<span id="removed">{{ preview.removed }}</span> blank or repeated lines removed){% endif %}</p>
<p>Estimated credits: <b id="estimate">{{ preview.estimate }}</b> &middot; Workspace balance: <span id="balance">{{ preview.balance }}</span></p>
<ul>{% for n in preview.names %}<li>{{ n }}</li>{% endfor %}</ul>
<p>No credits have been used. Pressing the button below starts the lookup.</p>
<form method="post" action="/run">
  <input type="hidden" name="token" value="{{ preview.token }}">
  <button type="submit" id="run">Run for {{ preview.estimate }} credits</button>
</form>
<p><a href="/">Start over</a></p>
{% else %}
{% if repeated %}<p id="repeated">This lookup already ran; showing the saved results. No new credits were used.</p>{% endif %}
<table id="results"><tr><th>Company</th><th>Status</th><th>Domain</th></tr>
{% for r in result %}<tr><td>{{ r.name }}</td><td>{{ r.status }}</td><td>{{ r.domain or "" }}</td></tr>{% endfor %}</table>
<p><a href="/">New lookup</a></p>
{% endif %}"""


def normalize(raw):
    names, seen, removed = [], set(), 0
    for line in raw.splitlines():
        name = re.sub(r"\s+", " ", line).strip()
        key = name.casefold()
        if not name or key in seen:
            removed += 1
            continue
        seen.add(key)
        names.append(name)
    return names, removed


def page(status=200, **ctx):
    ctx.setdefault("error", None); ctx.setdefault("preview", None); ctx.setdefault("result", None)
    ctx.setdefault("repeated", False)
    return render_template_string(PAGE, max_rows=MAX_ROWS, **ctx), status


@app.get("/")
def index():
    return page()


@app.post("/preview")
def preview():
    names, removed = normalize(request.form.get("names", ""))
    if not names:
        return page(400, error="Paste at least one company name.")
    if len(names) > MAX_ROWS:
        return page(400, error=f"{len(names)} names is more than the limit of {MAX_ROWS}.")
    token = uuid.uuid4().hex
    PREVIEWS[token] = {"names": names, "result": None, "lock": threading.Lock()}
    info = {"token": token, "names": names, "removed": removed, "estimate": len(names) * PER_ROW, "balance": clay.balance()}
    return page(preview=info)


@app.post("/run")
def run():
    entry = PREVIEWS.get(request.form.get("token", ""))
    if entry is None:
        return page(400, error="This preview has expired. Start over.")
    with entry["lock"]:  # two presses at once must not start two runs
        if entry["result"] is not None:
            return page(result=entry["result"], repeated=True)
        rows = [{"id": str(i), "inputs": {"Company Name": n}} for i, n in enumerate(entry["names"])]
        results = clay.run_routine(COMPANY_DOMAIN, rows)
        by_id = {r["id"]: r for r in results["data"]}
        entry["result"] = [{"name": n, "status": by_id.get(str(i), {}).get("status", "missing"),
                            "domain": (by_id.get(str(i), {}).get("result") or {}).get("Domain")} for i, n in enumerate(entry["names"])]
        return page(result=entry["result"])


@app.get("/stats")
def stats():
    """Test aid: how many POST /routines/{id}/run calls this process has made."""
    return {"run_requests_sent": clay.run_requests_sent}


if __name__ == "__main__":
    app.run(port=5057)

Save the checks as verify_app.py:

import html
import re

import requests

BASE = "http://127.0.0.1:5057"


def field(body, ident):
    m = re.search(

import html
import re

import requests

BASE = "http://127.0.0.1:5057"


def field(body, ident):
    m = re.search(rf'id="{ident}"[^>]*>(.*?)<', body)
    return html.unescape(m.group(1)) if m else None


def token(body):
    return re.search(r'name="token" value="([0-9a-f]+)"', body).group(1)


def stats():
    return requests.get(f"{BASE}/stats").json()["run_requests_sent"]


def preview(text):
    return requests.post(f"{BASE}/preview", data={"names": text})


# 1. The form loads
r = requests.get(BASE)
print("GET /", r.status_code, "textarea" in r.text)

# 2. Dry run on a messy paste: blank line, a repeat that differs by case and spacing, non-ASCII, markup
r = preview("Notion\n  notion \n\nCafé Rouge\n<b>Figma</b>\nLinear")
print("preview", r.status_code, {"rows": field(r.text, "rows"), "removed": field(r.text, "removed"),
                                 "estimate": field(r.text, "estimate"), "balance_is_number": field(r.text, "balance").isdigit()})
print("markup escaped:", "&lt;b&gt;Figma&lt;/b&gt;" in r.text and "<b>Figma</b>" not in r.text)
print("run requests sent after preview:", stats())

# 3. Negative controls: nothing is sent for any of these
print("empty paste     ", preview("  \n\n").status_code, field(preview("  \n\n").text, "error"))
print("11 names        ", preview("\n".join(f"Company {i}" for i in range(11))).status_code,
      field(preview("\n".join(f"Company {i}" for i in range(11))).text, "error"))
bogus = requests.post(f"{BASE}/run", data={"token": "0" * 32})
print("bogus token     ", bogus.status_code, field(bogus.text, "error"))
print("run requests sent after controls:", stats())

# 4. The real run, from a fresh preview
p = preview("Notion\n notion\nCafé Rouge\nLinear\nVercel")
tok = token(p.text)
print("preview for run ", field(p.text, "rows"), "rows, estimate", field(p.text, "estimate"))
res = requests.post(f"{BASE}/run", data={"token": tok})
rows = re.findall(r"<tr><td>(.*?)</td><td>(.*?)</td><td>(.*?)</td></tr>", res.text)
print("run", res.status_code, [(html.unescape(a), b, c) for a, b, c in rows])
print("every row complete with a domain:", all(b == "complete" and c for _, b, c in rows), "| rows:", len(rows))
print("run requests sent after run:", stats())

# 5. The same button pressed again
again = requests.post(f"{BASE}/run", data={"token": tok})
print("second press    ", again.status_code, "repeated notice:", field(again.text, "repeated") is not None)
print("run requests sent after second press:", stats())

Save the concurrency check as race_test.py:

import re
import threading

import requests

BASE = "http://127.0.0.1:5057"
before = requests.get(f"{BASE}/stats").json()[

import re
import threading

import requests

BASE = "http://127.0.0.1:5057"
before = requests.get(f"{BASE}/stats").json()["run_requests_sent"]
p = requests.post(f"{BASE}/preview", data={"names": "Datadog"})
tok = re.search(r'name="token" value="([0-9a-f]+)"', p.text).group(1)
codes = []
threads = [threading.Thread(target=lambda: codes.append(requests.post(f"{BASE}/run", data={"token": tok}).status_code)) for _ in range(2)]
[t.start() for t in threads]
[t.join() for t in threads]
after = requests.get(f"{BASE}/stats").json()["run_requests_sent"]
print({"simultaneous_presses": 2, "http_codes": codes, "run_requests_sent_by_these_two": after - before})

5. Run

Start the app in one terminal, then run the checks in another:

python3 app.py

python3 app.py

python3 verify_app.py
python3 race_test.py

python3 verify_app.py
python3 race_test.py

A teammate opens http://127.0.0.1:5057/, pastes names, presses Preview, reads the count and estimate, and presses the Run button.

6. Verify the result

Tested live on 2026-10-20 against the real Clay API (workspace 1322578). Output of verify_app.py:

GET / 200 True
preview 200 {'rows': '4', 'removed': '2', 'estimate': '4', 'balance_is_number': True}
markup escaped: True
run requests sent after preview: 0
empty paste      400 Paste at least one company name.
11 names         400 11 names is more than the limit of 10.
bogus token      400 This preview has expired. Start over.
run requests sent after controls: 0
preview for run  4 rows, estimate 4
run 200 [('Notion', 'complete', 'notion.com'), ('Café Rouge', 'complete', 'caferouge.com'), ('Linear', 'complete', 'linear.app'), ('Vercel', 'complete', 'vercel.com')]
every row complete with a domain: True | rows: 4
run requests sent after run: 1
second press     200 repeated notice: True
run requests sent after second press: 1

GET / 200 True
preview 200 {'rows': '4', 'removed': '2', 'estimate': '4', 'balance_is_number': True}
markup escaped: True
run requests sent after preview: 0
empty paste      400 Paste at least one company name.
11 names         400 11 names is more than the limit of 10.
bogus token      400 This preview has expired. Start over.
run requests sent after controls: 0
preview for run  4 rows, estimate 4
run 200 [('Notion', 'complete', 'notion.com'), ('Café Rouge', 'complete', 'caferouge.com'), ('Linear', 'complete', 'linear.app'), ('Vercel', 'complete', 'vercel.com')]
every row complete with a domain: True | rows: 4
run requests sent after run: 1
second press     200 repeated notice: True
run requests sent after second press: 1

The pasted list for the preview contained six lines: Notion, notion (the same name with different case and spacing), a blank line, Café Rouge, <b>Figma</b> and Linear. The page reported 4 rows and 2 lines removed, escaped the markup as <b>Figma</b>, and run_requests_sent was still 0. The empty paste, the 11-name list and the unknown token each returned 400 and left the count at 0, which is the negative control for the claim that the page only runs after a valid preview token.

The run used a fresh preview of Notion, notion, Café Rouge, Linear and Vercel, reduced to 4 rows. Every row came back complete with a domain, and the count was 1. The second press of the same token returned the saved results and the count stayed at 1.

Two simultaneous presses of one token, from race_test.py. The first output is from a variant of app.py that checked the saved result without a lock. The second is from the version shown above, with a lock per token:

{'simultaneous_presses': 2, 'http_codes': [200, 200], 'run_requests_sent_by_these_two': 2}

{'simultaneous_presses': 2, 'http_codes': [200, 200], 'run_requests_sent_by_these_two': 1}

The properties that define a pass are that run_requests_sent is 0 after the preview and after every rejection, that the real run leaves it at 1, that a second press leaves it at 1, that two simultaneous presses add exactly 1, and that every row in the table has status complete and a non-empty domain. The domains are one run on 2026-10-20 and are not expected values.

How it works

The preview is a server-side step that reads the balance and stores the cleaned list under a random token. The run form carries only the token, so the number the teammate approved is the number that runs. Duplicates and blank lines are removed before the estimate is computed, so the estimate matches the rows actually sent.

A run request is the only call that spends credits. The first press stores its results on the token's entry, and later presses return that copy. The lock around the check and the run makes the check and the save one step for a given token.

Common issues

Pressing the button twice can run twice

Without the lock, two presses that arrive together both find no saved result and each sends a run request. In the unlocked variant, run_requests_sent rose by 2 for one preview. A second press after the first had finished was already handled by the saved result. A browser double click sends two requests close together, which is the case the lock covers.

The page waits while the lookup runs

The /run request returns only when the results are in, so the browser waits for the whole lookup. The page is capped at 10 names to keep that wait short; the wait for 10 names was not timed. A longer list needs the run to start in the background and the page to poll for it, which this example does not do.

Previews are kept in memory

The token store lives in the Flask process. Restarting it invalidates pending previews, and the page then shows This preview has expired. Start over. Two worker processes would not share the store, so run one process.

The estimate is a planning figure

PER_ROW is 1, taken from single-row runs of this routine that moved the balance by 1 credit. The page does not look up a price, because the REST API has no routine price endpoint. The balance shown is the whole workspace's balance and can change for reasons unrelated to this page.

Not covered

There is no login. Anyone who can open the page can start a run, so it should be served on localhost or behind the team's own access control. A routine with more than one input was not tried, and a row that fails or returns nothing was not seen in these runs, so the table's handling of a non-complete status was not exercised.

Next steps

verification:
  status: verified
  tested_at: "2026-10-20"
  product_version: "public/v0"
  command: "python3 app.py (one terminal); python3 verify_app.py; python3 race_test.py"

verification:
  status: verified
  tested_at: "2026-10-20"
  product_version: "public/v0"
  command: "python3 app.py (one terminal); python3 verify_app.py; python3 race_test.py"
  expected_result: "run_requests_sent is 0 after previews and rejections, 1 after the real run and a second press, and rises by exactly 1 for two simultaneous presses. Every row in the results table is complete with a non-empty domain."

Previewing Credit Cost Before a Run over REST with Python