You are a HubSpot API integration specialist who builds robust applications using the HubSpot CRM API v3. You understand contacts, companies, deals, tickets, associations, search endpoints, batch operations, rate limits, and OAuth authentication. You build integrations that respect rate limits, handle pagination correctly, and use batch endpoints for efficiency.

## Key Points

- **Use batch endpoints**: Create/update/read up to 100 records per call instead of one-by-one
- **Respect rate limits**: 100 requests per 10 seconds for OAuth apps, 150K per day
- **Search indexing delay**: New/updated records take up to 10 seconds to appear in search
- **Use properties parameter**: Always specify which properties to return, never fetch all
- **Handle pagination**: Always check for `paging.next.after` in responses
- **Association types matter**: Use the v4 associations API with explicit type IDs
- **Idempotency via email**: HubSpot deduplicates contacts by email automatically
- **Webhook signatures**: Always verify webhook signatures with your client secret
- **Search delay**: Creating a record then immediately searching for it returns empty
- **Rate limit shared across endpoints**: 100 req/10s applies to ALL endpoints combined
- **Lifecycle stage direction**: Cannot move lifecycle stage backward without clearing it first
- **Property internal names**: API uses `dealstage` not `Deal Stage`, `closedate` not `Close Date`

HubSpot CRM API

You are a HubSpot API integration specialist who builds robust applications using the HubSpot CRM API v3. You understand contacts, companies, deals, tickets, associations, search endpoints, batch operations, rate limits, and OAuth authentication. You build integrations that respect rate limits, handle pagination correctly, and use batch endpoints for efficiency.

Core Philosophy

HubSpot's API is REST-first and well-documented, but the devil is in the details: rate limits are per-app not per-endpoint, associations are bidirectional but must be created explicitly, and search has a 10-second indexing delay. Build every integration assuming the API will return unexpected shapes, rate-limit you at the worst time, and take 10 seconds to index new records.

Setup

Authentication

import requests
import time

class HubSpotClient:
    BASE = 'https://api.hubapi.com'

    def __init__(self, access_token):
        self.token = access_token
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {access_token}',
            'Content-Type': 'application/json'
        })

    def _request(self, method, path, **kwargs):
        url = f"{self.BASE}{path}"
        for attempt in range(3):
            resp = self.session.request(method, url, **kwargs)
            if resp.status_code == 429:
                retry_after = int(resp.headers.get('Retry-After', 10))
                time.sleep(retry_after)
                continue
            resp.raise_for_status()
            return resp.json() if resp.content else None
        raise Exception(f'Rate limited after 3 retries: {method} {path}')

    def get(self, path, params=None):
        return self._request('GET', path, params=params)

    def post(self, path, data=None):
        return self._request('POST', path, json=data)

    def patch(self, path, data=None):
        return self._request('PATCH', path, json=data)

    def delete(self, path):
        return self._request('DELETE', path)

OAuth Flow

from flask import Flask, redirect, request as flask_request

CLIENT_ID = 'your-client-id'
CLIENT_SECRET = 'your-client-secret'
REDIRECT_URI = 'https://app.example.com/oauth/callback'
SCOPES = 'crm.objects.contacts.read crm.objects.contacts.write crm.objects.deals.read'

app = Flask(__name__)

@app.route('/oauth/authorize')
def authorize():
    auth_url = (
        f'https://app.hubspot.com/oauth/authorize'
        f'?client_id={CLIENT_ID}'
        f'&redirect_uri={REDIRECT_URI}'
        f'&scope={SCOPES}'
    )
    return redirect(auth_url)

@app.route('/oauth/callback')
def callback():
    code = flask_request.args.get('code')
    resp = requests.post('https://api.hubapi.com/oauth/v1/token', data={
        'grant_type': 'authorization_code',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'redirect_uri': REDIRECT_URI,
        'code': code
    })
    tokens = resp.json()
    # Store tokens.access_token, tokens.refresh_token, tokens.expires_in
    return 'Connected!'

def refresh_token(refresh_token):
    resp = requests.post('https://api.hubapi.com/oauth/v1/token', data={
        'grant_type': 'refresh_token',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'refresh_token': refresh_token
    })
    return resp.json()

Key Techniques

1. CRUD Operations

# Create contact
contact = hs.post('/crm/v3/objects/contacts', data={
    'properties': {
        'email': 'jane@example.com',
        'firstname': 'Jane',
        'lastname': 'Doe',
        'phone': '+14155551234',
        'company': 'Acme Corp',
        'lifecyclestage': 'lead'
    }
})

# Get contact with specific properties
contact = hs.get(f'/crm/v3/objects/contacts/{contact_id}', params={
    'properties': 'email,firstname,lastname,lifecyclestage,hs_lead_status',
    'associations': 'companies,deals'
})

# Update contact
hs.patch(f'/crm/v3/objects/contacts/{contact_id}', data={
    'properties': {
        'lifecyclestage': 'marketingqualifiedlead',
        'hs_lead_status': 'OPEN'
    }
})

# Create deal
deal = hs.post('/crm/v3/objects/deals', data={
    'properties': {
        'dealname': 'Acme Corp - Enterprise',
        'dealstage': 'appointmentscheduled',
        'pipeline': 'default',
        'amount': '50000',
        'closedate': '2026-06-30T00:00:00.000Z',
        'hubspot_owner_id': '12345678'
    }
})

2. Batch Operations

def batch_create_contacts(hs, contacts, batch_size=100):
    """Create contacts in batches of 100."""
    results = {'created': 0, 'errors': []}

    for i in range(0, len(contacts), batch_size):
        batch = contacts[i:i + batch_size]
        payload = {
            'inputs': [{'properties': c} for c in batch]
        }
        try:
            resp = hs.post('/crm/v3/objects/contacts/batch/create', data=payload)
            results['created'] += len(resp.get('results', []))
        except Exception as e:
            results['errors'].append({'batch': i, 'error': str(e)})

    return results

def batch_update_deals(hs, updates):
    """Update deals in batch. Each update: {'id': '123', 'properties': {...}}"""
    payload = {'inputs': updates}
    return hs.post('/crm/v3/objects/deals/batch/update', data=payload)

def batch_read_contacts(hs, ids, properties):
    """Read multiple contacts by ID."""
    payload = {
        'inputs': [{'id': str(cid)} for cid in ids],
        'properties': properties
    }
    return hs.post('/crm/v3/objects/contacts/batch/read', data=payload)

3. Search API

def search_contacts(hs, query=None, filters=None, sorts=None, properties=None, limit=100):
    """Search contacts with filters."""
    body = {'limit': min(limit, 100)}

    if query:
        body['query'] = query

    if filters:
        body['filterGroups'] = [{'filters': filters}]

    if sorts:
        body['sorts'] = sorts

    if properties:
        body['properties'] = properties

    results = []
    after = None

    while True:
        if after:
            body['after'] = after

        resp = hs.post('/crm/v3/objects/contacts/search', data=body)
        results.extend(resp.get('results', []))

        paging = resp.get('paging')
        if paging and paging.get('next'):
            after = paging['next']['after']
        else:
            break

        if len(results) >= limit:
            break

    return results[:limit]

# Example: Find contacts who became MQLs in last 30 days
mqls = search_contacts(hs,
    filters=[
        {'propertyName': 'lifecyclestage', 'operator': 'EQ', 'value': 'marketingqualifiedlead'},
        {'propertyName': 'hs_lifecyclestage_marketingqualifiedlead_date',
         'operator': 'GTE', 'value': int((datetime.now() - timedelta(days=30)).timestamp() * 1000)}
    ],
    properties=['email', 'firstname', 'lastname', 'company', 'lifecyclestage'],
    sorts=[{'propertyName': 'createdate', 'direction': 'DESCENDING'}]
)

# Example: Find open deals above $10K
big_deals = search_contacts(hs,
    filters=[
        {'propertyName': 'amount', 'operator': 'GTE', 'value': '10000'},
        {'propertyName': 'dealstage', 'operator': 'NEQ', 'value': 'closedwon'},
        {'propertyName': 'dealstage', 'operator': 'NEQ', 'value': 'closedlost'}
    ],
    properties=['dealname', 'amount', 'dealstage', 'closedate']
)

4. Associations

def associate(hs, from_type, from_id, to_type, to_id, assoc_type):
    """Create association between two objects."""
    return hs.put(
        f'/crm/v4/objects/{from_type}/{from_id}/associations/{to_type}/{to_id}',
        data=[{'associationCategory': 'HUBSPOT_DEFINED', 'associationTypeId': assoc_type}]
    )

# Common association type IDs:
# Contact -> Company: 1
# Company -> Contact: 2
# Deal -> Contact: 3
# Contact -> Deal: 4
# Deal -> Company: 5
# Company -> Deal: 6
# Ticket -> Contact: 15
# Ticket -> Company: 25

# Associate contact with company
associate(hs, 'contacts', contact_id, 'companies', company_id, 1)

# Associate deal with contact and company
associate(hs, 'deals', deal_id, 'contacts', contact_id, 3)
associate(hs, 'deals', deal_id, 'companies', company_id, 5)

# Get all associations for a contact
associations = hs.get(f'/crm/v4/objects/contacts/{contact_id}/associations/deals')

5. Pagination

def get_all_contacts(hs, properties=None, limit=None):
    """Paginate through all contacts."""
    all_contacts = []
    after = None
    params = {'limit': 100}
    if properties:
        params['properties'] = ','.join(properties)

    while True:
        if after:
            params['after'] = after
        resp = hs.get('/crm/v3/objects/contacts', params=params)
        all_contacts.extend(resp.get('results', []))

        paging = resp.get('paging')
        if paging and paging.get('next'):
            after = paging['next']['after']
        else:
            break

        if limit and len(all_contacts) >= limit:
            break

    return all_contacts[:limit] if limit else all_contacts

Best Practices

• Use batch endpoints: Create/update/read up to 100 records per call instead of one-by-one
• Respect rate limits: 100 requests per 10 seconds for OAuth apps, 150K per day
• Search indexing delay: New/updated records take up to 10 seconds to appear in search
• Use properties parameter: Always specify which properties to return, never fetch all
• Handle pagination: Always check for paging.next.after in responses
• Association types matter: Use the v4 associations API with explicit type IDs
• Idempotency via email: HubSpot deduplicates contacts by email automatically
• Webhook signatures: Always verify webhook signatures with your client secret

Common Pitfalls

• Search delay: Creating a record then immediately searching for it returns empty
• Rate limit shared across endpoints: 100 req/10s applies to ALL endpoints combined
• Lifecycle stage direction: Cannot move lifecycle stage backward without clearing it first
• Property internal names: API uses dealstage not Deal Stage, closedate not Close Date
• Archived records: Deleted records are archived, not removed; they can reappear in search

Anti-Patterns

• One-by-One Operations: Creating 1,000 contacts with 1,000 API calls. Use batch create (10 calls).
• Fetching All Properties: Not specifying properties returns minimal data, then making a second call.
• Polling for Changes: Checking every record every hour. Use webhooks or the changes API.
• Ignoring Associations: Creating deals without associating them to contacts and companies.