Python SDK

pip install cueapi-sdk

from cueapi import CueAPI
 
client = CueAPI("cue_sk_your_key")
 
cue = client.cues.create(
    name="daily-report",
    cron="0 9 * * *",
    callback="https://my-app.com/webhook",
)
 
print(cue.id, cue.next_run_at)

with CueAPI("cue_sk_your_key") as client:
    for cue in client.cues.list().cues:
        print(cue.name)

# Recurring
cue = client.cues.create(
    name="daily-sync",
    cron="0 9 * * *",
    timezone="America/Los_Angeles",
    callback="https://api.example.com/sync",
)
 
# One-time
cue = client.cues.create(
    name="reminder",
    at="2026-05-01T14:00:00Z",
    callback="https://api.example.com/notify",
)
 
# Worker transport (no callback URL needed)
cue = client.cues.create(
    name="agent-task",
    cron="0 * * * *",
    transport="worker",
    payload={"task": "summarize-inbox"},
)
 
# With on-failure escalation
cue = client.cues.create(
    name="critical-job",
    cron="*/5 * * * *",
    callback="https://api.example.com/run",
    on_failure={"after_attempts": 3, "notify": "[email protected]"},
)

client.executions.report_outcome(
    execution_id="exec_a1b2c3",
    success=True,
    summary="Generated 47 leads",
    external_id="lead-batch-2026-04-19",  # link to your system
    result_ref="batch-id:7842",            # short opaque reference
    result_url="https://my-app.com/batches/7842",
    result_type="lead-batch",
    artifacts=[{"name": "leads.csv", "url": "https://..."}],
    metadata={"agent": "lead-finder-v3"},
)

with client.executions.handle(exec_id) as ctx:
    result = do_work(ctx.payload)
    ctx.result = f"Processed {result.count} records"
    ctx.evidence = {"external_id": result.batch_id, "result_url": result.url}
# Clean exit -> POST success=True with result + evidence
# Exception  -> POST success=False with stringified exception

from cueapi import verify_webhook
 
# In your webhook handler
is_valid = verify_webhook(
    payload=request.body,
    signature=request.headers["X-CueAPI-Signature"],
    timestamp=request.headers["X-CueAPI-Timestamp"],
    secret="whsec_your_secret",  # from /v1/auth/webhook-secret
    tolerance=300,  # max age in seconds
)
 
if not is_valid:
    return Response(status=401)

from cueapi import CueAPI
from cueapi.exceptions import RateLimitError, CueNotFoundError
 
client = CueAPI("cue_sk_your_key")
 
try:
    client.cues.get("cue_does_not_exist")
except CueNotFoundError:
    print("not found")
except RateLimitError as e:
    print(f"slow down — retry after {e.retry_after}s")