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

```python
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

List creation requires objectTypeId: Always include "objectTypeId": "0-1" for contacts
Response is nested: List ID is at result["list"]["listId"], not result["listId"]
Add-to-list payload format: Use simple array ["id1", "id2"], NOT {"recordIdsToAdd": [...]}
Contact IDs are strings: Even though they look numeric, treat them as strings
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

Go to Settings (gear icon) in HubSpot
Navigate to Integrations > Private Apps
Click Create a private app
Give it a name (e.g., "AI Agent Integration")
Under Scopes, enable:
- crm.lists.read
- crm.lists.write
- crm.objects.contacts.read
- crm.objects.contacts.write
Click Create app and copy the token
Set as HUBSPOT_API_TOKEN environment variable

hubspot-crm

HubSpot CRM Integration

Environment Variables

Quick Start