Business Apps

Sync HubSpot Contacts with Python

This guide shows you how to pull and push HubSpot contacts from Python in under 15 minutes, using a private-app token and the lightweight httpx HTTP client. You will read every contact in your account, then create or update records without making duplicates.

A contact in HubSpot is a person record (name, email, company, and so on). The CRM API is the web interface HubSpot exposes so your code can read and write those records. We talk to it over plain HTTP, so there is no special SDK to learn. If HTTP requests and tokens are new to you, the parent section CRM Data Integration with AI walks through the bigger picture first.

Prerequisites

You need Python 3.10 or newer and two small packages. httpx makes the API calls and python-dotenv loads your token from a file so it never lives in your code.

pip install httpx python-dotenv

Next, create the token. In HubSpot, open Settings → Integrations → Private Apps → Create a private app. Under the Scopes tab, tick crm.objects.contacts.read and crm.objects.contacts.write, then create the app and copy the access token it shows you.

Save that token in a file named .env in your project folder:

HUBSPOT_TOKEN=pat-na1-your-token-here

Add .env to your .gitignore now so the token never gets committed to a repository.

echo ".env" >> .gitignore

Step 1: Authenticate and make your first call

Every request to HubSpot carries your token in an Authorization: Bearer header. The snippet below loads the token, builds a reusable httpx.Client (which keeps the connection open and attaches the header to every call), and confirms the credentials work by fetching a single contact.

import os
import httpx
from dotenv import load_dotenv

load_dotenv()
TOKEN = os.environ["HUBSPOT_TOKEN"]
BASE = "https://api.hubapi.com"

client = httpx.Client(
    base_url=BASE,
    headers={"Authorization": f"Bearer {TOKEN}"},
    timeout=30.0,
)

# Ask for just one contact to confirm the token works.
resp = client.get("/crm/v3/objects/contacts", params={"limit": 1})
resp.raise_for_status()  # turns any 4xx/5xx into a clear Python error
print(resp.json())

If you see a JSON object back, your token and scopes are correct. A 401 here means the token is wrong or missing; a 403 means it lacks the contacts scopes.

Step 2: Pull every contact with pagination

The list endpoint returns at most 100 contacts per call. To read more, you follow a cursor (a bookmark HubSpot hands back so the next call resumes where the last one stopped). The response includes paging.next.after while more pages remain, and omits it on the final page. Loop until it disappears.

def fetch_all_contacts(client, properties=None):
    contacts = []
    after = None
    params = {"limit": 100}
    if properties:
        # Ask only for the fields you need to keep responses small.
        params["properties"] = ",".join(properties)

    while True:
        if after:
            params["after"] = after
        resp = client.get("/crm/v3/objects/contacts", params=params)
        resp.raise_for_status()
        data = resp.json()

        contacts.extend(data["results"])

        paging = data.get("paging")
        if paging and "next" in paging:
            after = paging["next"]["after"]
        else:
            break  # no cursor means we read the last page

    return contacts


people = fetch_all_contacts(client, properties=["email", "firstname", "lastname"])
print(f"Pulled {len(people)} contacts")
for person in people[:3]:
    props = person["properties"]
    print(person["id"], props.get("email"), props.get("firstname"))

Each item in results has a stable id (the contact's internal HubSpot id) and a properties dictionary holding the fields you requested. You will use that id in the next step to update records.

Step 3: Upsert a single contact

An upsert means "update if it already exists, otherwise create it." HubSpot has no single upsert call for contacts, so you search by email first. If the search returns a match you PATCH that record by its id; if not, you POST a new one. Both endpoints expect a {"properties": {...}} body.

def upsert_contact(client, email, properties):
    body = {"properties": {"email": email, **properties}}

    # 1. Look for an existing contact with this email.
    search = client.post(
        "/crm/v3/objects/contacts/search",
        json={
            "filterGroups": [{
                "filters": [
                    {"propertyName": "email", "operator": "EQ", "value": email}
                ]
            }],
            "properties": ["email"],
            "limit": 1,
        },
    )
    search.raise_for_status()
    results = search.json()["results"]

    if results:
        # 2a. Found one: update it by id.
        contact_id = results[0]["id"]
        resp = client.patch(f"/crm/v3/objects/contacts/{contact_id}", json=body)
    else:
        # 2b. None found: create a new contact.
        resp = client.post("/crm/v3/objects/contacts", json=body)

    resp.raise_for_status()
    return resp.json()


saved = upsert_contact(
    client,
    email="ada@example.com",
    properties={"firstname": "Ada", "lastname": "Lovelace", "company": "Analytical Engines"},
)
print("Saved contact id:", saved["id"])

Run this twice with the same email and you will see the same id both times: the first call creates the contact, the second updates it in place rather than making a duplicate.

Step 4: Run a full two-way sync

Now combine the pieces. The script below pulls everyone from HubSpot to show what you already have, then upserts a small source list (imagine it came from a spreadsheet or signup form). A short time.sleep between writes keeps you under HubSpot's rate limit. The whole thing is wrapped so the client always closes cleanly.

import os
import time
import httpx
from dotenv import load_dotenv

load_dotenv()
TOKEN = os.environ["HUBSPOT_TOKEN"]

# Contacts you want to push into HubSpot.
SOURCE = [
    {"email": "grace@example.com", "firstname": "Grace", "lastname": "Hopper"},
    {"email": "alan@example.com", "firstname": "Alan", "lastname": "Turing"},
]

with httpx.Client(
    base_url="https://api.hubapi.com",
    headers={"Authorization": f"Bearer {TOKEN}"},
    timeout=30.0,
) as client:
    # Pull side: see what is already there.
    existing = fetch_all_contacts(client, properties=["email"])
    print(f"HubSpot currently has {len(existing)} contacts")

    # Push side: upsert each source record.
    for row in SOURCE:
        email = row.pop("email")
        try:
            result = upsert_contact(client, email, row)
            print(f"Synced {email} -> {result['id']}")
        except httpx.HTTPStatusError as err:
            if err.response.status_code == 429:
                wait = int(err.response.headers.get("Retry-After", "10"))
                print(f"Rate limited; pausing {wait}s")
                time.sleep(wait)
            else:
                raise
        time.sleep(0.2)  # stay comfortably under the rate limit

This is the core of a repeatable sync: pull to understand the current state, then push your changes with upserts. From here you can schedule it to run on a timer or trigger it from a webhook.

Key parameter quick-reference

ParameterTypeDefaultEffect
limitint10Contacts per page on the list endpoint; max is 100.
afterstringnonePaging cursor from paging.next.after; resumes the list at the next page.
propertiescomma-separated stringa few defaultsWhich contact fields to return; request only what you need.
Retry-Afterresponse header (seconds)noneOn a 429, how long to wait before retrying.

Troubleshooting

  1. 401 Unauthorized — The token is missing, mistyped, or expired. Confirm .env holds the full pat-na1-... string and that load_dotenv() runs before you read it. Regenerate the token in the private app if needed.
  2. 403 Forbidden — The token is valid but lacks a scope. Open your private app, add crm.objects.contacts.read and crm.objects.contacts.write, save, and copy the refreshed token.
  3. 429 Too Many Requests — You exceeded the rate limit. Read the Retry-After header, time.sleep for that many seconds, and retry. Adding a small sleep between writes, as in Step 4, usually prevents it.
  4. KeyError: 'paging' or an endless loop — You are reading the cursor incorrectly. Use data.get("paging") and break when it is absent; the final page has no paging key at all.

When to use this vs. alternatives

  • Use this httpx approach when you want full control, minimal dependencies, and a clear view of exactly what each request sends. It is ideal for scripts, scheduled jobs, and small business apps where you would rather not learn a heavier library.
  • Use the official hubspot-api-client SDK when you call many different HubSpot objects (deals, tickets, companies) and want typed models and built-in retries. It hides the raw HTTP at the cost of an extra dependency and some abstraction.
  • Use HubSpot's no-code imports or workflows when you only need a one-time CSV upload or a simple in-app automation. Reach for Python the moment you need custom logic, scheduling, or to combine HubSpot with another system.

Once your contacts are flowing in cleanly, the natural next step is to make them more useful: Enrich CRM Leads with AI in Python fills in missing fields automatically, and Summarize Sales Calls to Your CRM with Python writes call notes straight onto the contact. Back to CRM Data Integration with AI.