Find decision makers at a HubSpot company using Apollo and code

Prerequisites

Python 3.9+ or Node.js 18+
HubSpot private app token with crm.objects.contacts.read, crm.objects.contacts.write, crm.objects.companies.read scopes
Apollo API key with People Search and Enrichment credits
A scheduling environment (cron, GitHub Actions) if running on a recurring basis

Step 1: Set up the project

# Verify your API keys work
curl -s "https://api.hubapi.com/crm/v3/objects/companies?limit=1" \
  -H "Authorization: Bearer $HUBSPOT_TOKEN" | head -c 200
 
curl -s -X POST "https://api.apollo.io/api/v1/mixed_people/search" \
  -H "Content-Type: application/json" \
  -H "X-Api-Key: $APOLLO_API_KEY" \
  -d '{"per_page": 1}' | head -c 200

Step 2: Fetch target companies from HubSpot

Pull companies tagged as target accounts. This example filters for a custom abm_tier property, but adjust to match your setup.

import requests
import os
 
HUBSPOT_TOKEN = os.environ["HUBSPOT_TOKEN"]
HUBSPOT_HEADERS = {
    "Authorization": f"Bearer {HUBSPOT_TOKEN}",
    "Content-Type": "application/json",
}
 
def get_target_companies():
    """Fetch companies marked as Tier 1 ABM targets."""
    resp = requests.post(
        "https://api.hubapi.com/crm/v3/objects/companies/search",
        headers=HUBSPOT_HEADERS,
        json={
            "filterGroups": [{"filters": [{
                "propertyName": "abm_tier",
                "operator": "EQ",
                "value": "Tier 1"
            }]}],
            "properties": ["domain", "name"],
            "limit": 100,
        },
    )
    resp.raise_for_status()
    return resp.json()["results"]
 
companies = get_target_companies()
print(f"Found {len(companies)} target companies")

Step 3: Search Apollo for decision makers

For each target company, search Apollo's People Search API by domain, titles, and seniority.

import time
 
APOLLO_API_KEY = os.environ["APOLLO_API_KEY"]
APOLLO_HEADERS = {
    "Content-Type": "application/json",
    "X-Api-Key": APOLLO_API_KEY,
}
 
TARGET_TITLES = [
    "VP Sales", "CRO", "VP Marketing", "CMO",
    "VP RevOps", "Head of Sales", "VP Business Development",
]
TARGET_SENIORITIES = ["vp", "c_suite", "director"]
 
def search_decision_makers(domain):
    """Search Apollo for decision makers at a company domain."""
    resp = requests.post(
        "https://api.apollo.io/api/v1/mixed_people/search",
        headers=APOLLO_HEADERS,
        json={
            "q_organization_domains_list": [domain],
            "person_titles": TARGET_TITLES,
            "person_seniorities": TARGET_SENIORITIES,
            "page": 1,
            "per_page": 25,
        },
    )
    resp.raise_for_status()
    return resp.json().get("people", [])

Apollo rate limits

Apollo enforces rate limits of ~5 requests/second on most plans. Add a 200ms delay between requests when processing multiple companies. The API returns a 429 status when you hit the limit.

Step 4: Enrich contacts for verified emails

The People Search results may not include verified emails. Use Apollo's People Match endpoint to get them.

def enrich_person(person):
    """Get verified email via Apollo People Match."""
    resp = requests.post(
        "https://api.apollo.io/api/v1/people/match",
        headers=APOLLO_HEADERS,
        json={
            "first_name": person.get("first_name"),
            "last_name": person.get("last_name"),
            "organization_name": person.get("organization", {}).get("name"),
            "reveal_personal_emails": False,
        },
    )
    resp.raise_for_status()
    data = resp.json().get("person", {})
    return {
        "email": data.get("email"),
        "first_name": data.get("first_name"),
        "last_name": data.get("last_name"),
        "title": data.get("title"),
        "linkedin_url": data.get("linkedin_url"),
        "phone": (data.get("phone_numbers") or [{}])[0].get("sanitized_number"),
        "company": data.get("organization", {}).get("name"),
    }

Step 5: Deduplicate against existing HubSpot contacts

Before creating contacts, check if they already exist in HubSpot to avoid duplicates.

def contact_exists(email):
    """Check if a contact with this email already exists in HubSpot."""
    resp = requests.post(
        "https://api.hubapi.com/crm/v3/objects/contacts/search",
        headers=HUBSPOT_HEADERS,
        json={
            "filterGroups": [{"filters": [{
                "propertyName": "email",
                "operator": "EQ",
                "value": email,
            }]}],
        },
    )
    resp.raise_for_status()
    results = resp.json().get("results", [])
    return results[0]["id"] if results else None

Step 6: Create contacts and associate with the company

def create_contact_and_associate(person, company_id):
    """Create a HubSpot contact and associate it with the company."""
    if not person.get("email"):
        print(f"  Skipping {person['first_name']} {person['last_name']} — no email")
        return None
 
    existing_id = contact_exists(person["email"])
    if existing_id:
        print(f"  {person['email']} already exists (ID: {existing_id}), associating")
        contact_id = existing_id
    else:
        resp = requests.post(
            "https://api.hubapi.com/crm/v3/objects/contacts",
            headers=HUBSPOT_HEADERS,
            json={
                "properties": {
                    "email": person["email"],
                    "firstname": person["first_name"],
                    "lastname": person["last_name"],
                    "jobtitle": person.get("title", ""),
                    "company": person.get("company", ""),
                    "phone": person.get("phone", ""),
                    "hs_lead_status": "NEW",
                }
            },
        )
        resp.raise_for_status()
        contact_id = resp.json()["id"]
        print(f"  Created contact: {person['email']} (ID: {contact_id})")
 
    # Associate contact with company
    requests.put(
        f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}"
        f"/associations/companies/{company_id}/contact_to_company",
        headers=HUBSPOT_HEADERS,
    )
    return contact_id
 
 
# --- Main execution ---
for company in companies:
    domain = company["properties"].get("domain")
    if not domain:
        print(f"Skipping {company['properties']['name']} — no domain")
        continue
 
    print(f"\nProcessing: {company['properties']['name']} ({domain})")
    people = search_decision_makers(domain)
    print(f"  Found {len(people)} decision makers in Apollo")
 
    for person in people:
        enriched = enrich_person(person)
        create_contact_and_associate(enriched, company["id"])
        time.sleep(0.2)  # Respect rate limits

Apollo credit math

Each company costs: (number of search results x 1 credit) + (number of enrichments x 1 credit). A company with 8 decision makers = 16 credits. Processing 50 companies averaging 6 people each = ~600 credits/month.

Step 7: Schedule with cron or GitHub Actions

# .github/workflows/find-decision-makers.yml
name: Find Decision Makers
on:
  schedule:
    - cron: '0 14 * * 1'  # Weekly on Monday at 9 AM ET
  workflow_dispatch: {}
jobs:
  run:
    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 find_decision_makers.py
        env:
          HUBSPOT_TOKEN: ${{ secrets.HUBSPOT_TOKEN }}
          APOLLO_API_KEY: ${{ secrets.APOLLO_API_KEY }}

Cost

Hosting: Free on GitHub Actions (2,000 min/month free tier)
Apollo: ~2 credits per decision maker. Free plan includes 10,000 credits/year. Paid plans start at $49/mo.
HubSpot API: Free with any HubSpot plan that supports private apps