keinsaasforever/gtm-pipeline-skills/gtm-people-enrichment

People Enrichment

Enrich contacts with work email, phone number, and/or LinkedIn URL. Returns an updated CSV.

Read `~/.claude/skills/gtm-pipeline/_shared/conventions.md` before executing.

When to Use

• You have a contact list and need emails and/or phone numbers
• Contacts from people-search need enrichment before outreach
• A user-provided CSV has names + companies but missing contact info

Inputs

Demo Mode

When invoked from the demo flow: email only, skip all phone providers entirely.

Cloudflare: BC and FE (like Pipe0) are Cloudflare-fronted — call them with curl + a browser User-Agent (conventions rule #8); never Python requests/urllib.

Provider Selection

Email

Ask the user which provider to use. Default recommendation:

Recommended flow (email waterfall): FE Enrich (v2) → Pipe0 waterfall → BC (last resort). FE is email-first; BC drops to last for email because its async email hit rate is low (~14% production SA). Max 2 attempts per source, then fall through. The order follows the email priority table above (for email, BC ranks below Pipe0, unlike the finder cadence in conventions).

Domain-Identity Cross-Check (critical)

Deliverability status is blind to identity. Providers return DELIVERABLE emails on the wrong company's domain (a genuine wrong-company hit — the exact "made-up address" failure clients fear). Provider status alone is not enough.

MANDATORY before an enriched email is kept:

• The email's domain must belong to the target company — match by domain-root against the target's known domain / company name.
• If the domain is unfamiliar, verify what it actually is (e.g. scrape its title) before trusting it.
• Drop identity mismatches even when status is DELIVERABLE.
• A company's mail domain can differ from its website. Accept a second domain only when the company's own evidence

ties it to it: the address on its imprint or contact page, or a domain that redirects to the website. A similar name is not evidence. Without this, the right person's address on the company's mail domain is dropped as a mismatch.

This is the real anti-"made-up-address" guard; the sanitize.py deliverability filter (below) is only the last net.

Phone

Skip entirely in demo mode.

LinkedIn URL (when missing)

• Pipe0 `people:profileurl:name@1` — 0.60 cr/person, needs name + company_name + location_hint
• PhantomBuster Profile Scraper (config key: PB_AGENT_PROFILE) — scrape full profile data from existing LinkedIn URL; use /phantombuster skill with phantom name "LinkedIn Profile Scraper" and CSV path. See _shared/phantombuster.md.

Execution Protocol

1. Assess Input Data

• Check what fields are available (LinkedIn URL, domain, name, company)
• Determine which enrichment types are needed
• Estimate credit cost: n_contacts × max_cost_per_record
• Confirm credit balance with user before proceeding

2. Test Batch (5–10 contacts)

• Run chosen provider on 5–10 contacts in production
• Inspect every result: email validity status, phone format, match accuracy
• Check hit rate — if below 10%, investigate before proceeding

3. Review with User

• Present test results with hit rate, status breakdown, sample data
• Confirm provider choice and credit spend for full batch
• Get approval

4. Full Batch

• Split into chunks of max 100 contacts per request (all providers)
• Save request/task IDs immediately
• Poll with appropriate timeouts:
• FE: 15s interval, 600s timeout (usually instant)
• Pipe0: 30s interval, 600s timeout (≤100 contacts)
• BC phone: 30s interval, 900s timeout
• BC email: 60s interval, 3600s timeout
• Save results incrementally after each batch

5. Fallback Pass

• Identify contacts with missing email/phone after primary provider
• Run fallback provider on misses only
• Merge results

6. Final Output

• Merge all provider results into csv/intermediate/contacts_enriched.csv — keep every field (email_source/phone_source, email_status, provider labels), per conventions rule #6
• Confirm each email passed the domain-identity cross-check above before it can ship
• Sanitize before the lead-facing view — run the shared sanitizer (do not reimplement); it drops bad emails, strips provider/source labels + internal status codes, and drops empty columns. See conventions Output Sanitization:

import sys, os
sys.path.insert(0, os.path.expanduser("~/.claude/skills/gtm-pipeline/_shared"))
from sanitize import sanitize_rows
clean, report = sanitize_rows(rows, email_policy="standard")  # "strict" also drops catch-all

• Deliverability policy is applied by sanitize.py via email_policy: default standard keeps DELIVERABLE + HIGH_PROBABILITY + CATCH_ALL and drops UNKNOWN/RISKY/invalid; strict additionally drops catch-all
• Write the sanitized rows to csv/output/contacts_enriched.csv

Provider 1: FullEnrich v2 (Email + Phone)

Endpoint: POST https://app.fullenrich.com/api/v2/contact/enrich/bulk Auth: Authorization: Bearer $FULLENRICH_API_KEY

Submit (max 100 contacts/batch)

{
  "name": "My enrichment job",
  "data": [
    {
      "firstname": "Jane",
      "lastname": "Doe",
      "domain": "example.com",
      "linkedin_url": "https://www.linkedin.com/in/janedoe",
      "enrich_fields": ["contact.emails"],
      "custom": {"row_id": "0"}
    }
  ]
}

For phone: "enrich_fields": ["contact.phones"] or both: ["contact.emails", "contact.phones"]

Poll

GET https://app.fullenrich.com/api/v2/contact/enrich/bulk/{enrichment_id}

Done when: response["status"] == "FINISHED" Speed: usually instant — often finished before first poll. Use 15s interval, 600s timeout.

Parse

for entry in result.get("data", []):
    row_id = str(entry.get("custom", {}).get("row_id", ""))
    ci = entry.get("contact_info", {})
    # Email
    best = ci.get("most_probable_work_email", {})
    email = best.get("email", "") or ""
    if not email:
        for e in ci.get("work_emails", []):
            if e.get("email"):
                email = e["email"]; break
    email_status = best.get("status", "")  # DELIVERABLE / HIGH_PROBABILITY / CATCH_ALL
    # Phone
    phone_list = ci.get("phones", [])
    phone = phone_list[0].get("number", "") if phone_list else ""

Key Notes

• Cloudflare-fronted — call with curl + a browser User-Agent (conventions rule #8); never Python requests/urllib
• Base URL is app.fullenrich.com (NOT api.fullenrich.com)
• Use v2 (not v1)
• Max 100 contacts per batch
• Submit response key is enrichment_id (UUID)
• Email: contact_info.most_probable_work_email.email
• Phone: contact_info.phones[0].number
• If credits run out mid-batch: status = "CREDITS_INSUFFICIENT", partial results returned
• Email status values: DELIVERABLE, HIGH_PROBABILITY, CATCH_ALL

Cost

~0.8 credits/contact

Docs

https://docs.fullenrich.com

Provider 2: Pipe0 Email Waterfall

Pipe ID: people:workemail:profileurl:waterfall@1 Auth: Authorization: Bearer $PIPE0_API_KEY Use curl — Python requests blocked by Cloudflare.

Submit

{
  "config": {"environment": "production"},
  "pipes": [{"pipe_id": "people:workemail:profileurl:waterfall@1"}],
  "input": [
    {"id": "1", "profile_url": "https://linkedin.com/in/janedoe"}
  ]
}

Key Notes

• Submit response key is `id` (not task_id) — use this value for polling: GET /pipes/check/{id}
• email_validation_status field is also returned alongside work_email in each record
• Email quality: ~15–25% of hits may be stale (old employer email) for EU audiences. Waterfall picks up the most recent known email regardless of current role. Always post-filter: remove personal domains (gmail, yahoo, etc.) and cross-check email domain against current company. In EU SME test: 37/43 raw hits, 30/43 valid after filtering 7 stale/personal.

Poll

GET https://api.pipe0.com/v1/pipes/check/{id}

Done when: status == "completed". Use 30s interval, 600s timeout (~3 min for 43 contacts). Keep batches ≤100 contacts.

Parse

records = result.get("records", {})
for rid, rec in records.items():
    fields = rec.get("fields", {})
    email  = (fields.get("work_email", {}) or {}).get("value", "") or ""
    status = (fields.get("email_validation_status", {}) or {}).get("value", "") or ""
    # work_email.value is a plain string, NOT nested

Post-filtering (Recommended)

After enrichment, filter out stale/personal emails:

PERSONAL_DOMAINS = {'gmail.com','yahoo.com','hotmail.com','outlook.com','icloud.com'}
df['email'] = df['email'].apply(
    lambda e: "" if (e and e.split('@')[-1].lower() in PERSONAL_DOMAINS) else e
)
# Also manually flag emails where domain doesn't match current company

Waterfall Providers & Cost

Stops at first hit. Total: 1.00–3.50 cr/contact.

Provider 3: BetterContact (Email or Phone)

Endpoint: POST https://app.bettercontact.rocks/api/v2/async Auth: X-API-KEY: $BETTERCONTACT_API_KEY

Submit (max 100 contacts/batch)

{
  "enrich_email_address": false,
  "enrich_phone_number": true,
  "data": [
    {
      "first_name": "Jane",
      "last_name": "Doe",
      "company_domain": "example.com",
      "linkedin_url": "https://www.linkedin.com/in/janedoe",
      "custom_fields": {"row_id": "0"}
    }
  ]
}

Poll

GET https://app.bettercontact.rocks/api/v2/async/{request_id}

Done when: response["status"] == "terminated" (NOT "completed")

• Phone only: ~3–5 min/100 contacts. Use 900s timeout.
• Email only or both: 30–90 min. Use 3600s timeout.
• on hold = credits exhausted, batch paused — no cancel endpoint.