hubspot-crm

```bash

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

```

```python

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'))

```

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

```

```python

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

```

```python

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.

```python

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")

```

```python

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

```

```python

# 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"

)

```

For CSV files, use the provided script:

```bash

.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.

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

```python

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}")

```

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