Source Leads by Title, Seniority, and Location with Clay

POST /source-leads (title_keywords, seniority, countries, max_results)
    ↓
GET /search/filters-mode/fields?source_type=people   (validate filters live)
    ↓
POST /search/filters-mode          (create)
    ↓
POST /search/filters-mode/{id}/run (page until max_results or no more)
    ↓
leads.csv

Implement a Flask service that sources a lead list by title, seniority, and
location using Clay's Search API filters-mode for people.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- Call GET /search/filters-mode/fields?source_type=people first and confirm
  every filter field name and, where present, its allowed_values before
  using it. Do not assume field names or valid values.
- Confirmed real people filter fields include: job_title_keywords (free
  text, no allowed_values -- a typo returns 0 rows silently, not an error),
  job_title_seniority_levels_v2 (controlled vocabulary, includes real values
  like "vp" and "head"), location_countries_include (free text, no
  allowed_values -- confirm the exact expected string format, e.g.
  "United States", empirically).
- Create the search with POST /search/filters-mode, then page with
  POST /search/filters-mode/{search_id}/run until max_results is reached or
  has_more is false, deduplicating by a stable identifier across pages.
- Export the collected rows to leads.csv.
- Run the verification step below before finishing.

mkdir people-search-title-location && cd people-search-title-location
python -m venv venv && source venv/bin/activate
pip install flask requests

curl -s "https://api.clay.com/public/v0/search/filters-mode/fields?source_type=people" \
  -H "clay-api-key: $CLAY_API_KEY"

CLAY_API_KEY=

"""
Lead sourcing service over the Clay Search API (filters-mode, source_type=people).

POST /source-leads
  {"title_keywords": [...], "seniority": [...], "countries": [...], "max_results": int}

Every filter name is validated against the LIVE GET /search/filters-mode/fields
response before the search is created -- no field names are hardcoded blindly.
job_title_keywords and location_countries_include are free-text fields with no
allowed_values -- a typo there returns 0 rows silently, not an error, so this
is flagged in the response as a validation_note rather than pretended away.
"""

import csv
import os

import requests
from flask import Flask, jsonify, request

app = Flask(__name__)

BASE_URL = "https://api.clay.com/public/v0"
HEADERS = {"clay-api-key": os.environ["CLAY_API_KEY"], "Content-Type": "application/json"}


def get_people_fields():
    resp = requests.get(f"{BASE_URL}/search/filters-mode/fields", params={"source_type": "people"}, headers=HEADERS)
    resp.raise_for_status()
    return {f["name"]: f for f in resp.json()["fields"]}


def build_filters(title_keywords, seniority, countries, fields):
    filters, notes = {}, []
    if title_keywords:
        filters["job_title_keywords"] = title_keywords
        if not fields.get("job_title_keywords", {}).get("allowed_values"):
            notes.append("job_title_keywords is free text (no allowed_values in /fields); values cannot be pre-validated and a typo returns 0 rows")
    if seniority:
        filters["job_title_seniority_levels_v2"] = seniority
    if countries:
        filters["location_countries_include"] = countries
        if not fields.get("location_countries_include", {}).get("allowed_values"):
            notes.append("location_countries_include is free text; confirm exact expected format (e.g. 'United States') empirically")
    return filters, notes


def create_search(filters):
    resp = requests.post(f"{BASE_URL}/search/filters-mode", json={"source_type": "people", "filters": filters}, headers=HEADERS)
    resp.raise_for_status()
    return resp.json()["search_id"]


def collect_results(search_id, max_results):
    rows, seen_ids, pages = [], set(), []
    while len(rows) < max_results:
        resp = requests.post(
            f"{BASE_URL}/search/filters-mode/{search_id}/run",
            json={"limit": min(20, max_results - len(rows))},
            headers=HEADERS,
        )
        resp.raise_for_status()
        page = resp.json()
        page_rows = page["data"]
        new_rows = [r for r in page_rows if r.get("clay_profile_id") not in seen_ids]
        for r in new_rows:
            seen_ids.add(r.get("clay_profile_id"))
        rows.extend(new_rows)
        pages.append({"page": len(pages) + 1, "returned": len(page_rows), "new_after_dedupe": len(new_rows), "cumulative": len(rows), "has_more": page.get("has_more"), "http_status": 200})
        if not page.get("has_more"):
            break
    return rows[:max_results], pages


@app.route("/source-leads", methods=["POST"])
def source_leads():
    payload = request.get_json(silent=True) or {}
    title_keywords = payload.get("title_keywords", [])
    seniority = payload.get("seniority", [])
    countries = payload.get("countries", [])
    max_results = payload.get("max_results", 25)

    fields = get_people_fields()
    filters, validation_note = build_filters(title_keywords, seniority, countries, fields)
    search_id = create_search(filters)
    rows, pages = collect_results(search_id, max_results)

    with open("leads.csv", "w", newline="") as f:
        if rows:
            writer = csv.DictWriter(f, fieldnames=sorted({k for row in rows for k in row}))
            writer.writeheader()
            writer.writerows(rows)

    return jsonify({
        "search_id": search_id,
        "filters_sent": filters,
        "row_count": len(rows),
        "requested_max_results": max_results,
        "pages": pages,
        "validation_note": validation_note,
        "csv_path": "leads.csv",
    })


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

python app.py

curl -s -X POST http://localhost:5018/source-leads \
  -H "Content-Type: application/json" \
  -d '{"title_keywords": ["sales"], "seniority": ["vp", "head"], "countries": ["United States"], "max_results": 25}'

{
  "search_id": "search_0tjzcmw3ZzS25x65Hqf",
  "filters_sent": {"job_title_keywords": ["sales"], "job_title_seniority_levels_v2": ["vp", "head"], "location_countries_include": ["United States"]},
  "row_count": 25
}

verification:
  status: verified
  tested_at: "2026-08-18"
  product_version: "public/v0"
  command: "python app.py && curl -s -X POST http://localhost:5018/source-leads -d '{\"title_keywords\":[\"sales\"],\"seniority\":[\"vp\",\"head\"],\"countries\":[\"United States\"],\"max_results\":25}'"
  expected_result: "HTTP 200, row_count=25, real people matching VP/Head sales titles in the United States"