Sync enrichment into a CRM

The JSON bulk email API is the preferred workflow for CRM, warehouse, and application integrations. It accepts up to 100,000 records, returns immediately, and preserves your external_id on every result.

Bulk email jobs require a user-scoped API key. Each external_id must be unique within the submitted job.

Submit an idempotent job

Use an Idempotency-Key derived from the source batch. Store the key before submission. If the initial response is lost, you can safely retry the identical request with the same key.

curl --fail-with-body "https://api.leadx.com/v1/emails/find/bulk/jobs" \
  -H "X-API-KEY: $LEADX_API_KEY" \
  -H "Idempotency-Key: crm-import-2026-08-11-001" \
  -H "Content-Type: application/json" \
  --data '{
    "records": [
      {
        "external_id": "crm-1001",
        "first_name": "Jane",
        "last_name": "Smith",
        "url": "acmeplumbing.com"
      },
      {
        "external_id": "crm-1002",
        "linkedin_url": "https://www.linkedin.com/in/john-doe"
      }
    ],
    "notify_email": false
  }'

{
  "success": true,
  "id": "6cd80f18-9b2d-4f16-bb2e-0d8ee8a1f56a",
  "status": "queued",
  "total": 2,
  "idempotent_replay": false,
  "status_url": "https://api.leadx.com/v1/emails/find/bulk/jobs/6cd80f18-9b2d-4f16-bb2e-0d8ee8a1f56a",
  "results_url": "https://api.leadx.com/v1/emails/find/bulk/jobs/6cd80f18-9b2d-4f16-bb2e-0d8ee8a1f56a/results"
}

Poll and collect results

This Python example safely retries the idempotent submission and status GET requests, applies a bounded polling interval, and retrieves every result page.

import os
import random
import time
import requests

API_KEY = os.environ["LEADX_API_KEY"]
BASE_URL = "https://api.leadx.com/v1"
HEADERS = {"X-API-KEY": API_KEY}
IDEMPOTENCY_KEY = "crm-import-2026-08-11-001"
RETRYABLE_STATUSES = {429, 502, 503, 504}
session = requests.Session()


def retry_delay(response, attempt):
    retry_after = response.headers.get("Retry-After")
    if retry_after:
        return max(0, float(retry_after))
    return min(30, 2 ** attempt) + random.uniform(0, 0.5)


def request_json(method, url, **kwargs):
    for attempt in range(4):
        try:
            response = session.request(
                method,
                url,
                timeout=(10, 30),
                **kwargs,
            )
        except requests.RequestException:
            if attempt == 3:
                raise
            time.sleep(min(30, 2 ** attempt) + random.uniform(0, 0.5))
            continue

        if response.status_code not in RETRYABLE_STATUSES:
            response.raise_for_status()
            return response.json()
        if attempt == 3:
            response.raise_for_status()
        time.sleep(retry_delay(response, attempt))

    raise RuntimeError("Retry loop ended without a response")

records = [
    {
        "external_id": "crm-1001",
        "first_name": "Jane",
        "last_name": "Smith",
        "url": "acmeplumbing.com",
    },
    {
        "external_id": "crm-1002",
        "linkedin_url": "https://www.linkedin.com/in/john-doe",
    },
]

job = request_json(
    "POST",
    f"{BASE_URL}/emails/find/bulk/jobs",
    headers={
        **HEADERS,
        "Idempotency-Key": IDEMPOTENCY_KEY,
    },
    json={"records": records, "notify_email": False},
)

poll_delay = 5.0
poll_deadline = time.monotonic() + (2 * 60 * 60)
while True:
    status = request_json("GET", job["status_url"], headers=HEADERS)["data"]
    print(f'{status["processed"]}/{status["total"]} ({status["progress_pct"]}%)')

    if status["status"].lower() in {"done", "failed", "halted"}:
        break
    if time.monotonic() >= poll_deadline:
        raise TimeoutError(f'Job {job["id"]} exceeded the polling deadline')
    time.sleep(poll_delay + random.uniform(0, 0.5))
    poll_delay = min(30, poll_delay * 1.5)

results = []
cursor = -1

while True:
    page = request_json(
        "GET",
        job["results_url"],
        headers=HEADERS,
        params={"cursor": cursor, "limit": 1000},
    )
    results.extend(page["data"])

    if not page["has_more"]:
        break
    cursor = page["next_cursor"]

results_by_external_id = {row["external_id"]: row for row in results}
print(results_by_external_id)

if status["status"].lower() != "done":
    raise RuntimeError(
        f'Job {job["id"]} ended as {status["status"]}; partial results were preserved'
    )

Interpret each row

Every result has one of four states:

Result	Meaning
`pending`	The row has not been processed yet
`found`	An email was returned
`not_found`	Processing completed without an email
`error`	The row failed; inspect `error` and `reason`

The row also includes credits_debited, processed_at, reason, and the original lookup fields. See Results and billing for the distinction between processing, match, quality, and credit outcomes.

Production safeguards

Persist the idempotency key with the source batch before submission.
Persist the job ID immediately after receiving it.
Reuse the same idempotency key only for the same logical batch.
Merge by external_id, not row position.
Treat queued, running, done, failed, and halted as the complete public job-state set.
Poll no faster than your application needs and stop at a defined deadline.
Read partial results while a large job is running when early rows are useful.
Treat done, failed, and halted as terminal states and retain the job ID and partial results.
Retrieve and archive results promptly because the API does not publish a retention duration.
Keep API keys in a secrets manager and never write them into CRM fields or logs.

See Create a bulk email enrichment job for the request and initial response schemas. Review Reliability and retries before adapting the retry policy.