Guide

Jobs & streaming

Every batch call is a job. By default the SDK hides that; here is what you get when you take the handle instead.

The lifecycle

queued → running → done, or stopped / failed. Product methods block until done and return the typed result; pass wait=False (Python) or wait: false (Node) to get the Job back immediately:

job = client.leads.rank(customers, offer=offer, wait=False)

job.id            # "lsj_7c2f…"
job.status        # "queued" | "running" | "done" | "stopped" | "failed"
job.refresh()     # one poll
job.wait()        # block until terminal; raises JobFailedError on failure
job.stop()        # idempotent; the job keeps nothing half-written

wait takes on_progress/onProgress (called with each progress block: pct, stage, completed, total, eta_s) and a timeout. Timing out stops your wait, never the job — call wait() again, or stop() it.

Streaming live events

A running job emits the same event feed the EGGai app renders — progress, per-customer live lines, and the running outcome tallies:

for ev in job.stream():
    if ev["type"] == "progress":
        print(f"{ev['pct']}% · {ev['stage']}")
    elif ev["type"] == "live":
        print(ev["who"], "→", ev["text"])   # the twin's words
EventPayload
progresspct, stage, completed, total, elapsed_s, eta_s
liveOne customer being answered right now: who, term, text (the twin's words), ok
twinRunning outcome counts across the panel so far
done · stopped · errorTerminal — the stream ends after one of these

The stream rides the documented /events?after=SEQ polling endpoint, so it survives reconnects and needs no SSE client; raw SSE is also available at /v1/jobs/{id}/stream if you prefer it — see the API reference.

Results

result_rows() / resultRows() pages every row out of a finished job (1,000 per page, handled for you). The typed wrappers accept those rows directly:

from eggai import RankedLeads

job.wait()
ranked = RankedLeads(job.result_rows())

Retries and idempotency

Creating a job is not idempotent by default — a retry after a network blip could start the same campaign twice. Pass an idempotency key and it cannot:

job = client.offers.cheapest_accepted(
    rows=rows, idempotency_key="campaign-2026-08-fresh-picks", wait=False)

The same key on the same product returns the same job. The SDKs already retry reads (and only reads) on 429/5xx with backoff.

Limits