HubSpot API

Provides patterns for interacting with the HubSpot CRM API.

Authentication

import requests

def get_headers(access_token: str) -> dict:
    """Get headers for API requests."""
    return {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }

BASE_URL = "https://api.hubapi.com"

Fetching Contacts

def get_contacts(access_token: str, limit: int = 100, after: str = None) -> dict:
    """
    Get contacts with pagination.

    Usage:
        result = get_contacts(access_token, limit=50)
        contacts = result["results"]
        next_cursor = result.get("paging", {}).get("next", {}).get("after")
    """
    url = f"{BASE_URL}/crm/v3/objects/contacts"
    params = {
        "limit": limit,
        "properties": "email,firstname,lastname,phone,company"
    }
    if after:
        params["after"] = after

    response = requests.get(url, headers=get_headers(access_token), params=params)
    response.raise_for_status()
    return response.json()

def get_all_contacts(access_token: str) -> list[dict]:
    """Get all contacts with automatic pagination."""
    contacts = []
    after = None

    while True:
        result = get_contacts(access_token, limit=100, after=after)
        contacts.extend(result.get("results", []))

        paging = result.get("paging", {})
        after = paging.get("next", {}).get("after")

        if not after:
            break

    return contacts

def get_contact_by_id(access_token: str, contact_id: str) -> dict:
    """Get a single contact by ID."""
    url = f"{BASE_URL}/crm/v3/objects/contacts/{contact_id}"
    params = {"properties": "email,firstname,lastname,phone,company"}

    response = requests.get(url, headers=get_headers(access_token), params=params)
    response.raise_for_status()
    return response.json()

def get_contact_by_email(access_token: str, email: str) -> dict:
    """Get contact by email address."""
    url = f"{BASE_URL}/crm/v3/objects/contacts/{email}"
    params = {
        "idProperty": "email",
        "properties": "email,firstname,lastname,phone,company"
    }

    response = requests.get(url, headers=get_headers(access_token), params=params)
    response.raise_for_status()
    return response.json()

Creating and Updating Contacts

def create_contact(access_token: str, properties: dict) -> dict:
    """
    Create a new contact.

    Usage:
        contact = create_contact(access_token, {
            "email": "john@example.com",
            "firstname": "John",
            "lastname": "Doe",
            "phone": "555-1234"
        })
    """
    url = f"{BASE_URL}/crm/v3/objects/contacts"

    response = requests.post(
        url,
        headers=get_headers(access_token),
        json={"properties": properties}
    )
    response.raise_for_status()
    return response.json()

def update_contact(access_token: str, contact_id: str, properties: dict) -> dict:
    """
    Update an existing contact.

    Usage:
        updated = update_contact(access_token, "12345", {
            "phone": "555-5678"
        })
    """
    url = f"{BASE_URL}/crm/v3/objects/contacts/{contact_id}"

    response = requests.patch(
        url,
        headers=get_headers(access_token),
        json={"properties": properties}
    )
    response.raise_for_status()
    return response.json()

def delete_contact(access_token: str, contact_id: str) -> bool:
    """Delete a contact."""
    url = f"{BASE_URL}/crm/v3/objects/contacts/{contact_id}"

    response = requests.delete(url, headers=get_headers(access_token))
    return response.status_code == 204

Batch Operations

def batch_create_contacts(access_token: str, contacts: list[dict]) -> dict:
    """
    Create multiple contacts in batch.

    Usage:
        results = batch_create_contacts(access_token, [
            {"email": "john@example.com", "firstname": "John"},
            {"email": "jane@example.com", "firstname": "Jane"}
        ])
    """
    url = f"{BASE_URL}/crm/v3/objects/contacts/batch/create"

    payload = {
        "inputs": [{"properties": contact} for contact in contacts]
    }

    response = requests.post(url, headers=get_headers(access_token), json=payload)
    response.raise_for_status()
    return response.json()

def batch_update_contacts(access_token: str, updates: list[dict]) -> dict:
    """
    Update multiple contacts in batch.

    Usage:
        results = batch_update_contacts(access_token, [
            {"id": "123", "properties": {"phone": "555-1234"}},
            {"id": "456", "properties": {"phone": "555-5678"}}
        ])
    """
    url = f"{BASE_URL}/crm/v3/objects/contacts/batch/update"

    payload = {"inputs": updates}

    response = requests.post(url, headers=get_headers(access_token), json=payload)
    response.raise_for_status()
    return response.json()

Searching Contacts

def search_contacts(access_token: str, filters: list[dict], properties: list[str] = None) -> list[dict]:
    """
    Search contacts with filters.

    Usage:
        # Find contacts at a company
        results = search_contacts(access_token, [
            {"propertyName": "company", "operator": "EQ", "value": "Acme Corp"}
        ])

        # Find contacts with email domain
        results = search_contacts(access_token, [
            {"propertyName": "email", "operator": "CONTAINS_TOKEN", "value": "@acme.com"}
        ])
    """
    url = f"{BASE_URL}/crm/v3/objects/contacts/search"

    payload = {
        "filterGroups": [{"filters": filters}],
        "properties": properties or ["email", "firstname", "lastname", "phone", "company"],
        "limit": 100
    }

    response = requests.post(url, headers=get_headers(access_token), json=payload)
    response.raise_for_status()
    return response.json().get("results", [])

Working with Companies

def get_companies(access_token: str, limit: int = 100) -> dict:
    """Get companies."""
    url = f"{BASE_URL}/crm/v3/objects/companies"
    params = {
        "limit": limit,
        "properties": "name,domain,industry,numberofemployees"
    }

    response = requests.get(url, headers=get_headers(access_token), params=params)
    response.raise_for_status()
    return response.json()

def create_company(access_token: str, properties: dict) -> dict:
    """Create a company."""
    url = f"{BASE_URL}/crm/v3/objects/companies"

    response = requests.post(
        url,
        headers=get_headers(access_token),
        json={"properties": properties}
    )
    response.raise_for_status()
    return response.json()

Associations

def associate_contact_to_company(access_token: str, contact_id: str, company_id: str) -> bool:
    """Associate a contact with a company."""
    url = f"{BASE_URL}/crm/v3/objects/contacts/{contact_id}/associations/companies/{company_id}/contact_to_company"

    response = requests.put(url, headers=get_headers(access_token))
    return response.status_code == 200

Helper Script

Use helper.py for the HubSpotClient class with comprehensive API features.