Get Verified Email and Phone from a LinkedIn URL with Clay

POST /contact-details (linkedin_urls)
    ↓
POST /routines/{routine_id}/run   (Enrich Person and Find Contact Details)
    ↓
GET /routines/run/{run_id}/results (poll until complete)
    ↓
{ count, results: [{ linkedin_url, work_email, mobile_phone, name, title, org }] }

Implement a Flask service that gets verified work email and mobile phone for
a list of LinkedIn profile URLs using Clay's "Enrich Person and Find Contact
Details" managed function.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- Find the real routine id via your workspace's routine catalog.
- Confirmed real input schema: {"Social Profile URL": uri} (required only).
- Confirmed real result has three top-level keys: "Work Email" (string),
  "Mobile Phone" (string, may be partially masked), and "Enrich person"
  (lowercase "person" -- nested profile with name/title/org/country/
  headline/education/experience).
- COST NOTE: this routine costs approximately 18.2 credits per item, notably
  more expensive than most other Clay functions. Add an explicit batch-size
  guard (e.g. a MAX_ITEMS_PER_REQUEST env var) and reject oversized requests
  with a 413, rather than letting an accidental large batch run up cost
  silently.
- Handle absent/null values in the result gracefully -- do not raise a
  KeyError on a partial result.
- Run the verification step below before finishing.

mkdir verified-email-phone-from-linkedin && cd verified-email-phone-from-linkedin
python -m venv venv && source venv/bin/activate
pip install flask requests

clay routines list | grep -A2 '"Enrich Person and Find Contact Details"'

clay routines get function:t_XXXXXXXXXXXXXXXXXXXX

CLAY_API_KEY=
CLAY_ROUTINE_ID_CONTACT_DETAILS=function:t_XXXXXXXXXXXXXXXXXXXX
MAX_ITEMS_PER_REQUEST=10

"""
Rev-ops contact enrichment service.

POST /contact-details  {"linkedin_urls": ["https://www.linkedin.com/in/...", ...]}
  -> per-url {work_email, mobile_phone, name, title, org}

Backed by the Clay routine 'Enrich Person and Find Contact Details'.
Cost is ~18.2 credits per URL, so batch size is capped by
MAX_ITEMS_PER_REQUEST to avoid an accidentally expensive request.
"""

import logging
import os
import time

import requests
from flask import Flask, jsonify, request

logging.basicConfig(level=logging.INFO)
app = Flask(__name__)

BASE_URL = "https://api.clay.com/public/v0"
HEADERS = {"clay-api-key": os.environ["CLAY_API_KEY"], "Content-Type": "application/json"}
ROUTINE_ID = os.environ["CLAY_ROUTINE_ID_CONTACT_DETAILS"]
MAX_ITEMS_PER_REQUEST = int(os.environ.get("MAX_ITEMS_PER_REQUEST", 10))
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL_SECONDS", 10))
MAX_WAIT = int(os.environ.get("MAX_WAIT_SECONDS", 300))


def flatten(result):
    """Real result -> flat dict, tolerant of absent/null values everywhere."""
    result = result or {}
    person = result.get("Enrich person") or {}
    return {
        "work_email": result.get("Work Email"),
        "mobile_phone": result.get("Mobile Phone"),
        "name": person.get("name"),
        "title": person.get("title"),
        "org": person.get("org"),
    }


def submit(linkedin_urls):
    items = [{"id": f"c{i}", "inputs": {"Social Profile URL": url}} for i, url in enumerate(linkedin_urls)]
    resp = requests.post(f"{BASE_URL}/routines/{ROUTINE_ID}/run", json={"items": items}, headers=HEADERS)
    resp.raise_for_status()
    return resp.json()["routine_run_id"]


def poll(run_id):
    deadline = time.time() + MAX_WAIT
    while time.time() < deadline:
        resp = requests.get(f"{BASE_URL}/routines/run/{run_id}/results", headers=HEADERS)
        resp.raise_for_status()
        body = resp.json()
        if body["status"] == "complete":
            return body.get("data", [])
        time.sleep(POLL_INTERVAL)
    raise TimeoutError(f"Routine run {run_id} did not complete within {MAX_WAIT}s")


@app.route("/contact-details", methods=["POST"])
def contact_details():
    urls = (request.get_json(silent=True) or {}).get("linkedin_urls", [])
    if len(urls) > MAX_ITEMS_PER_REQUEST:
        return jsonify({"error": f"request exceeds MAX_ITEMS_PER_REQUEST={MAX_ITEMS_PER_REQUEST} (~18.2 credits/item)"}), 413

    run_id = submit(urls)
    items = poll(run_id)
    results = [
        {"linkedin_url": urls[i], **flatten(item.get("result"))}
        for i, item in enumerate(items)
    ]
    return jsonify({"count": len(results), "results": results})


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

python app.py

curl -s -X POST http://localhost:5011/contact-details \
  -H "Content-Type: application/json" \
  -d '{"linkedin_urls": ["https://www.linkedin.com/in/example"]}'

{
  "count": 1,
  "results": [
    {
      "linkedin_url": "https://www.linkedin.com/in/example-profile",
      "work_email": "[email protected]",
      "mobile_phone": "+1XX****1234",
      "name": "Jane Doe",
      "title": "Principal Solution Architect",
      "org": "Example Corp"
    }
  ]
}

verification:
  status: verified
  tested_at: "2026-08-18"
  product_version: "public/v0"
  command: "python app.py && curl -s -X POST http://localhost:5011/contact-details -d '{\"linkedin_urls\":[\"https://www.linkedin.com/in/example-profile\"]}'"
  expected_result: "HTTP 200 in ~94s, real work_email/mobile_phone/name/title/org populated for the tested profile"