Python SDK

The cestra package needs Python 3.9 or newer and only uses the standard library. It's fully typed.

pip install https://docs.cestra.ai/dl/cestra.tar.gz

The wheel is at /dl/cestra-0.1.0-py3-none-any.whl if you prefer it.

import os
from cestra import Agent

agent = Agent(key=os.environ["CESTRA_KEY"])
agent.send("Hello from Cestra")

Agent() with no arguments reads CESTRA_KEY. base_url defaults to $CESTRA_URL, then https://api.cestra.ai. Other options are timeout=30 (seconds per request) and max_retries=4.

#Messages and notices

agent.send("Build **passed** in 3m 12s")       # Markdown; returns the message id
agent.notice("Backup done", tone="success")    # info | success | warning | error
agent.send("Quiet update", notify=False)       # no push notification

#Asks

ask = agent.ask(
    "Commit to main?",
    ["Approve", {"label": "Discard", "style": "destructive"}],
    expires_in=3600,      # seconds, 60 .. 30 days
    allow_text=False,     # True adds a text box
    detail="3 files changed",
)
reply = ask.wait(timeout=600)
# Answer(status, answer, option_id, text, answered_by, answered_at, ...)

if reply.status == "answered" and reply.option_id == "o1":
    ...
elif reply.status == "pending":
    ...   # nobody answered within 10 minutes; the question stays open

Options get ids o1, o2, … in order (ask.options lists them). wait() long-polls and never raises on timeout. You get back status="pending". With timeout=None it waits until the ask is answered, expires or is cancelled. Other calls: ask.refresh(), ask.cancel(), agent.ask_status(ask_id, wait=0).

#Jobs

with agent.job("Reviewing spec", steps=["Read", "Compare"]) as job:
    job.update(pct=40)
    job.step(0, "done")       # marks step 0 done and starts step 1
    ...
# Leaving the block calls job.done(). An exception calls job.fail(str(e)) and re-raises.

Without with, call job.done(), job.fail("reason") or job.cancel() yourself. job.update(pct=, title=, steps=, error=) changes the card.

When someone presses Stop in the app, the next job.update() or job.step() raises JobStopped. Inside a with block that exception is swallowed and the block ends. If agent.run() is running, job.stop_requested also turns True as soon as the stop event arrives. A long loop can check it:

with agent.job("Crawling", steps=["Fetch", "Parse"]) as job:
    for i, url in enumerate(urls):
        if job.stop_requested:
            break
        crawl(url)
        job.update(pct=100 * (i + 1) / len(urls))

#Artifacts

ref = agent.artifact(
    type="table",
    title="Coverage matrix",
    key="matrix",     # same key again = version 2, 3, ...
    data={"columns": ["Requirement", "Status"], "rows": [["Login", {"value": "ok", "tone": "ok"}]]},
)
ref.id, ref.version   # ArtifactRef(id, message_id, key, version)

agent.artifact(type="markdown", title="Notes", data={"text": "# Findings\n..."})
agent.artifact(type="html", title="Picker", html="<!doctype html>...")

Schemas for each type are on Artifacts.

#Files

agent.file("report.pdf", caption="Quote")                # path; ≤ 50 MB; returns FileRef
agent.file(b"a,b\n1,2\n", name="data.csv")               # or bytes + name
agent.download(file_id, "downloads/")                    # writes downloads/<original name>, returns the path
data = agent.download(file_id)                           # or get bytes

#Receiving events

@agent.on("message")
def on_message(event):
    if event.get("file"):
        path = agent.download(event["file"]["id"], ".")
        agent.send(f"Got {path}")
    else:
        agent.send(f"You said: {event.text}")

@agent.on("ask.answered")
def on_answer(event):
    print(event["ask_id"], event["answer"], event["by"]["name"])

@agent.on("*")
def log_all(event):
    print(event.seq, event.type, event.data)

agent.run()   # blocks until agent.stop()

event is an Event(seq, type, created_at, data). event["x"] and event.get("x") read from data. The types are message, ask.answered, ask.expired, form.submitted, job.stop_requested and member.joined, and their payloads are listed here.

How run() behaves:

• It long-polls GET /v1/events for 30 seconds at a time and resumes from the server-stored cursor, so a restarted script picks up where it left off.
• Each event is acknowledged only after its handlers return, so delivery is at-least-once. Make handlers safe to run twice if that matters.
• If a handler raises, the exception is logged to the cestra logger and the loop moves on.
• Network errors and 5xx responses back off and retry. A 429 waits for Retry-After.
• A bad or revoked key raises CestraError with status 401.
• agent.stop() works from a handler or another thread, and run() returns within a fraction of a second.

To run it alongside other work, use threading.Thread(target=agent.run, daemon=True).start().

The low-level calls are agent.events(after=None, wait=0), which returns {"events": [...], "cursor": n} (passing after acknowledges up to it), and agent.ack(seq).

#Errors and retries

Every failure raises CestraError(status, code, message). status is 0 for network errors. JobStopped is a subclass.

Each create sends a random client_id, so automatic retries of 429, 5xx and network errors never create duplicates. Pass your own client_id= to make a call idempotent across restarts too. File uploads have no client_id, so they only retry on 429, 502, 503 and 504.

#Reference