Batch-enrich Salesforce contacts with Apollo using an agent skill

Prerequisites

Claude Code or another agent that supports the Agent Skills standard
Salesforce instance URL stored as SALESFORCE_INSTANCE_URL environment variable
Salesforce access token stored as SALESFORCE_ACCESS_TOKEN environment variable
Apollo API key stored as APOLLO_API_KEY environment variable

Overview

This agent skill queries Salesforce for leads or contacts missing key enrichment fields (title, phone, industry), enriches them in batch via Apollo's People Enrich API, and writes the results back to Salesforce. Unlike the n8n approach which triggers on new records, this handles backfill — enriching existing records that were never enriched.

Step 1: Create the skill

Create .claude/skills/sf-apollo-enrich/SKILL.md:

---
name: sf-apollo-enrich
description: Batch-enrich un-enriched Salesforce leads with Apollo contact data
disable-model-invocation: true
allowed-tools: Bash(python *)
---
 
Find Salesforce Leads missing key fields and enrich them with Apollo data.
 
Run: `python $SKILL_DIR/scripts/enrich.py`

Step 2: Write the script

Create .claude/skills/sf-apollo-enrich/scripts/enrich.py:

#!/usr/bin/env python3
import os, sys, time, requests
 
INSTANCE_URL = os.environ.get("SALESFORCE_INSTANCE_URL")
ACCESS_TOKEN = os.environ.get("SALESFORCE_ACCESS_TOKEN")
APOLLO_KEY = os.environ.get("APOLLO_API_KEY")
 
if not all([INSTANCE_URL, ACCESS_TOKEN, APOLLO_KEY]):
    print("ERROR: Set SALESFORCE_INSTANCE_URL, SALESFORCE_ACCESS_TOKEN, APOLLO_API_KEY")
    sys.exit(1)
 
SF_HEADERS = {"Authorization": f"Bearer {ACCESS_TOKEN}", "Content-Type": "application/json"}
APOLLO_HEADERS = {"x-api-key": APOLLO_KEY, "Content-Type": "application/json"}
 
# Find un-enriched leads
soql = (
    "SELECT Id, Email, FirstName, LastName, Company "
    "FROM Lead "
    "WHERE Title = null AND Email != null "
    "ORDER BY CreatedDate DESC LIMIT 50"
)
resp = requests.get(
    f"{INSTANCE_URL}/services/data/v59.0/query",
    headers=SF_HEADERS,
    params={"q": soql},
)
resp.raise_for_status()
leads = resp.json().get("records", [])
 
if not leads:
    print("No un-enriched leads found")
    sys.exit(0)
 
print(f"Found {len(leads)} un-enriched leads. Enriching...")
 
enriched = 0
skipped = 0
 
for lead in leads:
    email = lead.get("Email")
    if not email:
        skipped += 1
        continue
 
    # Call Apollo People Enrich
    try:
        apollo_resp = requests.post(
            "https://api.apollo.io/v1/people/enrich",
            headers=APOLLO_HEADERS,
            json={"email": email},
        )
 
        if apollo_resp.status_code == 429:
            print("Rate limited — waiting 60 seconds")
            time.sleep(60)
            apollo_resp = requests.post(
                "https://api.apollo.io/v1/people/enrich",
                headers=APOLLO_HEADERS,
                json={"email": email},
            )
 
        apollo_resp.raise_for_status()
        person = apollo_resp.json().get("person")
 
        if not person:
            skipped += 1
            continue
 
        # Build update payload
        update = {}
        if person.get("title"):
            update["Title"] = person["title"]
        if person.get("linkedin_url"):
            update["LinkedIn_URL__c"] = person["linkedin_url"]
        if person.get("phone_numbers"):
            phone = person["phone_numbers"][0].get("sanitized_number")
            if phone:
                update["Phone"] = phone
        if person.get("departments"):
            update["Department"] = person["departments"][0]
        if person.get("organization", {}).get("name"):
            update["Company"] = person["organization"]["name"]
 
        if not update:
            skipped += 1
            continue
 
        # Update Salesforce Lead
        sf_resp = requests.patch(
            f"{INSTANCE_URL}/services/data/v59.0/sobjects/Lead/{lead['Id']}",
            headers=SF_HEADERS,
            json=update,
        )
        sf_resp.raise_for_status()
        enriched += 1
        print(f"  ✓ {lead['FirstName']} {lead['LastName']} — {person.get('title', 'no title')}")
 
    except requests.exceptions.RequestException as e:
        print(f"  ✗ {lead['FirstName']} {lead['LastName']} — {e}")
        skipped += 1
 
    # Respect Apollo rate limit (100/min)
    time.sleep(0.7)
 
print(f"\nDone. Enriched {enriched}/{len(leads)} leads. Skipped {skipped}.")

Step 3: Run it

/sf-apollo-enrich

A successful run looks like:

Found 50 un-enriched leads. Enriching...
  ✓ Sarah Chen — VP of Engineering
  ✓ Mike Johnson — Director of Sales
  ✓ Lisa Park — Head of Marketing
  ✗ David Kim — 404 Not Found
  ...
 
Done. Enriched 47/50 leads. Skipped 3.

Step 4: Schedule (optional)

For daily enrichment of new un-enriched leads:

# crontab — run every day at 9 AM
0 9 * * * cd /path/to/project && python .claude/skills/sf-apollo-enrich/scripts/enrich.py

When to use this approach

You need to backfill existing leads that were never enriched
You want to run enrichment on demand before a sales blitz or campaign
You prefer a script you can modify and version control
You don't want to maintain n8n workflows or Salesforce Connected Apps for n8n