Program AI Automations.

Provides a task-scoped key-value store with get, set, and a required TTL. Holds an expensive result, such as an API pull or a model call, so a later run reuses it until it expires.

rate = qbash.cache.get_or_set(
    "fx-usd-nzd",
    lambda: fetch_rate("USD", "NZD"),
    ttl="1h",
)
qbash.cache.set("last-cursor", cursor, ttl="1d")

Searches and reads the chat threads of the person running the task, and only that person's threads. Shows a consent banner on the run form. Turns prior conversations into task input.

threads = qbash.chats.search("supplier dispute", limit=10)["threads"]
messages = qbash.chats.read(threads[0]["id"], limit=50)

Records that a unit of work finished, under a key you choose. On retry, qbash.checkpoint.done() reports which keys were already marked, so work you guard runs once and its model calls and side effects are not repeated. A Slack post inside that guard does not post twice.

for invoice in invoices:
    if qbash.checkpoint.done(f"invoice-{invoice['id']}"):
        continue
    reconcile(invoice)
    qbash.checkpoint.mark(f"invoice-{invoice['id']}")

Reaches any HTTP API through a protected request helper, so an internal or niche service with no prebuilt connector still works from the same script. Returns status, headers, and body for the next step to use.

res = qbash.http.request(
    "GET",
    "https://api.vertical-tool.example/v2/orders",
    headers={"Authorization": f"Bearer {token}"},
)
orders = res["body"]["data"]

Reads run inputs with qbash.inputs.get, merging the trigger payload with values a person entered. You declare each input with a type, including text, number, date, select, file, and credential, and qbash renders those declarations as a run form, so a teammate reruns the task by filling fields. A credential input lets the runner pick which account the task acts on.

period = qbash.inputs.get("period", default="last-month")
dry_run = qbash.inputs.get("dry_run", default=False)

Fans a function out over a list and collects the results. Takes concurrency, a per-minute rate limit, retries, and an optional checkpoint name as arguments, so fan-out over hundreds of items stays a single call. Pass checkpoint= and each completed item is recorded and skipped on retry.

results = qbash.parallel(
    fetch_statement,
    accounts,
    concurrency=5,
    checkpoint="statements",
    continue_on_error=True,
)

Fills {{placeholders}} in a PPTX file and preserves the original formatting, so a recurring deck generates from the template the team already approved. Placeholder keys are word characters only.

deck = qbash.powerpoint.fill(
    template_bytes=qbash.files.read(template, encoding="bytes"),
    data={"client": client, "arr": arr},
)

Exposes the execution id for correlating against logs, the task author, the person who pressed Run, the company and project, timestamps, and a test-run flag. qbash.run.last_run_at returns the previous successful run’s time, and None on the first run, so "process everything since last run" is one property.

since = qbash.run.last_run_at        # None on the first run
if qbash.run.is_test:
    accounts = accounts[:1]