Agent skill

hubspot-crm

Use when syncing contacts or lists to HubSpot CRM. Automatically uses HUBSPOT_API_TOKEN from environment.

View SKILL.md on GitHub Repository
Stars 1
Forks 0

Install this agent skill to your Project

npx add-skill https://github.com/FunnelEnvy/agents_webinar_demos/tree/main/.claude/skills/hubspot-crm

SKILL.md

HubSpot CRM Integration

Sync contacts and lists to HubSpot using the REST API.

Environment Variables

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

Quick Start

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

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

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

Create Contact

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.

Add Contacts to List

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

Full Upload Flow

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

Usage Example

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

Using the Integration Script

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.

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

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

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
  6. Click Create app and copy the token
  7. Set as HUBSPOT_API_TOKEN environment variable

Recommended Agent Skills

Expand your agent's capabilities with these related and highly-rated skills.

FunnelEnvy/agents_webinar_demos

playwright-browser

Use when capturing screenshots, automating browser interactions, or scraping web content. Covers Playwright Python API for page navigation, screenshots, element selection, form filling, and waiting strategies.

1 0
Explore
FunnelEnvy/agents_webinar_demos

bq-query-optimization

Use when writing BigQuery queries, optimizing query performance, analyzing execution plans, or avoiding common SQL gotchas. Covers parameterized queries, UDFs, scripting, window functions (QUALIFY, ROW_NUMBER, RANK, LEAD/LAG), JSON functions, ARRAY/STRUCT operations, BigQuery-specific features (EXCEPT, REPLACE, SAFE_*), CTE re-execution issues, NOT IN with NULLs, DML performance, Standard vs Legacy SQL, and performance best practices.

1 0
Explore
FunnelEnvy/agents_webinar_demos

n8n-workflow

Use when creating, deploying, or managing N8N workflows. Automatically uses N8N_HOST and N8N_API_KEY from environment.

1 0
Explore
FunnelEnvy/agents_webinar_demos

gcp-bq-data-loading

Use when loading data into BigQuery from CSV, JSON, Avro, Parquet files, Cloud Storage, or local files. Covers bq load command, source formats, schema detection, incremental loading, and handling parsing errors.

1 0
Explore
FunnelEnvy/agents_webinar_demos

gcp-cli-gotchas

Use when encountering gcloud or bq CLI formatting errors, quote escaping issues, command substitution problems, or when debugging CLI commands. Provides solutions for backtick usage, heredoc syntax, timestamp filters, parameter escaping, and multiline command formatting.

1 0
Explore
FunnelEnvy/agents_webinar_demos

gcp-bq-table-management

Use when creating BigQuery tables, implementing partitioning or clustering, managing table schemas, or optimizing table structure. Covers time-based partitioning, range partitioning, clustering strategies, DDL commands, and table configuration.

1 0
Explore

Didn't find tool you were looking for?

Be as detailed as possible for better results