Skip to content

Build and run a workflow#

This page writes a workflow four ways — the console's box editor, YAML with astra, Python, the API — then runs it, follows its executions and changes it. What the boxes are is in Boxes; every field is in the Workflow specification.

Before you begin#

  • The editor role in the workspace (flows:create, flows:update, flows:trigger); viewers watch executions.
  • What the workflow names, in the workspace: here, an assistant (or a deployment) that is up. Below, <assistant> stands for its name.
  • For the CLI: astra login and astra use <org>/<workspace> (see Install the CLI).
  • For the API: ASTRALYX_API and ASTRALYX_TOKEN as in the API Reference.

The example#

message-triage takes a message sent to its address, has a model summarise it and classify its urgency, and tells you at once when it is urgent — or asks you when the model is not sure. If a model box fails, an error path tells you that too.

flowchart LR
  T([inbox<br/>HTTP call]) --> S[summary<br/>ask a model]
  T --> U[urgency<br/>classify]
  U --> R{route}
  R -- now --> N[tell]
  R -- ask --> Q[unsure]
  S --> N
  S --> Q
  S -. fails .-> E[on-error]
  U -. fails .-> E
message-triage.yaml
apiVersion: anemoi.astralyx/v1
kind: Flow
metadata:
  name: message-triage
spec:
  description: Reads a message, judges its urgency, tells me at once when it is urgent.
  triggers:
    - {id: inbox, kind: http, dedupe: "${{ event.id }}", dedupe_seconds: 600}
  steps:
    - id: summary
      kind: model
      assistant: <assistant>
      prompt: "Summarise this message in one sentence:\n\n${{ event.text }}"
    - id: urgency
      kind: classify
      assistant: <assistant>
      input: "${{ event.text }}"
      labels:
        - {name: low, description: can wait days}
        - {name: normal, description: today}
        - {name: urgent, description: now}
    - id: route
      kind: switch
      needs: [urgency]
      cases:
        - {name: now, when: "steps.urgency.outputs.label == 'urgent'"}
        - {name: ask, when: "steps.urgency.outputs.label == 'uncertain'"}
      default: later
    - id: tell
      kind: notify
      needs: [route, summary]
      when: steps.route.outputs.branch == 'now'
      title: "Urgent: ${{ steps.summary.outputs.text }}"
      message: "${{ steps.urgency.outputs.reason }}"
    - id: unsure
      kind: notify
      needs: [route, summary]
      when: steps.route.outputs.branch == 'ask'
      title: "Not sure how urgent: ${{ steps.summary.outputs.text }}"
      message: "The model guessed ${{ steps.urgency.outputs.guess }} (${{ steps.urgency.outputs.confidence }})."
    - id: on-error
      kind: notify
      needs: [summary, urgency]
      runs: failure
      title: message-triage could not read a message
      message: "Open ${{ execution.name }} to see why."
  outputs:
    urgency: "${{ steps.urgency.outputs.label }}"
    summary: "${{ steps.summary.outputs.text }}"

What to notice: summary and urgency need nothing, so they run side by side. route is a condition; tell and unsure each take one of its ways, and a message of low or normal urgency takes later, where no box is: both are skipped, and the execution succeeds. on-error runs only when summary or urgency fails for good — and the execution still fails.

Write the workflow#

  1. Open Iris → Workflows and press New workflow (or, under Start from a recipe, Use it on one).
  2. Name: message-triage.
  3. On Boxes, the palette groups the boxes: Start (HTTP call, Schedule), Light (Ask a model, Classify, Function, Notify), Control (Condition, Error path, Approval, Wait, Wait for event) and Cluster (Run, Agent, Sub-workflow). Drag a box onto the graph: dropped on a box, it comes after that box; dropped elsewhere, it stands alone. A click adds it after the selected box.
  4. Connect boxes: drag from a box's dot to the next box (or click the dot, then the box). From a condition, the connection takes the condition's next free way — the box's Only when is set to steps.route.outputs.branch == '<way>' for you. Click an arrow to choose its way (Along the way), make it an error path (Runs: When it fails), or Remove the connection.
  5. Click a box to fill it in on the right: Who answers, the prompt, the labels, the ways, a Recipe to start from; then Needs, Only when, Runs and Retries, time limit, repeat, common to every box.
  6. Inputs, outputs & limits holds the workflow's inputs, its outputs, When a step fails (stop or continue), its time limit, its drive, and At most at once.
  7. YAML shows the same workflow as a file: an edit on either side shows on the other. You can paste the file above there (and replace <assistant>) instead of steps 3 to 6.
  8. Under the editor, the checks — the console's own, then the cluster's — list problems as you go; click one to open its box. Press Create the workflow.
$ astra iris workflows validate -f message-triage.yaml
message-triage.yaml: flow message-triage is valid
$ astra iris workflows apply -f message-triage.yaml
flow message-triage: version 1 written, now current

validate runs the cluster's own checks on the file, here, and prints their warnings; --remote also sends it to the cluster's check. Neither looks for what the workflow names (the assistant, agents, functions, drives): one that does not exist fails its box when it runs. apply creates the workflow, or writes its next version (--no-current keeps the current one). A file may also be JSON.

The astralyx package (Python 3.9 or newer, standard library only). It is not published on PyPI yet.

message_triage.py
from astralyx.iris import Client, Workflow

ASSISTANT = "<assistant>"

wf = Workflow("message-triage", description="Reads a message, judges its urgency, tells me at once when it is urgent.")
wf.http_trigger("inbox", dedupe="${{ event.id }}", dedupe_seconds=600)
summary = wf.model("summary", assistant=ASSISTANT, prompt="Summarise this message in one sentence:\n\n${{ event.text }}")
urgency = wf.classify("urgency", assistant=ASSISTANT, input="${{ event.text }}",
                      labels={"low": "can wait days", "normal": "today", "urgent": "now"})
route = wf.switch("route", needs=[urgency], default="later",
                  cases={"now": f"{urgency.outputs.label} == 'urgent'", "ask": f"{urgency.outputs.label} == 'uncertain'"})
wf.notify("tell", needs=[route, summary], when=f"{route.outputs.branch} == 'now'",
          title=f"Urgent: {summary.output('text')}", message=urgency.output("reason"))
wf.notify("unsure", needs=[route, summary], when=f"{route.outputs.branch} == 'ask'",
          title=f"Not sure how urgent: {summary.output('text')}",
          message=f"The model guessed {urgency.output('guess')} ({urgency.output('confidence')}).")
wf.notify("on-error", needs=[summary, urgency], runs="failure",
          title="message-triage could not read a message", message="Open ${{ execution.name }} to see why.")
wf.output("urgency", urgency.output("label"))
wf.output("summary", summary.output("text"))

assert wf.validate() == [], wf.validate()
c = Client.from_config()          # what `astra login` and `astra use` saved
print(c.flows.check(wf))           # the cluster's checks: {ok, errors, warnings}
c.flows.apply(wf)                  # a new workflow, or its next version

step.output(key) is ${{ steps.<id>.outputs.<key> }}, for templates; step.outputs.<key> and step.state are the bare names, for when and until. wf.to_json() is the same document as the YAML file. Client.from_config() reads ~/.config/astra/config.json and the ASTRA_URL, ASTRA_TOKEN, ASTRA_ORG, ASTRA_WORKSPACE and ASTRA_CLUSTER variables. The builder also has run, agent, approval, wait_event, sleep, subflow, function, schedule and limit (concurrency).

$ yq -o=json '{"metadata": .metadata, "spec": .spec}' message-triage.yaml \
    | curl -sS -X POST "$ASTRALYX_API/anemoi/flows" -H "Authorization: Bearer $ASTRALYX_TOKEN" \
        -H "Content-Type: application/json" -d @-

POST $ASTRALYX_API/anemoi/flow-checks with {"name", "spec"} checks without writing: {"ok", "errors": [{"path", "message"}], "warnings": [...]}. A new version is PUT $ASTRALYX_API/anemoi/flows/{name} with {"spec", "current"}.

The checks refuse what cannot work when the workflow is written, not when it runs: a box that reads a box it does not need, an input that is not declared, a condition with two ways of the same name, a reference to another workspace. Warnings (an approval with no approvers, a wait with no time limit) do not stop it.

Run it#

Its trigger runs it when its address is called (see Start a workflow with an HTTP call). By hand:

On the workflow's page press Run now, fill in its inputs (and a Version other than the current one, if you like) and press Run. The execution's page opens. A workflow that reads event also offers Event (JSON, optional): what a trigger would send, to try it by hand.

$ astra iris workflows run message-triage --wait \
    --event '{"id": "m1", "text": "The production database is down and customers cannot pay."}'
execution message-triage-5b2e90 of flow message-triage started

--event is JSON, or @file.json; --input key=value (repeatable) gives inputs; --version N another version; --wait follows the execution until it ends and prints its boxes and outputs (exit status 1 unless it Succeeded).

e = c.flows.run("message-triage", event={"id": "m1", "text": "The production database is down and customers cannot pay."})
done = c.executions.wait(e["metadata"]["name"], timeout=600)
print(done["status"]["state"], done.get("outputs"))

run(..., wait_seconds=60) answers once the execution has ended (at most 90 s).

$ curl -sS -X POST "$ASTRALYX_API/anemoi/flows/message-triage/executions" \
    -H "Authorization: Bearer $ASTRALYX_TOKEN" -H "Content-Type: application/json" \
    -d '{"event": {"id": "m1", "text": "The production database is down and customers cannot pay."}, "wait_seconds": 60}' \
    | jq '{state: .status.state, outputs}'

{inputs, version, event, wait_seconds}, each optional. Without wait_seconds the answer is 201 at once; with it (1 to 90), 200 once the execution has ended, or 202 when it has not by then.

A run by hand runs as you, and applies none of a trigger's repeats or calls a minute.

Follow an execution#

The execution's page draws the graph live: each box's state (pending, running, waiting, succeeded, failed, skipped), its attempts, their reasons (Asking …, Waking deployment …, Took now, Notified), links to their runs and approvals, the machine, the outputs; then the execution's Trigger, Inputs (with its event), Outputs, the Receipts of its agent runs and the Events it received. Click a box to see only it.

$ astra iris workflows executions --flow message-triage
$ astra iris workflows get message-triage-5b2e90
$ astra iris workflows get message-triage-5b2e90 --json

get prints the state, then STEP STATE ATTEMPTS LAST MACHINE WHY OUTPUTS, the receipts of its agent runs, and its outputs.

$ curl -sS "$ASTRALYX_API/anemoi/flow-executions?flow=message-triage&state=Running" -H "Authorization: Bearer $ASTRALYX_TOKEN"
$ curl -sS "$ASTRALYX_API/anemoi/flow-executions/message-triage-5b2e90?wait=30" -H "Authorization: Bearer $ASTRALYX_TOKEN" \
    | jq '{state: .status.state, steps: (.steps | map_values({state, reason}))}'

?wait=<seconds> (at most 30) answers once the execution has ended, or after that long: a long poll.

Send an event, retry a box, cancel#

An event goes to a wait_event box (the examples send deployed to one so named; message-triage has none); a retry, to a box that failed.

On the execution's page: Send event on a waiting wait_event box (its name and key=value data); Retry step on a failed box (a new attempt); Cancel stops the boxes under way.

$ astra iris workflows events send <execution> deployed env=prod sha=4be1c2
$ astra iris workflows retry <execution> summary
$ astra iris workflows cancel <execution>

c.executions.send_event("<execution>", "deployed", {"env": "prod"})
c.executions.retry("<execution>", "summary")
c.executions.cancel("<execution>")

$ curl -sS -X POST "$ASTRALYX_API/anemoi/flow-executions/<execution>/events/deployed" -H "Authorization: Bearer $ASTRALYX_TOKEN" \
    -H "Content-Type: application/json" -d '{"data": {"env": "prod"}}'
$ curl -sS -X POST "$ASTRALYX_API/anemoi/flow-executions/<execution>/steps/summary/retry" -H "Authorization: Bearer $ASTRALYX_TOKEN"
$ curl -sS -X POST "$ASTRALYX_API/anemoi/flow-executions/<execution>/cancel" -H "Authorization: Bearer $ASTRALYX_TOKEN"

An event's data is key=value text, at most 4 KiB. A retried approval or event box asks again. DELETE $ASTRALYX_API/anemoi/flow-executions/{name} removes an execution that ended, with its own drive.

Change a workflow#

Edit (new version) on the workflow's page opens the editor with the current version; tick Runs (and its triggers) use this version from now on or not, and press Write the version. Versions lists them, with Make current.

astra iris workflows apply -f again writes the next version; astra iris workflows list shows FLOW CURRENT LATEST AGE.

$ curl -sS -X PUT "$ASTRALYX_API/anemoi/flows/message-triage/current" -H "Authorization: Bearer $ASTRALYX_TOKEN" \
    -H "Content-Type: application/json" -d '{"version": 1}'

Executions keep the version they started with. The current version's triggers are the ones that fire. Delete on the workflow's page removes it, its versions and its triggers' state; its executions stay, and its addresses answer 404 HOOK_FLOW_GONE.