Find and verify emails for HubSpot prospects using Apollo and Hunter with an agent skill

mkdir -p .claude/skills/find-emails/scripts

---
name: find-emails
description: Finds and verifies work emails for HubSpot contacts missing email addresses. Uses Apollo for email lookup and Hunter for deliverability verification. Writes verified emails back to HubSpot.
disable-model-invocation: true
allowed-tools: Bash(python *)
---
 
Find and verify emails for HubSpot prospects:
 
1. Run: `python $SKILL_DIR/scripts/find_emails.py`
2. Review the per-contact results
3. Check the summary for verified, risky, and not-found counts

#!/usr/bin/env python3
"""
Email Finder: HubSpot → Apollo (find) → Hunter (verify) → HubSpot (update)
Finds contacts missing emails, looks up via Apollo, verifies with Hunter.
"""
import os
import sys
import time
 
try:
    import requests
except ImportError:
    os.system("pip install requests -q")
    import requests
 
HUBSPOT_TOKEN = os.environ.get("HUBSPOT_TOKEN")
APOLLO_API_KEY = os.environ.get("APOLLO_API_KEY")
HUNTER_API_KEY = os.environ.get("HUNTER_API_KEY")
 
if not all([HUBSPOT_TOKEN, APOLLO_API_KEY, HUNTER_API_KEY]):
    print("ERROR: Set HUBSPOT_TOKEN, APOLLO_API_KEY, and HUNTER_API_KEY")
    sys.exit(1)
 
HS_HEADERS = {"Authorization": f"Bearer {HUBSPOT_TOKEN}", "Content-Type": "application/json"}
 
# --- Find contacts without emails ---
print("Searching for contacts without email addresses...")
contacts = []
after = 0
while True:
    resp = requests.post(
        "https://api.hubapi.com/crm/v3/objects/contacts/search",
        headers=HS_HEADERS,
        json={
            "filterGroups": [{"filters": [
                {"propertyName": "email", "operator": "NOT_HAS_PROPERTY"},
                {"propertyName": "firstname", "operator": "HAS_PROPERTY"},
                {"propertyName": "company", "operator": "HAS_PROPERTY"},
            ]}],
            "properties": ["firstname", "lastname", "company"],
            "limit": 100,
            "after": after,
        }
    )
    resp.raise_for_status()
    data = resp.json()
    contacts.extend(data["results"])
    if data.get("paging", {}).get("next"):
        after = data["paging"]["next"]["after"]
    else:
        break
 
print(f"Found {len(contacts)} contacts without emails\n")
 
# --- Process each contact ---
stats = {"verified": 0, "risky": 0, "not_found": 0}
 
for contact in contacts:
    props = contact["properties"]
    name = f"{props.get('firstname', '')} {props.get('lastname', '')}".strip()
    company = props.get("company", "")
 
    # Find email via Apollo
    apollo_resp = requests.post(
        "https://api.apollo.io/api/v1/people/match",
        headers={"x-api-key": APOLLO_API_KEY, "Content-Type": "application/json"},
        json={
            "first_name": props.get("firstname", ""),
            "last_name": props.get("lastname", ""),
            "organization_name": company,
        }
    )
    apollo_resp.raise_for_status()
    person = apollo_resp.json().get("person")
 
    if not person or not person.get("email"):
        stats["not_found"] += 1
        print(f"  {name} @ {company} -> not found")
        time.sleep(0.5)
        continue
 
    email = person["email"]
    email_status = person.get("email_status", "unknown")
 
    # Skip Hunter if Apollo already verified
    if email_status == "verified":
        verification = "verified"
        source = "apollo"
    else:
        # Verify with Hunter
        hunter_resp = requests.get(
            "https://api.hunter.io/v2/email-verifier",
            params={"email": email, "api_key": HUNTER_API_KEY}
        )
        hunter_resp.raise_for_status()
        verification = hunter_resp.json()["data"]["result"]
        source = "apollo+hunter"
 
    # Update HubSpot
    update_props = {"email_source": source, "email_verification_status": verification}
    if verification in ("deliverable", "verified"):
        update_props["email"] = email
        stats["verified"] += 1
        print(f"  {name} @ {company} -> {email} ({source})")
    elif verification == "risky":
        update_props["email"] = email  # store but flag
        stats["risky"] += 1
        print(f"  {name} @ {company} -> {email} (RISKY)")
    else:
        stats["not_found"] += 1
        print(f"  {name} @ {company} -> {email} (UNDELIVERABLE, skipped)")
 
    requests.patch(
        f"https://api.hubapi.com/crm/v3/objects/contacts/{contact['id']}",
        headers=HS_HEADERS,
        json={"properties": update_props}
    ).raise_for_status()
 
    time.sleep(0.5)
 
print(f"\nDone. Verified: {stats['verified']}, Risky: {stats['risky']}, Not found: {stats['not_found']}")

# Via Claude Code
/find-emails
 
# Or directly
python .claude/skills/find-emails/scripts/find_emails.py

# .github/workflows/find-emails.yml
name: Find and Verify Emails
on:
  schedule:
    - cron: '0 11 * * *'  # Daily at 6 AM ET
  workflow_dispatch: {}
jobs:
  find:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install requests
      - run: python .claude/skills/find-emails/scripts/find_emails.py
        env:
          HUBSPOT_TOKEN: ${{ secrets.HUBSPOT_TOKEN }}
          APOLLO_API_KEY: ${{ secrets.APOLLO_API_KEY }}
          HUNTER_API_KEY: ${{ secrets.HUNTER_API_KEY }}