CRM Bulk Import - Zeeg Developer Hub

This guide walks you through migrating contacts, companies, and custom records from another system — HubSpot, Salesforce, Airtable, a spreadsheet — into the Zeeg CRM via the API. The flow is built around the assert (upsert) endpoints, so an import is idempotent: re-running the same job will update the existing rows instead of creating duplicates.

The CRM API requires a workspace on a plan that includes CRM. Custom objects require a plan that supports them — see the CRM overview for the full schema.

Prerequisites

Before you start, make sure you have:

An API token with the crm:write scope. crm:read is sufficient for discovery, but every write call requires crm:write.
A source export — typically a CSV, JSON dump, or a query against the source system’s API.
A clear dedup key for each entity in your source data:
- Companies → a domain (e.g. acme.com)
- People → an email address
- Custom records → any attribute you mark isUnique

The companies assert endpoint only supports domain as the matching attribute. If your source has companies without a domain, you must either enrich them first or fall back to plain POST /crm/companies and dedupe on your side.

Migration flow at a glance

1. Discover the target schema

Before importing, list all CRM objects and their attribute definitions. This tells you exactly which keys your payloads need to use.

curl -X GET "https://api.zeeg.me/v2/crm/objects" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Accept: application/json"

import requests

response = requests.get(
    "https://api.zeeg.me/v2/crm/objects",
    headers={
        "Authorization": "Bearer YOUR_TOKEN",
        "Accept": "application/json",
    },
)

schema = response.json()

const response = await fetch("https://api.zeeg.me/v2/crm/objects", {
  headers: {
    Authorization: "Bearer YOUR_TOKEN",
    Accept: "application/json",
  },
});

const schema = await response.json();

The response contains every standard and custom object with its full attribute list. Use it to build a mapping table from source fields → Zeeg attribute keys.

{
  "success": true,
  "status": 200,
  "collection": [
    {
      "slug": "people",
      "singularName": "Person",
      "pluralName": "People",
      "isStandard": true,
      "attributes": [
        { "key": "first_name", "label": "First Name", "type": "text", "isStandard": true },
        { "key": "last_name", "label": "Last Name", "type": "text", "isStandard": true },
        { "key": "emails", "label": "Emails", "type": "text", "isStandard": true },
        { "key": "phone_number", "label": "Phone Number", "type": "phone_number", "isStandard": true }
      ]
    },
    {
      "slug": "companies",
      "singularName": "Company",
      "pluralName": "Companies",
      "isStandard": true,
      "attributes": [
        { "key": "name", "label": "Name", "type": "text", "isStandard": true, "isRequired": true },
        { "key": "domain", "label": "Domain", "type": "text", "isStandard": true }
      ]
    }
  ]
}

Snapshot the schema once at the start of your import job and reuse it for the rest of the run — the schema is stable and there is no benefit to re-fetching it for every row.

2. Add custom attributes

If your source has fields that don’t fit the standard schema (a HubSpot Lifecycle Stage, a Salesforce Lead Score, an Airtable status), create a custom attribute before you import. Add a custom attribute with POST /crm/objects/{slug}/attributes. The attribute key is auto-generated from the label.

curl -X POST "https://api.zeeg.me/v2/crm/objects/people/attributes" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "select",
    "label": "Lifecycle Stage",
    "isRequired": false,
    "options": [
      { "value": "Lead", "color": "warning" },
      { "value": "MQL", "color": "sky" },
      { "value": "SQL", "color": "primary" },
      { "value": "Customer", "color": "success" }
    ]
  }'

import requests

response = requests.post(
    "https://api.zeeg.me/v2/crm/objects/people/attributes",
    headers={
        "Authorization": "Bearer YOUR_TOKEN",
        "Content-Type": "application/json",
    },
    json={
        "type": "select",
        "label": "Lifecycle Stage",
        "isRequired": False,
        "options": [
            {"value": "Lead", "color": "warning"},
            {"value": "MQL", "color": "sky"},
            {"value": "SQL", "color": "primary"},
            {"value": "Customer", "color": "success"},
        ],
    },
)

attribute = response.json()["attribute"]
print(attribute["key"])  # e.g. "lifecycle_stage"

const response = await fetch(
  "https://api.zeeg.me/v2/crm/objects/people/attributes",
  {
    method: "POST",
    headers: {
      Authorization: "Bearer YOUR_TOKEN",
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      type: "select",
      label: "Lifecycle Stage",
      isRequired: false,
      options: [
        { value: "Lead", color: "warning" },
        { value: "MQL", color: "sky" },
        { value: "SQL", color: "primary" },
        { value: "Customer", color: "success" },
      ],
    }),
  }
);

const { attribute } = await response.json();

Type-specific required fields — see Add an attribute to a CRM object for the full reference:

Attribute type	Required fields
`select`, `multiselect`, `status`	`options` (max 20)
`relation`	`relationType`, `relatedObjectSlug`, `relatedObjectLabel`
`text`, `phone_number`, `number`	`isUnique`
`currency`	`currency`, `currencyDisplay`, `currencyDecimal`, `currencyGrouping`

The attribute type cannot be changed after creation. Pick it carefully — switching from text to select later requires deleting and recreating the attribute, which loses any data already imported into it.

If you plan to use a custom attribute as the matching key for assert (e.g. an external_id carried over from the source system), set isUnique: true when you create it.

3. Create custom objects

If the source system has entities that aren’t people or companies — deals, properties, tickets, vehicles — create a custom object before importing its records.

curl -X POST "https://api.zeeg.me/v2/crm/objects" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "slug": "deals",
    "singularName": "Deal",
    "pluralName": "Deals"
  }'

import requests

response = requests.post(
    "https://api.zeeg.me/v2/crm/objects",
    headers={
        "Authorization": "Bearer YOUR_TOKEN",
        "Content-Type": "application/json",
    },
    json={
        "slug": "deals",
        "singularName": "Deal",
        "pluralName": "Deals",
    },
)

const response = await fetch("https://api.zeeg.me/v2/crm/objects", {
  method: "POST",
  headers: {
    Authorization: "Bearer YOUR_TOKEN",
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    slug: "deals",
    singularName: "Deal",
    pluralName: "Deals",
  }),
});

A new custom object only ships with the system attributes (id, created_at, updated_at). Add the rest with the attributes endpoint described above — at minimum, add a unique external_id (or equivalent) text attribute so you can upsert records by it.

4. Upsert companies

The PUT /crm/companies endpoint is the recommended path for imports. Send the full payload and let Zeeg decide whether to create or update:

200 OK — a company with the same domain already existed and was updated.
201 Created — no match was found and a new record was created.

The matching attribute for companies is always domain — no other attribute is supported on the companies upsert.

curl -X PUT "https://api.zeeg.me/v2/crm/companies?matchingAttribute=domain" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Acme Corp",
    "domain": "acme.com",
    "description": "A global provider of anvils and gadgets.",
    "websiteUrl": "https://acme.com",
    "primaryLocation": "Phoenix, AZ",
    "socials": {
      "linkedin": "https://linkedin.com/company/acme",
      "twitter": "https://twitter.com/acme"
    }
  }'

import requests

response = requests.put(
    "https://api.zeeg.me/v2/crm/companies",
    params={"matchingAttribute": "domain"},
    headers={
        "Authorization": "Bearer YOUR_TOKEN",
        "Content-Type": "application/json",
    },
    json={
        "name": "Acme Corp",
        "domain": "acme.com",
        "description": "A global provider of anvils and gadgets.",
        "websiteUrl": "https://acme.com",
        "primaryLocation": "Phoenix, AZ",
        "socials": {
            "linkedin": "https://linkedin.com/company/acme",
            "twitter": "https://twitter.com/acme",
        },
    },
)

company = response.json()["company"]
company_id = company["id"]  # save this for linking people later

const response = await fetch(
  "https://api.zeeg.me/v2/crm/companies?matchingAttribute=domain",
  {
    method: "PUT",
    headers: {
      Authorization: "Bearer YOUR_TOKEN",
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      name: "Acme Corp",
      domain: "acme.com",
      description: "A global provider of anvils and gadgets.",
      websiteUrl: "https://acme.com",
      primaryLocation: "Phoenix, AZ",
      socials: {
        linkedin: "https://linkedin.com/company/acme",
        twitter: "https://twitter.com/acme",
      },
    }),
  }
);

const { company } = await response.json();

Keep a domain → company.id map in memory as you iterate — you will need the company UUID when upserting people in the next step.

5. Upsert people

People work the same way, but the matching attribute is configurable. Use email for typical contact imports — it matches against both the primary and any secondary emails on a person.

curl -X PUT "https://api.zeeg.me/v2/crm/people?matchingAttribute=email" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "firstName": "Jane",
    "lastName": "Doe",
    "emails": ["jane.doe@acme.com"],
    "jobTitle": "Head of Product",
    "phoneNumber": "+4930123456789",
    "primaryLocation": "Berlin, Germany",
    "companyId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "socials": {
      "linkedin": "https://linkedin.com/in/janedoe"
    }
  }'

import requests

response = requests.put(
    "https://api.zeeg.me/v2/crm/people",
    params={"matchingAttribute": "email"},
    headers={
        "Authorization": "Bearer YOUR_TOKEN",
        "Content-Type": "application/json",
    },
    json={
        "firstName": "Jane",
        "lastName": "Doe",
        "emails": ["jane.doe@acme.com"],
        "jobTitle": "Head of Product",
        "phoneNumber": "+4930123456789",
        "primaryLocation": "Berlin, Germany",
        "companyId": company_id,  # from step 4
        "socials": {
            "linkedin": "https://linkedin.com/in/janedoe",
        },
    },
)

const response = await fetch(
  "https://api.zeeg.me/v2/crm/people?matchingAttribute=email",
  {
    method: "PUT",
    headers: {
      Authorization: "Bearer YOUR_TOKEN",
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      firstName: "Jane",
      lastName: "Doe",
      emails: ["jane.doe@acme.com"],
      jobTitle: "Head of Product",
      phoneNumber: "+4930123456789",
      primaryLocation: "Berlin, Germany",
      companyId: companyId, // from step 4
      socials: {
        linkedin: "https://linkedin.com/in/janedoe",
      },
    }),
  }
);

The matchingAttribute query parameter accepts any standard or custom attribute slug on people. If you tracked a stable external_id from the source system (and created it as a unique custom attribute in step 2), pass matchingAttribute=external_id instead — that’s the most robust dedup key for ongoing syncs because it survives email changes.

The first entry in emails becomes the primary email and is the one used for matching when matchingAttribute=email. Make sure the source system’s primary email maps to position [0].

6. Upsert custom records

Custom records use the same upsert pattern via PUT /crm/{objectSlug}, with matchingAttribute set to any unique attribute on the object (typically external_id or a domain-specific key like sku). The request body is a flat object of attributeSlug → value pairs.

curl -X PUT "https://api.zeeg.me/v2/crm/deals?matchingAttribute=external_id" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "external_id": "HUBSPOT-DEAL-12345",
    "name": "Acme — Q3 Renewal",
    "amount": 24000,
    "stage": "Negotiation"
  }'

import requests

response = requests.put(
    "https://api.zeeg.me/v2/crm/deals",
    params={"matchingAttribute": "external_id"},
    headers={
        "Authorization": "Bearer YOUR_TOKEN",
        "Content-Type": "application/json",
    },
    json={
        "external_id": "HUBSPOT-DEAL-12345",
        "name": "Acme — Q3 Renewal",
        "amount": 24000,
        "stage": "Negotiation",
    },
)

const response = await fetch(
  "https://api.zeeg.me/v2/crm/deals?matchingAttribute=external_id",
  {
    method: "PUT",
    headers: {
      Authorization: "Bearer YOUR_TOKEN",
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      external_id: "HUBSPOT-DEAL-12345",
      name: "Acme — Q3 Renewal",
      amount: 24000,
      stage: "Negotiation",
    }),
  }
);

7. Link records via relations

The companyId field on a person is enough for the standard person-to-company link. For any other relation — many-to-many, custom-object-to-custom-object, person-to-deal — use the relation endpoint. PATCH /crm/{objectSlug}/{recordId}/relation adds or removes IDs from a relation or user attribute. Supply add, remove, or both. The same ID cannot appear in both arrays.

curl -X PATCH "https://api.zeeg.me/v2/crm/deals/c1d2e3f4-a5b6-7890-cdef-123456789012/relation" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "attributeSlug": "contacts",
    "add": [
      "b2c3d4e5-f6a7-8901-bcde-f23456789012",
      "d4e5f6a7-b8c9-0123-defa-456789012345"
    ],
    "remove": []
  }'

import requests

response = requests.patch(
    "https://api.zeeg.me/v2/crm/deals/c1d2e3f4-a5b6-7890-cdef-123456789012/relation",
    headers={
        "Authorization": "Bearer YOUR_TOKEN",
        "Content-Type": "application/json",
    },
    json={
        "attributeSlug": "contacts",
        "add": [
            "b2c3d4e5-f6a7-8901-bcde-f23456789012",
            "d4e5f6a7-b8c9-0123-defa-456789012345",
        ],
        "remove": [],
    },
)

const response = await fetch(
  "https://api.zeeg.me/v2/crm/deals/c1d2e3f4-a5b6-7890-cdef-123456789012/relation",
  {
    method: "PATCH",
    headers: {
      Authorization: "Bearer YOUR_TOKEN",
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      attributeSlug: "contacts",
      add: [
        "b2c3d4e5-f6a7-8901-bcde-f23456789012",
        "d4e5f6a7-b8c9-0123-defa-456789012345",
      ],
      remove: [],
    }),
  }
);

The endpoint is also the way to populate user attributes (e.g. setting an Account Owner on a company) — pass the workspace member’s ID in add.

Batching, rate limits, and error handling

The CRM API rate-limits per endpoint and per workspace — see Rate Limits for the general policy. For a bulk import:

Send sequentially or in small parallel batches. Five to ten parallel workers is usually a safe starting point; back off if you start seeing 429s.
Implement exponential backoff on 429. Sleep 2^attempt + jitter seconds, up to a 60-second cap.
Log failed rows, don’t abort. A failed row should not stop the whole import. Capture the row, the request, and the error response so you can re-run only the failures — the upsert semantics make this safe.
Validate locally first. Required fields, type constraints, and length limits are checked server-side; pre-validating saves round trips on a large run.

A minimal upsert helper with retries:

import time
import random
import requests

BASE_URL = "https://api.zeeg.me/v2"
TOKEN = "YOUR_TOKEN"

def upsert(path, payload, matching_attribute, max_attempts=6):
    url = f"{BASE_URL}{path}"
    headers = {
        "Authorization": f"Bearer {TOKEN}",
        "Content-Type": "application/json",
    }
    params = {"matchingAttribute": matching_attribute}

    for attempt in range(max_attempts):
        response = requests.put(url, params=params, headers=headers, json=payload)

        if response.status_code in (200, 201):
            return response.json()

        if response.status_code == 429:
            sleep = min(2 ** attempt + random.random(), 60)
            time.sleep(sleep)
            continue

        # Non-retriable: 4xx (other than 429) — surface to the caller for logging.
        response.raise_for_status()

    raise RuntimeError(f"Gave up after {max_attempts} attempts: {url}")

const BASE_URL = "https://api.zeeg.me/v2";
const TOKEN = "YOUR_TOKEN";

async function upsert(path, payload, matchingAttribute, maxAttempts = 6) {
  const url = `${BASE_URL}${path}?matchingAttribute=${encodeURIComponent(matchingAttribute)}`;
  const headers = {
    Authorization: `Bearer ${TOKEN}`,
    "Content-Type": "application/json",
  };

  for (let attempt = 0; attempt < maxAttempts; attempt++) {
    const response = await fetch(url, {
      method: "PUT",
      headers,
      body: JSON.stringify(payload),
    });

    if (response.status === 200 || response.status === 201) {
      return await response.json();
    }

    if (response.status === 429) {
      const sleep = Math.min(2 ** attempt + Math.random(), 60) * 1000;
      await new Promise((r) => setTimeout(r, sleep));
      continue;
    }

    // Non-retriable: surface for logging.
    throw new Error(`${response.status} ${await response.text()}`);
  }

  throw new Error(`Gave up after ${maxAttempts} attempts: ${url}`);
}

For the full set of error codes and response shapes, see Errors.

Worked example: importing a CSV

A complete script that imports companies and people from two CSV files. The dedup keys are domain for companies and email for people; re-running the script updates existing rows without creating duplicates. It uses the upsert helper from the previous section so 429s are retried automatically.

Python

import csv

# Reuses the `upsert(path, payload, matching_attribute)` helper defined above.

# 1. Import companies — keep a domain → id map for the people pass.
domain_to_company_id = {}

with open("companies.csv") as f:
    for row in csv.DictReader(f):
        if not row.get("domain"):
            print(f"SKIP company without domain: {row['name']}")
            continue

        payload = {
            "name": row["name"],
            "domain": row["domain"],
            "description": row.get("description") or None,
            "websiteUrl": row.get("website") or None,
            "primaryLocation": row.get("location") or None,
        }

        try:
            result = upsert("/crm/companies", payload, "domain")
        except Exception as e:
            print(f"FAIL company {row['domain']}: {e}")
            continue

        domain_to_company_id[row["domain"]] = result["company"]["id"]

# 2. Import people — link to company by domain.
with open("people.csv") as f:
    for row in csv.DictReader(f):
        if not row.get("email"):
            print(f"SKIP person without email: {row.get('first_name')} {row.get('last_name')}")
            continue

        payload = {
            "firstName": row.get("first_name") or None,
            "lastName": row.get("last_name") or None,
            "emails": [row["email"]],
            "jobTitle": row.get("job_title") or None,
            "phoneNumber": row.get("phone") or None,
            "companyId": domain_to_company_id.get(row.get("company_domain")),
        }

        try:
            upsert("/crm/people", payload, "email")
        except Exception as e:
            print(f"FAIL person {row['email']}: {e}")

For larger imports (tens of thousands of rows), run this in batches with a small worker pool and write failures to a separate CSV so you can re-run only the failed rows.

The same script doubles as a recurring sync job. Schedule it on a cron, point it at a fresh export, and the assert semantics will keep Zeeg in step with the source system without manual reconciliation.

Next steps

Subscribe to webhooks to receive updates when CRM-linked events are scheduled or cancelled.
Browse the full CRM API reference for every endpoint, including delete, individual GET, and the non-upsert POST/PATCH variants.

​Prerequisites

​Migration flow at a glance

​1. Discover the target schema

​2. Add custom attributes

​3. Create custom objects

​4. Upsert companies

​5. Upsert people

​6. Upsert custom records

​7. Link records via relations

​Batching, rate limits, and error handling

​Worked example: importing a CSV

​Next steps

Prerequisites

Migration flow at a glance

1. Discover the target schema

2. Add custom attributes

3. Create custom objects

4. Upsert companies

5. Upsert people

6. Upsert custom records

7. Link records via relations

Batching, rate limits, and error handling

Worked example: importing a CSV

Next steps