Trigger Outbound Sequences from Clay Hiring Signals

Scheduler: every N seconds
    ↓
POST /routines/{routine_id}/run     (Company Job Openings, up to 100 accounts)
    ↓
GET /routines/run/{run_id}/results  (poll until complete)
    ↓
Derive signal_strength from real job data
    ↓
signal_strength >= threshold?  → POST to sequencer

Implement a Flask service that detects hiring signals for tracked accounts
using Clay's "Company Job Openings" managed function and forwards qualifying
accounts to an outbound sequencer.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- Find the real routine id for "Company Job Openings" via your workspace's
  routine catalog.
- After a live run, inspect the ACTUAL result JSON. Do not assume the result
  is keyed by the routine's display name -- confirm the real key from a live
  response body.
- The routine does not return a pre-built signal_strength field. Derive your
  own 0-1 signal score from real fields it does return (e.g. total job count,
  how many postings are recent), and state your exact formula in code.
- Poll for completion, filter on your derived score against a threshold, and
  forward qualifying accounts to SEQUENCER_WEBHOOK_URL.
- Run the verification step below before finishing.

mkdir outbound-sequence-trigger && cd outbound-sequence-trigger
python -m venv venv && source venv/bin/activate
pip install flask requests

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

clay routines get function:t_XXXXXXXXXXXXXXXXXXXX

CLAY_API_KEY=
CLAY_ROUTINE_ID_JOB_OPENINGS=function:t_XXXXXXXXXXXXXXXXXXXX
SEQUENCER_WEBHOOK_URL=https://your-sequencer.example.com/intake
TRACKED_ACCOUNTS_PATH=tracked_accounts.csv
MIN_SIGNAL_STRENGTH=0.6

import csv
import json
import os
import threading
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

import requests
from flask import Flask, jsonify

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_JOB_OPENINGS"]
SEQUENCER_URL = os.environ["SEQUENCER_WEBHOOK_URL"]
TRACKED_ACCOUNTS_PATH = os.environ.get("TRACKED_ACCOUNTS_PATH", "tracked_accounts.csv")
MIN_SIGNAL_STRENGTH = float(os.environ.get("MIN_SIGNAL_STRENGTH", 0.6))
RUN_STORE_PATH = Path("run_store.json")
_store_lock = threading.Lock()


def load_run_store():
    return json.loads(RUN_STORE_PATH.read_text()) if RUN_STORE_PATH.exists() else {}


def save_run_store(store):
    RUN_STORE_PATH.write_text(json.dumps(store, indent=2))


def load_tracked_accounts(path):
    with open(path) as f:
        return [row for row in csv.DictReader(f) if row.get("domain")]


def start_run(accounts):
    run_ids = []
    for i in range(0, len(accounts), 100):
        chunk = accounts[i:i + 100]
        items = [{"id": a["domain"], "inputs": {"Company Domain": a["domain"]}} for a in chunk]
        resp = requests.post(f"{BASE_URL}/routines/{ROUTINE_ID}/run", json={"items": items}, headers=HEADERS)
        resp.raise_for_status()
        run_ids.append(resp.json()["routineRunId"])
    return run_ids


def fetch_results(run_id):
    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 [], False
    return body.get("data", []), True


def compute_signal_strength(result):
    """
    Derive a 0-1 hiring-momentum signal from real Company Job Openings fields.
    Clay does NOT return a native signal_strength -- this formula is our own:
      volume  = min(total_job_count / 25, 1.0)
      recency = fraction of postings first_seen_at within the last 30 days
      signal_strength = 0.6 * volume + 0.4 * recency
    """
    jobs = result.get("Find Open Jobs") or {}  # confirmed real result key -- not the display name
    total = jobs.get("total_job_count") or 0
    postings = jobs.get("jobs") or jobs.get("data") or []
    cutoff = datetime.now(timezone.utc) - timedelta(days=30)
    recent = 0
    for p in postings:
        seen = p.get("first_seen_at")
        if seen:
            try:
                if datetime.fromisoformat(seen.replace("Z", "+00:00")) >= cutoff:
                    recent += 1
            except ValueError:
                continue
    volume = min(total / 25, 1.0)
    recency = (recent / len(postings)) if postings else 0.0
    return round(0.6 * volume + 0.4 * recency, 3)


def send_to_sequencer(domain, signal_strength):
    resp = requests.post(SEQUENCER_URL, json={"domain": domain, "signal_strength": signal_strength}, timeout=10)
    resp.raise_for_status()


@app.route("/start-once", methods=["POST"])
def start_once():
    accounts = load_tracked_accounts(TRACKED_ACCOUNTS_PATH)
    run_ids = start_run(accounts)
    with _store_lock:
        store = load_run_store()
        for rid in run_ids:
            store[rid] = {"processed": False}
        save_run_store(store)
    return jsonify({"status": "started", "accounts_submitted": len(accounts)})


@app.route("/poll-once", methods=["POST"])
def poll_once():
    with _store_lock:
        store = load_run_store()
    triggered = 0
    for run_id, state in store.items():
        if state.get("processed"):
            continue
        items, complete = fetch_results(run_id)
        if not complete:
            continue
        for item in items:
            if item["status"] != "complete":
                continue
            strength = compute_signal_strength(item.get("result") or {})
            if strength >= MIN_SIGNAL_STRENGTH:
                send_to_sequencer(item["id"], strength)
                triggered += 1
        with _store_lock:
            store = load_run_store()
            store[run_id]["processed"] = True
            save_run_store(store)
    return jsonify({"status": "processed", "triggered": triggered})


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

echo "domain
clay.com" > tracked_accounts.csv
python app.py

curl -X POST http://localhost:5002/start-once
sleep 15
curl -X POST http://localhost:5002/poll-once

{ "status": "processed", "triggered": 1 }

verification:
  status: verified
  tested_at: "2026-08-14"
  product_version: "public/v0"
  command: "python app.py && curl -X POST .../start-once && curl -X POST .../poll-once"
  expected_result: "real nonzero signal_strength forwarded to sequencer for an account with real job postings"