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

That single line is the whole security story for a script running on your own machine. The moment this sync moves onto a server or into a scheduled job, a file on disk stops being good enough — Manage API Keys Safely in Production walks through the alternatives and how to rotate a token you have already leaked.

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

resp = client.get("/crm/v3/objects/contacts", params={"limit": 1})  # one row proves the token works
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.

It is worth picturing where that token actually travels, because you only type it once. It is read out of .env when the script starts, handed to the httpx.Client, and from then on the client stamps it onto every outgoing request as an Authorization header. HubSpot answers with a JSON body holding two things you will use constantly: a results list and, when there is more to fetch, a paging object.

The timeout=30.0 argument is not decoration either. Without it a stalled connection can leave your script hanging with no output rather than failing quickly with a message you can act on. Fix Connection and Timeout Errors with AI APIs covers the handful of network errors worth catching by name and how to retry them without hammering the server.

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.

The loop has exactly one exit, and beginners usually trip over it. HubSpot keeps handing you a cursor for as long as pages remain; the response after the final page simply arrives without a paging key. That absence is the stop signal — not an error, not an empty results list.

Naming your properties up front matters more than it looks. Ask for nothing and HubSpot returns a small default set; ask for everything and each page carries fields you will never read. On an account with tens of thousands of contacts that difference shows up as minutes of waiting.

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.

Drawn out, the whole function is a single fork. One search decides which of two writes you make, so no contact ever costs you more than two requests.

Email works as the match key because HubSpot treats it as a unique identifier for contacts. Messier source data breaks that assumption fast — the same person under two spellings, or a personal address alongside a work one, will sail through as two separate records. Remove Duplicate Records with Embeddings in Python shows how to collapse near-matches before they reach the API.

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

SOURCE = [  # contacts you want to push into HubSpot
    {"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. Because every write is an upsert, the script is safe to run again. A second pass over the same source list refreshes the same records instead of doubling them, and that property is what turns a one-off import into something you can leave on a schedule.

From here you can put it on a timer — Schedule Python AI Jobs with GitHub Actions runs this exact script nightly without a server of your own — or trigger it from a webhook whenever a form is submitted.

Key parameter quick-reference

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.
5. 404 Not Found on a PATCH — The contact id vanished between your search and your write, usually because a colleague deleted or merged the record. Catch the 404, fall back to a POST, and the next run repairs itself.

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.

Related guides

• CRM Data Integration with AI — the main guide for connecting your CRM to Python and AI.
• Enrich CRM Leads with AI in Python — auto-fill missing lead details after you sync them.
• Summarize Sales Calls to Your CRM with Python — push AI-written call summaries onto your contacts.
• Building AI-Powered Business Applications — the wider track this section belongs to.