Export email data to Pipedrive

Pipedrive is deal-centric, which changes how you model email differently from Salesforce or HubSpot. Everything in Pipedrive ultimately ladders up to the visual pipeline, so the right shape for an email sync is Organization → Person → Deal → Activity, with the Deal step gated by an engagement threshold (otherwise every random sender gets a deal and your pipeline turns into noise).

This recipe walks through the four-object hierarchy, an engagement-threshold heuristic for Deal auto-creation, and the custom-field pattern for capturing email-specific signals.

The hierarchy

Organization (acme.test)
└── Person ([email protected])
    └── Deal (Q2 expansion — Acme)
        └── Activities (emails, calls, meetings)

In code terms: a sender’s domain becomes an Organization, the sender themselves a Person, repeated correspondence flips a switch to create a Deal, and every email becomes an Activity.

API endpoints

Four endpoints carry the work:

POST /api/v1/organizations — create an Organization from a domain
POST /api/v1/persons — create a Person, link to Organization
POST /api/v1/deals — create a Deal in a pipeline stage
POST /api/v1/activities — log an email as an Activity (type: "email", done: 1)

Pipedrive uses token-based rate limiting per the official rate-limit docs. Plan for ~80 requests/second on Professional and above.

Lead vs. Deal threshold

Not every sender deserves a Deal. The standard pattern:

Sender state	Pipedrive object
Brand-new sender, single message	`Lead` (sits in the Leads Inbox)
≥5 emails OR ≥1 reply OR a meeting on the calendar	`Person` + `Deal` in `Initial Contact` stage
Existing Person with ≥10 messages and no Deal	Promote: create `Deal`

The threshold is heuristic. Five messages is a reasonable default; tune up for high-volume sales teams (cold outreach inflates message counts) or down for low-velocity products.

def classify_sender(sender_state):
    if sender_state["msg_count"] >= 5 or sender_state["replies"] > 0 or sender_state["meetings"] > 0:
        return "person_with_deal"
    if sender_state["msg_count"] >= 1:
        return "lead"
    return "ignore"

Push the records

import requests

API = "https://yourcompany.pipedrive.com/api/v1"
TOKEN = PIPEDRIVE_TOKEN

def find_or_create_org(domain: str) -> int:
    # search first
    r = requests.get(f"{API}/organizations/search", params={
        "term": domain, "exact_match": "true", "api_token": TOKEN
    }).json()
    if r["data"]["items"]:
        return r["data"]["items"][0]["item"]["id"]
    # create
    r = requests.post(f"{API}/organizations", params={"api_token": TOKEN}, json={
        "name": domain,
    }).json()
    return r["data"]["id"]

def find_or_create_person(email: str, name: str, org_id: int) -> int:
    r = requests.get(f"{API}/persons/search", params={
        "term": email, "fields": "email", "api_token": TOKEN
    }).json()
    if r["data"]["items"]:
        return r["data"]["items"][0]["item"]["id"]
    r = requests.post(f"{API}/persons", params={"api_token": TOKEN}, json={
        "name": name,
        "email": [{"value": email, "primary": True}],
        "org_id": org_id,
    }).json()
    return r["data"]["id"]

def create_deal(person_id: int, org_id: int, title: str) -> int:
    r = requests.post(f"{API}/deals", params={"api_token": TOKEN}, json={
        "title": title,
        "person_id": person_id,
        "org_id": org_id,
        "stage_id": INITIAL_CONTACT_STAGE,
    }).json()
    return r["data"]["id"]

def log_email_activity(person_id: int, deal_id: int | None, msg: dict):
    requests.post(f"{API}/activities", params={"api_token": TOKEN}, json={
        "type": "email",
        "subject": msg["subject"],
        "due_date": msg["date"][:10],
        "done": 1,
        "person_id": person_id,
        "deal_id": deal_id,
        "note": msg["snippet"],
    })

Custom fields for email signals

Standard Person fields don’t capture email-specific signals. Add custom fields once via personFields:

custom = requests.post(f"{API}/personFields", params={"api_token": TOKEN}, json={
    "name": "Thread count",
    "field_type": "double",
}).json()

The response gives you a key like abc123_thread_count you store and reference when creating Persons:

requests.post(f"{API}/persons", params={"api_token": TOKEN}, json={
    "name": "Ada Lovelace",
    "email": [...],
    "org_id": org_id,
    "abc123_thread_count": 17,
    "abc123_last_contact": "2026-04-22",
    "abc123_avg_response_hours": 4.2,
}).json()

These are the fields sales actually want to filter and segment by. Add them once during setup and the rest of the sync populates them automatically.

Rate limit pacing

Pipedrive uses a token bucket — Professional and Enterprise tiers refill at roughly 80 tokens/sec, each request costs 1 token. Practical guidance:

import time

def with_pacing(fn, *args, **kwargs):
    while True:
        r = fn(*args, **kwargs)
        if r.status_code != 429:
            return r
        retry_after = float(r.headers.get("Retry-After", 1))
        time.sleep(retry_after)

The 429 response includes a Retry-After header. Honor it and you’ll never see sustained throttling.

Things to know

Leads vs. Persons. Leads live in a separate Leads Inbox, not in the main Persons table. Many teams forget this and end up with parallel records once a Lead is “qualified” — the convert workflow exists but is manual. Decide your conversion rules upfront.
Deal stages. Hardcode the stage ID by name lookup at script start; don’t rely on stable numeric IDs across environments.
Activity timestamps. due_date accepts YYYY-MM-DD only. For sub-day precision, populate the due_time field separately (HH:MM).