Import recently funded companies from Crunchbase into HubSpot using an agent skill

mkdir -p .claude/skills/funded-companies/scripts

---
name: funded-companies
description: Import recently funded companies from Crunchbase into HubSpot. Searches for Series A+ funding rounds in the last 30 days, filters by ICP, and creates company records with funding details.
disable-model-invocation: true
allowed-tools: Bash(python *)
---
 
Import recently funded companies into HubSpot from Crunchbase.
 
Usage: /funded-companies [days_back]
 
1. Run: `python $SKILL_DIR/scripts/import_funded.py --days ${1:-30}`
2. Review the output showing which companies were imported
3. Confirm companies appear in HubSpot with funding data

#!/usr/bin/env python3
"""
Funded Companies Import: Crunchbase → HubSpot
Finds recently funded companies and imports them as target accounts.
"""
import argparse
import os
import sys
import time
from datetime import datetime, timedelta
from urllib.parse import urlparse
 
try:
    import requests
except ImportError:
    os.system("pip install requests -q")
    import requests
 
# --- Config ---
CRUNCHBASE_API_KEY = os.environ.get("CRUNCHBASE_API_KEY")
HUBSPOT_TOKEN = os.environ.get("HUBSPOT_TOKEN")
 
if not all([CRUNCHBASE_API_KEY, HUBSPOT_TOKEN]):
    print("ERROR: Set CRUNCHBASE_API_KEY and HUBSPOT_TOKEN environment variables")
    sys.exit(1)
 
CB_HEADERS = {"X-cb-user-key": CRUNCHBASE_API_KEY, "Content-Type": "application/json"}
HS_HEADERS = {"Authorization": f"Bearer {HUBSPOT_TOKEN}", "Content-Type": "application/json"}
 
# ICP config — adjust these for your business
FUNDING_STAGES = ["series_a", "series_b", "series_c", "series_d"]
TARGET_EMPLOYEE_RANGES = [
    "c_0051_0100", "c_0101_0250", "c_0251_0500", "c_0501_1000",
]
 
# --- Functions ---
def search_funding_rounds(days_back):
    """Search Crunchbase for recent funding rounds."""
    since_date = (datetime.now() - timedelta(days=days_back)).strftime("%Y-%m-%d")
    all_entities = []
    after_id = None
 
    while True:
        body = {
            "field_ids": [
                "identifier", "announced_on", "money_raised",
                "funded_organization_identifier", "investment_type",
                "investor_identifiers",
            ],
            "query": [
                {"type": "predicate", "field_id": "announced_on",
                 "operator_id": "gte", "values": [since_date]},
                {"type": "predicate", "field_id": "investment_type",
                 "operator_id": "includes", "values": FUNDING_STAGES},
            ],
            "limit": 100,
        }
        if after_id:
            body["after_id"] = after_id
 
        resp = requests.post(
            "https://api.crunchbase.com/api/v4/searches/funding_rounds",
            headers=CB_HEADERS, json=body,
        )
        resp.raise_for_status()
        entities = resp.json().get("entities", [])
        all_entities.extend(entities)
 
        if len(entities) < 100:
            break
        after_id = entities[-1]["uuid"]
 
    return all_entities
 
 
def get_organization(uuid):
    """Fetch organization details from Crunchbase."""
    resp = requests.get(
        f"https://api.crunchbase.com/api/v4/entities/organizations/{uuid}",
        headers=CB_HEADERS,
        params={"field_ids": "short_description,categories,num_employees_enum,website_url,founded_on"},
    )
    resp.raise_for_status()
    return resp.json().get("properties", {})
 
 
def extract_domain(url):
    """Extract clean domain from URL."""
    if not url:
        return None
    parsed = urlparse(url if url.startswith("http") else f"https://{url}")
    domain = parsed.netloc or parsed.path
    return domain.replace("www.", "").strip("/")
 
 
def company_exists(domain):
    """Check if company exists in HubSpot."""
    resp = requests.post(
        "https://api.hubapi.com/crm/v3/objects/companies/search",
        headers=HS_HEADERS,
        json={"filterGroups": [{"filters": [{
            "propertyName": "domain", "operator": "EQ", "value": domain,
        }]}]},
    )
    resp.raise_for_status()
    return len(resp.json().get("results", [])) > 0
 
 
def create_company(org, funding_round):
    """Create company in HubSpot with funding metadata."""
    domain = extract_domain(org.get("website_url"))
    props = funding_round.get("properties", {})
    investors = ", ".join(
        inv.get("value", "") for inv in (props.get("investor_identifiers") or [])
    )
    money = props.get("money_raised", {})
    amount = money.get("value") if money else None
 
    resp = requests.post(
        "https://api.hubapi.com/crm/v3/objects/companies",
        headers=HS_HEADERS,
        json={"properties": {
            "domain": domain,
            "name": org.get("identifier", {}).get("value", ""),
            "description": org.get("short_description", ""),
            "industry": (org.get("categories") or [{}])[0].get("value", ""),
            "funding_stage": props.get("investment_type", ""),
            "funding_amount": str(amount) if amount else "",
            "funding_date": props.get("announced_on", ""),
            "investors": investors,
        }},
    )
    resp.raise_for_status()
    return resp.json()["id"]
 
 
# --- Main ---
parser = argparse.ArgumentParser()
parser.add_argument("--days", type=int, default=30, help="Look back N days for funding rounds")
args = parser.parse_args()
 
print(f"Searching Crunchbase for funding rounds in the last {args.days} days...")
rounds = search_funding_rounds(args.days)
print(f"Found {len(rounds)} funding rounds ({', '.join(FUNDING_STAGES)})\n")
 
seen_domains = set()
created = 0
skipped = 0
icp_miss = 0
 
for rnd in rounds:
    org_ref = rnd.get("properties", {}).get("funded_organization_identifier", {})
    org_uuid = org_ref.get("uuid")
    if not org_uuid:
        continue
 
    org = get_organization(org_uuid)
    time.sleep(0.2)
 
    # ICP filter
    emp_range = org.get("num_employees_enum", "")
    if emp_range not in TARGET_EMPLOYEE_RANGES or not org.get("website_url"):
        icp_miss += 1
        continue
 
    domain = extract_domain(org.get("website_url"))
    if not domain or domain in seen_domains:
        continue
    seen_domains.add(domain)
 
    if company_exists(domain):
        name = org.get("identifier", {}).get("value", domain)
        print(f"  EXISTS: {name} ({domain})")
        skipped += 1
        continue
 
    money = rnd.get("properties", {}).get("money_raised", {})
    amount_str = f"${money.get('value', 0):,.0f}" if money.get("value") else "undisclosed"
    stage = rnd.get("properties", {}).get("investment_type", "unknown")
    name = org.get("identifier", {}).get("value", "Unknown")
 
    company_id = create_company(org, rnd)
    print(f"  CREATED: {name} — {stage} — {amount_str}")
    created += 1
    time.sleep(0.2)
 
print(f"\n--- Summary ---")
print(f"Funding rounds found: {len(rounds)}")
print(f"Companies created: {created}")
print(f"Already in HubSpot: {skipped}")
print(f"Didn't match ICP: {icp_miss}")

# Default: last 30 days of funding rounds
/funded-companies
 
# Look back 7 days (for weekly runs)
/funded-companies 7
 
# Look back 90 days (for initial backfill)
/funded-companies 90
 
# Or run the script directly
python .claude/skills/funded-companies/scripts/import_funded.py --days 30

# Target only Series A and B (earlier-stage, actively buying)
FUNDING_STAGES = ["series_a", "series_b"]
 
# Target companies with 100-1000 employees
TARGET_EMPLOYEE_RANGES = ["c_0101_0250", "c_0251_0500", "c_0501_1000"]

# crontab — run weekly on Monday at 8 AM
0 8 * * 1 cd /path/to/project && python .claude/skills/funded-companies/scripts/import_funded.py --days 7

name: Import Funded Companies
on:
  schedule:
    - cron: '0 13 * * 1'  # Monday 8 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 .claude/skills/funded-companies/scripts/import_funded.py --days 7
        env:
          CRUNCHBASE_API_KEY: ${{ secrets.CRUNCHBASE_API_KEY }}
          HUBSPOT_TOKEN: ${{ secrets.HUBSPOT_TOKEN }}