Score Leads on Firmographic Signals with Clay

POST /score-lead (account_id, domain)
    ↓
POST /routines/{routine_id}/run       (Enrich Company)
    ↓
GET /routines/run/{run_id}/results    (poll until complete)
    ↓
Extract signals → normalize → apply weights → composite_score

Implement a FastAPI service that computes a composite lead score using Clay's
"Enrich Company" managed function for firmographic signals.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- Find the real routine id for "Enrich Company" via your workspace's routine
  catalog before writing code that assumes a fixed id.
- Submit the account (by domain) as a single-item routine run, poll until
  complete, and inspect the ACTUAL response schema before mapping any field
  to a scoring dimension -- do not assume field names like
  "headcount_growth_pct" or "ai_fit_score" exist without confirming them
  against a live response.
- Map whatever real signals the routine returns (e.g. employee count,
  follower count, business stage) into normalized 0-100 scores, and state
  your normalization/weighting logic explicitly in code comments since it is
  scoring policy, not an API fact.
- Run the verification step below before finishing.

mkdir lead-scoring-multi-signal && cd lead-scoring-multi-signal
python -m venv venv && source venv/bin/activate
pip install fastapi uvicorn requests pydantic

clay routines list | grep -A2 '"Enrich Company"'
# -> "id": "function:t_XXXXXXXXXXXXXXXXXXXX"

clay routines get function:t_XXXXXXXXXXXXXXXXXXXX

CLAY_API_KEY=
CLAY_ROUTINE_ID_ENRICH_COMPANY=function:t_XXXXXXXXXXXXXXXXXXXX
SCORE_WEIGHTS={"firmographic":0.4,"momentum":0.3,"stage":0.3}

import json
import os
import time

import requests
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

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_ENRICH_COMPANY"]
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL_SECONDS", 3))

# Confirmed real fields from a live Enrich Company response (2026-08-14).
# headcount_growth_pct and ai_fit_score do NOT exist -- do not reintroduce them.
BUSINESS_STAGE_SCORES = {"Seed": 20, "Early Stage": 40, "Growth Stage": 80, "Mature": 60, "Public": 50}


class LeadRequest(BaseModel):
    account_id: str
    domain: str


def get_weights():
    default = {"firmographic": 0.4, "momentum": 0.3, "stage": 0.3}
    raw = os.environ.get("SCORE_WEIGHTS")
    return json.loads(raw) if raw else default


def run_enrichment(req, max_attempts=60):
    resp = requests.post(
        f"{BASE_URL}/routines/{ROUTINE_ID}/run",
        json={"items": [{"id": req.account_id, "inputs": {"Company Identifier": req.domain}}]},
        headers=HEADERS,
    )
    resp.raise_for_status()
    run_id = resp.json()["routineRunId"]

    for _ in range(max_attempts):
        resp = requests.get(f"{BASE_URL}/routines/run/{run_id}/results", headers=HEADERS)
        resp.raise_for_status()
        body = resp.json()
        if body["status"] == "complete":
            item = body["data"][0]
            if item["status"] == "failed":
                raise HTTPException(502, f"Enrichment failed: {(item.get('error') or {}).get('message', '')}")
            return (item.get("result") or {}).get("Enrich Company") or {}
        time.sleep(POLL_INTERVAL)
    raise HTTPException(504, f"Routine run {run_id} did not complete in time")


def extract_signals(result):
    return {
        "firmographic": result.get("employee_count") or 0,
        "momentum": result.get("follower_count") or 0,
        "stage": BUSINESS_STAGE_SCORES.get((result.get("derived_datapoints") or {}).get("business_stage"), 30),
    }


def normalize(signals):
    return {
        "firmographic": min((signals["firmographic"] or 0) / 100, 100),
        "momentum": min((signals["momentum"] or 0) / 2000, 100),
        "stage": signals["stage"],
    }


def compute_composite_score(normalized, weights):
    return round(sum(normalized[k] * weights[k] for k in weights), 1)


@app.post("/score-lead")
def score_lead(req: LeadRequest):
    result = run_enrichment(req)
    normalized = normalize(extract_signals(result))
    score = compute_composite_score(normalized, get_weights())
    return {"account_id": req.account_id, "signals": normalized, "composite_score": score}

uvicorn app:app --reload

curl -X POST http://localhost:8000/score-lead \
  -H "Content-Type: application/json" \
  -d '{"account_id": "acct-1", "domain": "clay.com"}'

{
  "account_id": "acct-1",
  "signals": { "firmographic": 14.94, "momentum": 33.99, "stage": 80 },
  "composite_score": 45.2
}

verification:
  status: verified
  tested_at: "2026-08-14"
  product_version: "public/v0"
  command: "uvicorn app:app --reload && curl -X POST http://localhost:8000/score-lead -d '{...}'"
  expected_result: "real composite_score computed from real Enrich Company data for clay.com"