Automate data-collection projects without creating a fragile workflow

Common automations

Choose what should happen automatically

For each automation, create one manageable Connect project, save its project ID, and check its current status before moving to the next step.

How systems connect

Save Connect IDs with the related records in your system

Before the examples

Create one authenticated client for every request

Every Connect API request must include your API key in the X-API-KEY header. Create and manage keys from the API Keys page in Connect, then create one authenticated session and reuse it for every call. Each Python example on this page assumes this setup has already run.

import os
import requests

api = requests.Session()
api.headers.update({
    "X-API-KEY": os.environ["CONNECT_API_KEY"],
    "Content-Type": "application/json",
})

Values such as project_payload and process(...) in the examples below are placeholders for your application's own data and code. See the AI data guide for a complete project payload.

Every request that changes data should carry an IDEMPOTENCY-TOKEN header: a unique ID you make up for that one action. If Connect receives the same token again, it does not repeat the action — it returns the result of the first attempt.

That only protects you if the token survives a crash. The example below keeps the token in a small file next to the script, and the file answers one question: did the last attempt get an answer?

import uuid

TOKEN_FILE = "create_project.token"

# A leftover token means the last run never got an answer: retry with it.
if os.path.exists(TOKEN_FILE):
    with open(TOKEN_FILE) as f:
        operation_token = f.read()
else:
    operation_token = str(uuid.uuid4())
    with open(TOKEN_FILE, "w") as f:
        f.write(operation_token)

response = api.post(
    "https://connect-api.cloudresearch.com/api/v1/project",
    headers={"IDEMPOTENCY-TOKEN": operation_token},
    json=project_payload,
    timeout=30,
)

# Save both values before starting the next step.
trace_id = response.headers.get("X-TRACE-ID")
response.raise_for_status()
project_id = response.json()["project"]["projectId"]

# The result is recorded, so the operation is complete. Delete the token
# only now: if the process dies first, rerunning with the saved token
# returns this same stored response instead of creating a second project.
# If the request was rejected, delete the file before retrying with a
# corrected payload - a corrected request is a new action.
os.remove(TOKEN_FILE)

Check periodically rather than continuously. Wait longer between checks, stop after a reasonable deadline, and use each response to decide what your system should do next.

When a list is too large for one response, the API returns a nextToken. Include that token in the next request to get the next page. Continue until the response no longer includes a token.

next_token = None

while True:
    params = {"size": 100}
    if next_token:
        params["nextToken"] = next_token

    response = api.get(
        "https://connect-api.cloudresearch.com/api/v1/project",
        params=params,
    )
    response.raise_for_status()
    page = response.json()

    process(page["projects"])
    next_token = page.get("nextToken")
    if not next_token:
        break

Before you go live

Make the automation easy to monitor and troubleshoot