Running an async enrichment end to end

Beta. AgentSource v2 is in beta. Details may still change before GA.

When to use async

Every enrichment in v2 comes in two variants:

	Sync (`.../enrich`)	Async (`.../job`)
Input	A single ID or a list of up to 50	A dataset of up to 10,000 IDs
How you pass input	IDs in the request body	A `list_id` from the upload endpoint
Results	Returned in the response	Downloaded from a URL when the job finishes
Best for	Interactive lookups, small batches	Large batches, scheduled jobs

Async endpoints do not accept inline IDs. Every .../job request takes a list_id and nothing else (some enrichments also take a parameters object). You must upload your IDs first — that is step 1 below.

The flow

Upload a CSV of entity IDs

POST /v2/async/entity-id-datasets/upload returns a list_id.

Submit the job

POST /v2/{entity}/{enrichment}/job with that list_id returns a job_id.

Poll for status

GET /v2/jobs/status/{job_id} until the job finishes.

Download the results

The finished job exposes results.download_url — a CSV.

Step 1 — Prepare your CSV

The upload endpoint takes a CSV of entity IDs. Start from a sample:

sample-business-ids.csv

Six business IDs — download and use as-is

sample-prospect-ids.csv

Three prospect IDs

sample-business-ids.csv

business_id
e11943a6031a0e6114ae69c257617980
c17a9b84c3c3ade4cdce7373958ccb9c
8adce3ca1cef0c986b22310e369a0793
6eebb2af6bb0fb54b7ef435a98bb66d9
c5bfcfde8643f8870b5c5cbecb75cd22
d9ee9a9c0773759f16e6e458cbe14670

Get IDs of your own from Match businesses or Fetch businesses — or Match prospects and Fetch prospects for people.

Keep the file within the 10,000-row async limit. The upload response echoes a rowCount so you can confirm every row was ingested.

Step 2 — Upload the dataset

Send the file as multipart/form-data. All four fields are required.

Field	Value
`file`	Your CSV
`entity_type`	`business` or `prospect`
`name`	A short label for the dataset
`description`	What the dataset contains

curl --request POST \
  --url https://api.explorium.ai/v2/async/entity-id-datasets/upload \
  --header 'api_key: YOUR_API_KEY' \
  --form 'file=@sample-business-ids.csv' \
  --form 'entity_type=business' \
  --form 'name=Q1 target accounts' \
  --form 'description=Business IDs for the Q1 firmographics refresh'

import requests

BASE = "https://api.explorium.ai"
HEADERS = {"api_key": "YOUR_API_KEY"}

with open("sample-business-ids.csv", "rb") as f:
    resp = requests.post(
        f"{BASE}/v2/async/entity-id-datasets/upload",
        headers=HEADERS,
        files={"file": ("sample-business-ids.csv", f, "text/csv")},
        data={
            "entity_type": "business",
            "name": "Q1 target accounts",
            "description": "Business IDs for the Q1 firmographics refresh",
        },
    )
resp.raise_for_status()
dataset = resp.json()

list_id = dataset["list_id"]
print(f"uploaded {dataset['rowCount']} rows -> list_id={list_id}")
print(f"dataset expires at {dataset['expiresAt']}")

The response gives you the handle you need for step 2:

Response

{
  "list_id": "b8f2c1d4-...",
  "name": "Q1 target accounts",
  "description": "Business IDs for the Q1 firmographics refresh",
  "rowCount": 6,
  "entityType": "business",
  "tenantId": "...",
  "expiresAt": "2026-09-06T00:00:00Z"
}

Datasets expire. Note expiresAt and re-upload before running a job against a stale list.

Step 3 — Submit the job

Post the list_id to any .../job endpoint. This example uses Firmographics (async).

curl --request POST \
  --url https://api.explorium.ai/v2/businesses/firmographics/job \
  --header 'api_key: YOUR_API_KEY' \
  --header 'Content-Type: application/json' \
  --data '{
    "list_id": "b8f2c1d4-..."
  }'

resp = requests.post(
    f"{BASE}/v2/businesses/firmographics/job",
    headers={**HEADERS, "Content-Type": "application/json"},
    json={"list_id": list_id},
)
resp.raise_for_status()
job_id = resp.json()["job_id"]
print(f"submitted job {job_id}")

A few enrichments need a parameters object as well. company_website_keywords requires keywords, and both research endpoints take query or prompt_template plus an output_schema:

{
  "list_id": "b8f2c1d4-...",
  "parameters": { "keywords": ["cloud migration", "data platform"] }
}

Step 4 — Poll for status

Call the status endpoint until the job finishes. A finished job carries a results object.

curl --request GET \
  --url https://api.explorium.ai/v2/jobs/status/JOB_ID \
  --header 'api_key: YOUR_API_KEY'

import time

while True:
    resp = requests.get(f"{BASE}/v2/jobs/status/{job_id}", headers=HEADERS)
    resp.raise_for_status()
    job = resp.json()

    if job.get("error"):
        raise RuntimeError(f"job failed: {job['error']}")
    if job.get("results", {}).get("download_url"):
        break

    print(f"status={job['status']} ... waiting")
    time.sleep(30)

results = job["results"]
usage = job.get("credit_usage", {})
print(f"done — {usage.get('total_results')} results, {usage.get('total_credits')} credits")
print(f"download ({results['format']}) expires at {results['expires_at']}")

A completed response looks like this:

Response

{
  "job_id": "3f9a...",
  "status": "...",
  "input": { "list_id": "b8f2c1d4-...", "row_count": 6, "entity_type": "business" },
  "timing": {
    "created_at": "2026-08-09T10:00:00Z",
    "started_at": "2026-08-09T10:00:04Z",
    "finished_at": "2026-08-09T10:02:11Z"
  },
  "credit_usage": { "total_credits": 6, "total_results": 6 },
  "results": {
    "format": "csv",
    "download_url": "https://...",
    "expires_at": "2026-08-09T11:02:11Z",
    "file_expires_at": "2026-08-16T10:02:11Z"
  }
}

Job states are queued, running, succeeded, failed, cancelled, and expired. The sample above keys off results.download_url and error rather than matching status text — the more robust pattern either way.

Step 5 — Download the results

results.download_url points at a CSV. Two expirations apply:

Field	What expires	Lifetime
`expires_at`	The download link	About an hour after it is issued
`file_expires_at`	The result file	7 days after the job finishes

An expired link is not a problem — call the status endpoint again and it returns a freshly signed download_url. Once file_expires_at passes, though, the results are gone: download anything you want to keep within 7 days.

curl --location --output firmographics.csv "PASTE_DOWNLOAD_URL"

csv_bytes = requests.get(results["download_url"]).content
with open("firmographics.csv", "wb") as f:
    f.write(csv_bytes)
print(f"wrote {len(csv_bytes):,} bytes to firmographics.csv")

Complete script

Everything above, end to end:

async_enrich.py

import time
import requests

BASE = "https://api.explorium.ai"
API_KEY = "YOUR_API_KEY"
HEADERS = {"api_key": API_KEY}

CSV_PATH = "sample-business-ids.csv"
ENRICHMENT = "firmographics"      # any business enrichment
POLL_SECONDS = 30


def upload(path: str) -> str:
    """Upload a CSV of entity IDs and return its list_id."""
    with open(path, "rb") as f:
        r = requests.post(
            f"{BASE}/v2/async/entity-id-datasets/upload",
            headers=HEADERS,
            files={"file": (path, f, "text/csv")},
            data={
                "entity_type": "business",
                "name": "async guide demo",
                "description": "Business IDs uploaded by the end-to-end guide",
            },
        )
    r.raise_for_status()
    d = r.json()
    print(f"uploaded {d['rowCount']} rows -> {d['list_id']}")
    return d["list_id"]


def submit(list_id: str, enrichment: str) -> str:
    """Submit an async enrichment job and return its job_id."""
    r = requests.post(
        f"{BASE}/v2/businesses/{enrichment}/job",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"list_id": list_id},
    )
    r.raise_for_status()
    job_id = r.json()["job_id"]
    print(f"submitted job {job_id}")
    return job_id


def wait_for(job_id: str) -> dict:
    """Poll until the job produces results or fails."""
    while True:
        r = requests.get(f"{BASE}/v2/jobs/status/{job_id}", headers=HEADERS)
        r.raise_for_status()
        job = r.json()

        if job.get("error"):
            raise RuntimeError(f"job failed: {job['error']}")
        if job.get("results", {}).get("download_url"):
            return job

        print(f"  status={job['status']} — checking again in {POLL_SECONDS}s")
        time.sleep(POLL_SECONDS)


def download(job: dict, out_path: str) -> None:
    content = requests.get(job["results"]["download_url"]).content
    with open(out_path, "wb") as f:
        f.write(content)
    print(f"wrote {len(content):,} bytes to {out_path}")


if __name__ == "__main__":
    list_id = upload(CSV_PATH)
    job_id = submit(list_id, ENRICHMENT)
    job = wait_for(job_id)

    usage = job.get("credit_usage", {})
    print(f"{usage.get('total_results')} results for {usage.get('total_credits')} credits")
    download(job, f"{ENRICHMENT}.csv")

Cancelling a job

If you submitted the wrong list or no longer need a run, cancel it rather than letting it finish and consume credits:

cURL

curl --request POST \
  --url https://api.explorium.ai/v2/jobs/cancel/JOB_ID \
  --header 'api_key: YOUR_API_KEY'

Limits and gotchas


Max rows per dataset	10,000
Max job run time	24 hours
Result format	CSV, via a signed URL
Download link lifetime	~1 hour — re-poll job status for a fresh link
Result file retention	7 days after the job finishes (`file_expires_at`)
Dataset lifetime	Bounded — see `expiresAt` on upload

Async never takes inline IDs. If you send business_ids to a /job endpoint it will be rejected; upload first and send list_id.
Match entity_type to the endpoint. A business dataset belongs to /v2/businesses/..., a prospect dataset to /v2/prospects/....
Poll on an interval, not in a tight loop — jobs can run for a long time.
Reuse one dataset across enrichments. Upload once, then submit several jobs against the same list_id.

Async jobs

Concepts and endpoint reference

Upload dataset

Full request and response schema

Get job status

Status fields and playground

Business enrichments

Every enrichment and its async variant

Getting started

Businesses

Business enrichments

Prospects

Prospect enrichments

Research (GenAI)

Async jobs

Autocomplete

Credits

Events & Webhooks

Running an async enrichment end to end

When to use async

The flow

Step 1 — Prepare your CSV

sample-business-ids.csv

sample-prospect-ids.csv

Step 2 — Upload the dataset

Step 3 — Submit the job

Step 4 — Poll for status

Step 5 — Download the results

Complete script

Cancelling a job

Limits and gotchas

Async jobs

Upload dataset

Get job status

Business enrichments

​When to use async

​The flow

​Step 1 — Prepare your CSV

sample-business-ids.csv

sample-prospect-ids.csv

​Step 2 — Upload the dataset

​Step 3 — Submit the job

​Step 4 — Poll for status

​Step 5 — Download the results

​Complete script

​Cancelling a job

​Limits and gotchas

​Related

Async jobs

Upload dataset

Get job status

Business enrichments

When to use async

The flow

Step 1 — Prepare your CSV

Step 2 — Upload the dataset

Step 3 — Submit the job

Step 4 — Poll for status

Step 5 — Download the results

Complete script

Cancelling a job

Limits and gotchas

Related