Auto-enrich new HubSpot contacts with Apollo using an agent skill

mkdir -p .claude/skills/enrich-contacts/scripts

---
name: enrich-contacts
description: Enriches new HubSpot contacts with Apollo data. Finds contacts created recently, calls Apollo for job title, company, phone, and LinkedIn, then writes enriched fields back to HubSpot.
disable-model-invocation: true
allowed-tools: Bash(python *)
---
 
Enrich recently created HubSpot contacts with Apollo data:
 
1. Run: `python $SKILL_DIR/scripts/enrich.py`
2. Review the output for enrichment results and any errors
3. Confirm the number of contacts enriched

#!/usr/bin/env python3
"""
HubSpot Contact Enrichment via Apollo
Finds new contacts, enriches via Apollo People Match, writes back to HubSpot.
"""
import os
import sys
import time
from datetime import datetime, timedelta, timezone
 
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")
 
if not all([HUBSPOT_TOKEN, APOLLO_API_KEY]):
    print("ERROR: Set HUBSPOT_TOKEN and APOLLO_API_KEY environment variables")
    sys.exit(1)
 
HS_HEADERS = {"Authorization": f"Bearer {HUBSPOT_TOKEN}", "Content-Type": "application/json"}
APOLLO_HEADERS = {"x-api-key": APOLLO_API_KEY, "Content-Type": "application/json"}
 
# --- Fetch new contacts ---
since = int((datetime.now(timezone.utc) - timedelta(hours=24)).timestamp() * 1000)
print(f"Searching for contacts created in the last 24 hours...")
 
contacts = []
after = 0
while True:
    resp = requests.post(
        "https://api.hubapi.com/crm/v3/objects/contacts/search",
        headers=HS_HEADERS,
        json={
            "filterGroups": [{"filters": [{
                "propertyName": "createdate",
                "operator": "GTE",
                "value": str(since)
            }]}],
            "properties": ["email", "firstname", "lastname", "jobtitle", "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)} new contacts")
 
# --- Enrich and update ---
enriched = 0
skipped = 0
 
for contact in contacts:
    email = contact["properties"].get("email")
    existing_title = contact["properties"].get("jobtitle")
 
    if not email:
        skipped += 1
        continue
 
    if existing_title:
        skipped += 1
        continue
 
    # Skip personal email domains
    domain = email.split("@")[-1].lower()
    if domain in ("gmail.com", "yahoo.com", "hotmail.com", "outlook.com", "aol.com"):
        skipped += 1
        continue
 
    # Call Apollo
    apollo_resp = requests.post(
        "https://api.apollo.io/api/v1/people/match",
        headers=APOLLO_HEADERS,
        json={"email": email}
    )
    apollo_resp.raise_for_status()
    person = apollo_resp.json().get("person")
 
    if not person:
        skipped += 1
        time.sleep(0.5)
        continue
 
    # Build update payload (only non-null fields)
    properties = {}
    if person.get("title"):
        properties["jobtitle"] = person["title"]
    if person.get("organization", {}).get("name"):
        properties["company"] = person["organization"]["name"]
    if person.get("phone_numbers") and person["phone_numbers"][0].get("sanitized_number"):
        properties["phone"] = person["phone_numbers"][0]["sanitized_number"]
    if person.get("linkedin_url"):
        properties["linkedin_url"] = person["linkedin_url"]
    if person.get("organization", {}).get("industry"):
        properties["industry"] = person["organization"]["industry"]
 
    if properties:
        update_resp = requests.patch(
            f"https://api.hubapi.com/crm/v3/objects/contacts/{contact['id']}",
            headers=HS_HEADERS,
            json={"properties": properties}
        )
        update_resp.raise_for_status()
        enriched += 1
        print(f"  Enriched: {email} -> {list(properties.keys())}")
 
    time.sleep(0.5)  # respect Apollo rate limits
 
print(f"\nDone. Enriched: {enriched}, Skipped: {skipped}")

# Via Claude Code
/enrich-contacts
 
# Or directly
python .claude/skills/enrich-contacts/scripts/enrich.py

# crontab -e — run every hour
0 * * * * cd /path/to/project && python .claude/skills/enrich-contacts/scripts/enrich.py >> /var/log/enrich.log 2>&1

name: Enrich HubSpot Contacts
on:
  schedule:
    - cron: '0 * * * *'
  workflow_dispatch: {}
jobs:
  enrich:
    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/enrich-contacts/scripts/enrich.py
        env:
          HUBSPOT_TOKEN: ${{ secrets.HUBSPOT_TOKEN }}
          APOLLO_API_KEY: ${{ secrets.APOLLO_API_KEY }}