Metadata-Version: 2.4
Name: cestra
Version: 0.1.0
Summary: Give your own AI agents a screen on the web: messages, questions with buttons, progress, artifacts and files.
Project-URL: Homepage, https://cestra.ai
Project-URL: Documentation, https://docs.cestra.ai/python
Author: Cestra
License: MIT
Keywords: agents,ai,cestra,human-in-the-loop,notifications
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Typing :: Typed
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# cestra (Python)

Give your own AI agent a screen on the web. Your script sends messages, questions with buttons, progress and artifacts; people answer at [app.cestra.ai](https://app.cestra.ai) and the answers come back to your code.

Python 3.9+, standard library only.

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

Get a key at app.cestra.ai → Developer → New agent. It starts with `cs_live_` and is shown once.

```python
import os
from cestra import Agent

agent = Agent(key=os.environ["CESTRA_KEY"])   # or just Agent() to read $CESTRA_KEY
agent.send("Hello from Cestra")
agent.notice("Backup done", tone="success")   # info | success | warning | error

ask = agent.ask("Commit to main?", ["Approve", {"label": "Discard", "style": "destructive"}], expires_in=3600)
reply = ask.wait(timeout=600)                 # Answer(status, answer, option_id, text, answered_by, ...)
if reply.status == "answered" and reply.option_id == "o1":
    ...                                       # options get ids o1, o2, ... in order

with agent.job("Reviewing spec", steps=["Read", "Compare"]) as job:
    job.update(pct=40)
    job.step(0, "done")                       # also starts step 1
# leaving the block marks the job done; an exception marks it failed

agent.artifact(type="table", title="Matrix", key="matrix",
               data={"columns": ["Req", "Status"], "rows": [["Login", "ok"]]})
agent.file("report.pdf", caption="Quote")
```

## Receiving events

```python
@agent.on("message")
def on_message(event):
    if event.get("file"):
        agent.download(event["file"]["id"], ".")   # writes ./<original name>
    else:
        agent.send(f"You said: {event.text}")

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

agent.run()   # blocks; call agent.stop() from a handler or another thread
```

Event types: `message`, `ask.answered`, `ask.expired`, `form.submitted`, `job.stop_requested`, `member.joined`, or `"*"` for all.

`run()` long-polls `GET /v1/events`. An event is acknowledged only after its handlers return, so delivery is at-least-once. If a handler raises, the error is logged and the loop moves on. Network errors and 5xx responses back off and retry, and 429 waits for `Retry-After`. A bad or revoked key raises `CestraError` (status 401).

## Reference

| Call | Returns |
| --- | --- |
| `Agent(key=None, *, base_url=None, timeout=30, max_retries=4)` | `base_url` defaults to `$CESTRA_URL`, then `https://api.cestra.ai` |
| `send(text, *, notify=None, client_id=None)` | message id |
| `notice(text, tone="info")` | message id |
| `ask(question, options=(), *, detail=None, allow_text=False, expires_in=None)` | `Ask` (`.id`, `.options`, `.wait(timeout)`, `.refresh()`, `.cancel()`) |
| `ask_status(ask_id, wait=0)` | `Answer` |
| `job(title, steps=())` | `Job` (`.update(pct=, title=, steps=, error=)`, `.step(i, status)`, `.done()`, `.fail(msg)`, `.cancel()`, `.stop_requested`) |
| `artifact(type, title, data=None, *, html=None, key=None, summary=None)` | `ArtifactRef(id, message_id, key, version)` |
| `file(path_or_bytes, *, name=None, caption=None, mime=None)` | `FileRef(id, message_id, name, size, mime, sha256)` |
| `download(file_id, path=None)` | bytes, or the path written |
| `on(type, fn=None)` / `run(wait=30)` / `stop()` | |
| `events(after=None, wait=0)` / `ack(seq)` | low-level event access |
| `me()`, `status("working" \| "online")`, `disconnect()` | |

Every create sends a random `client_id`, so automatic retries never create duplicates. Pass your own `client_id=` to make a call idempotent across restarts. File uploads have no `client_id`, so they only retry on 429/502/503/504.

When someone presses Stop on a job, the next `job.update()` raises `JobStopped`. Inside a `with agent.job(...)` block that exception is swallowed, so the block just ends.

Errors are `CestraError(status, code, message)`. `status` is 0 for network failures.

Full docs: <https://docs.cestra.ai/python>
