hubspot-crm（HubSpot CRM 搜索）

HubSpot CRM Integration

Sync contacts and lists to HubSpot using the REST API.

Environment Variables

HUBSPOT_API_TOKEN="pat-na1-..."  # Private App token from HubSpot

Quick Start

import os
import json
from urllib.request import Request, urlopen
from urllib.error import HTTPError

class HubSpotClient:
    """Simple HubSpot API client."""

    def __init__(self):
        self.token = os.environ.get('HUBSPOT_API_TOKEN')
        if not self.token:
            raise ValueError("HUBSPOT_API_TOKEN environment variable not set")
        self.base_url = "https://api.hubapi.com"

    def _request(self, method: str, endpoint: str, data: dict = None) -> dict:
        url = f"{self.base_url}{endpoint}"
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        }
        body = json.dumps(data).encode('utf-8') if data else None
        request = Request(url, data=body, headers=headers, method=method)

        with urlopen(request, timeout=30) as response:
            return json.loads(response.read().decode('utf-8'))

Create Static List

def create_static_list(client, name: str) -> str:
    """Create a static list for contacts. Returns list ID."""
    payload = {
        "name": name,
        "objectTypeId": "0-1",  # REQUIRED: 0-1 = contacts
        "processingType": "MANUAL"
    }

    result = client._request("POST", "/crm/v3/lists", payload)
    # Response is nested: {"list": {"listId": "..."}}
    list_data = result.get("list", result)
    list_id = list_data.get("listId")

    print(f"✅ Created list: {name} (ID: {list_id})")
    return list_id

Search Contact by Email

def search_contact(client, email: str) -> str | None:
    """Find contact by email. Returns contact ID or None."""
    payload = {
        "filterGroups": [{
            "filters": [{
                "propertyName": "email",
                "operator": "EQ",
                "value": email
            }]
        }],
        "properties": ["email"],
        "limit": 1
    }

    result = client._request("POST", "/crm/v3/objects/contacts/search", payload)
    results = result.get("results", [])
    return results[0]["id"] if results else None

Create Contact

def create_contact(client, email: str, firstname: str = None, lastname: str = None) -> str:
    """Create a new contact with email only (vanilla upload).

    Only uses standard HubSpot properties (email, firstname, lastname)
    to avoid errors from missing custom properties in the target account.
    """
    properties = {"email": email}
    if firstname:
        properties["firstname"] = firstname
    if lastname:
        properties["lastname"] = lastname

    result = client._request("POST", "/crm/v3/objects/contacts", {"properties": properties})
    return result["id"]

Important: Do NOT pass arbitrary CSV columns as properties. HubSpot will reject any property names that don't exist in the target account. Only use standard fields (email, firstname, lastname) unless you've confirmed custom properties exist.

Add Contacts to List

def add_to_list(client, list_id: str, contact_ids: list[str]):
    """Add contacts to a static list. Batches in groups of 100."""
    endpoint = f"/crm/v3/lists/{list_id}/memberships/add"

    batch_size = 100
    total_added = 0

    for i in range(0, len(contact_ids), batch_size):
        batch = contact_ids[i:i + batch_size]
        # IMPORTANT: Payload is a simple array, NOT {"recordIdsToAdd": [...]}
        result = client._request("PUT", endpoint, batch)
        added = len(result.get("recordsIdsAdded", []))
        total_added += added
        print(f"  Added batch: {added} contacts")

    print(f"✅ Added {total_added} total contacts to list")

Full Upload Flow

def upload_users_to_hubspot(emails: list[str], list_name: str) -> str:
    """Upload a list of email addresses to HubSpot."""
    client = HubSpotClient()

    # Create list
    list_id = create_static_list(client, list_name)

    # Find or create contacts
    contact_ids = []
    for email in emails:
        contact_id = search_contact(client, email)
        if not contact_id:
            contact_id = create_contact(client, email)
        contact_ids.append(contact_id)

    # Add to list
    add_to_list(client, list_id, contact_ids)

    print(f"\n✅ Complete!")
    print(f"   List: https://app.hubspot.com/contacts/lists/{list_id}")

    return list_id

Usage Example

# Upload at-risk users from analysis
at_risk_emails = [
    "user_001@demo.reformapp.com",
    "user_002@demo.reformapp.com",
    "user_003@demo.reformapp.com"
]

list_id = upload_users_to_hubspot(
    emails=at_risk_emails,
    list_name="At-Risk Trial Users - Dec 2024"
)

Using the Integration Script

For CSV files, use the provided script:

.venv/bin/python demos/04-trial-to-paid/scripts/hubspot_integration.py \
  path/to/users.csv \
  "List Name Here"

The CSV must have an email column.

API Gotchas

1. List creation requires objectTypeId: Always include "objectTypeId": "0-1" for contacts
2. Response is nested: List ID is at result["list"]["listId"], not result["listId"]
3. Add-to-list payload format: Use simple array ["id1", "id2"], NOT {"recordIdsToAdd": [...]}
4. Contact IDs are strings: Even though they look numeric, treat them as strings
5. Batch limit: Add contacts in batches of 100 max

Error Handling

from urllib.error import HTTPError

try:
    result = client._request("POST", endpoint, payload)
except HTTPError as e:
    error_body = e.read().decode('utf-8')
    print(f"HubSpot API error: {e.code}")
    print(f"  {error_body}")

Getting Your HubSpot Private App Token

1. Go to Settings (gear icon) in HubSpot
2. Navigate to Integrations > Private Apps
3. Click Create a private app
4. Give it a name (e.g., "AI Agent Integration")
5. Under Scopes, enable:

- crm.lists.read - crm.lists.write - crm.objects.contacts.read - crm.objects.contacts.write

1. Click Create app and copy the token
2. Set as HUBSPOT_API_TOKEN environment variable