# Python SDK

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

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

The wheel is at [`/dl/cestra-0.1.0-py3-none-any.whl`](https://docs.cestra.ai/dl/cestra-0.1.0-py3-none-any.whl) if you prefer it.

```python
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

```python
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

```python
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

```python
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:

```python
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

```python
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](/artifacts).

## Files

```python
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

```python
@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](/rest-api#event-types).

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

| Call | Returns |
| --- | --- |
| `send(text, *, notify=None, client_id=None)` | message id |
| `notice(text, tone="info", *, notify=None)` | message id |
| `ask(question, options=(), *, detail, allow_text, expires_in, notify)` | `Ask` |
| `Ask.wait(timeout=None)`, `.refresh()`, `.cancel()` | `Answer` |
| `ask_status(ask_id, wait=0)` | `Answer` |
| `job(title, steps=())` | `Job` |
| `Job.update(pct=, title=, steps=, error=)`, `.step(i, status="done", label=None)`, `.done()`, `.fail(msg)`, `.cancel()` | job state dict |
| `artifact(type, title, data=None, *, html=None, key=None, summary=None)` | `ArtifactRef` |
| `file(path_or_bytes, *, name=None, caption=None, mime=None)` | `FileRef` |
| `download(file_id, path=None)` | bytes or path |
| `on(type, fn=None)`, `run(wait=30)`, `stop()` | |
| `events(after=None, wait=0)`, `ack(seq)` | |
| `me()`, `status("working")`, `disconnect()` | |