Score HubSpot leads based on firmographic and technographic fit using an agent skill

---
name: lead-scoring
description: Score or re-score HubSpot contacts based on firmographic and technographic fit
disable-model-invocation: true
allowed-tools: Bash(python *)
---
 
Score HubSpot contacts based on ICP fit. Writes the score to the `icp_fit_score` property.
 
Usage:
- `/lead-scoring` — score all unscored contacts
- `/lead-scoring --all` — re-score every contact (use after changing the scoring model)
 
Run: `python $SKILL_DIR/scripts/score_leads.py $@`

#!/usr/bin/env python3
"""Score HubSpot contacts based on firmographic/technographic fit."""
import os, sys, requests
 
HUBSPOT_TOKEN = os.environ.get("HUBSPOT_TOKEN")
if not HUBSPOT_TOKEN:
    print("ERROR: Set HUBSPOT_TOKEN env var")
    sys.exit(1)
 
HEADERS = {"Authorization": f"Bearer {HUBSPOT_TOKEN}", "Content-Type": "application/json"}
 
# --- Scoring model (edit these to match your ICP) ---
def score_contact(props):
    score = 0
 
    # Company size (0-30)
    emp = int(props.get("numberofemployees") or 0)
    if 200 <= emp <= 2000: score += 30
    elif 50 <= emp < 200: score += 20
    elif 2000 < emp <= 10000: score += 15
    elif emp > 0: score += 5
 
    # Industry (0-25)
    industry = (props.get("industry") or "").lower()
    ideal = ["saas", "technology", "software", "computer software"]
    good = ["financial services", "consulting", "marketing"]
    if industry in ideal: score += 25
    elif industry in good: score += 15
    elif industry: score += 5
 
    # Seniority (0-30)
    title = (props.get("jobtitle") or "").lower()
    if any(t in title for t in ["ceo","cto","cfo","coo","cmo","cro","chief"]): score += 30
    elif any(t in title for t in ["vp","vice president","head of"]): score += 25
    elif "director" in title: score += 20
    elif any(t in title for t in ["manager","lead"]): score += 10
 
    # Source (0-15)
    source = (props.get("hs_analytics_source") or "").lower()
    source_scores = {"organic_search":15,"direct_traffic":12,"referrals":10,"paid_search":8,"social_media":5}
    score += source_scores.get(source, 0)
 
    return min(score, 100)
 
# --- Fetch contacts ---
rescore_all = "--all" in sys.argv
filter_groups = [] if rescore_all else [{"filters": [
    {"propertyName": "icp_fit_score", "operator": "NOT_HAS_PROPERTY"}
]}]
 
contacts = []
after = None
while True:
    body = {
        "properties": ["firstname","lastname","jobtitle","company","numberofemployees","industry","hs_analytics_source"],
        "limit": 100,
    }
    if filter_groups:
        body["filterGroups"] = filter_groups
    if after:
        body["after"] = after
 
    resp = requests.post("https://api.hubapi.com/crm/v3/objects/contacts/search", headers=HEADERS, json=body)
    resp.raise_for_status()
    data = resp.json()
    contacts.extend(data.get("results", []))
    after = data.get("paging", {}).get("next", {}).get("after")
    if not after:
        break
 
if not contacts:
    print("No contacts to score")
    sys.exit(0)
 
print(f"Scoring {len(contacts)} contacts...")
 
# --- Score and update ---
scored = 0
for contact in contacts:
    props = contact["properties"]
    s = score_contact(props)
    requests.patch(
        f"https://api.hubapi.com/crm/v3/objects/contacts/{contact['id']}",
        headers=HEADERS,
        json={"properties": {"icp_fit_score": str(s)}}
    ).raise_for_status()
    name = f"{props.get('firstname','')} {props.get('lastname','')}".strip()
    print(f"  {name}: {s}/100")
    scored += 1
 
print(f"\nDone. Scored {scored} contacts.")

# Score unscored contacts
/lead-scoring
 
# Re-score all contacts after updating the model
/lead-scoring --all