Flow language

Flow describes a composition of actions as an action. A document can be loaded with application code or received and checked at runtime, then resolved against the host's current action registry. This makes the composition dynamic without giving the document a way to import code or call capabilities outside that registry.

This page is the language reference and Python API. Start with Compose actions through streamed data for a guided example. a11.flow.REFERENCE provides a compact version suitable for model prompts.

Find a topic

One flow, read from the top

flow research {
  describe "Search, read the best hits, and answer from them."

  in  question: string required
  out answer:   string
  out sources:  string stream

  header "x-a11-deadline" as deadline

  search = run web-search(query: question, limit: 3)

  brief = run llm-summarize(question: question)
      with "x-a11-deadline": deadline

  nodes fetched {
    for hit in search.hits parallel 2 {
      page = run web-fetch(url: hit.url)
      hit.url -> sources
      page.text | truncate 200 -> brief.pages
      skip page.bytes
    }
  }

  brief.summary -> answer
  skip search.debug
}

x = run action(port: source) runs a local action and binds its ports to x. x = call action(...) dispatches an action on the flow's attached stream. source -> destination pipes a stream into a node.

brief starts before pages arrive. The loop feeds its streaming pages port, which A11 closes after the loop's last writer finishes. Steps run concurrently; use after, wait, or drain to require an order.

A port carries one value unless declared stream, and is optional unless declared required. Keywords accept lower or upper case, such as for and FOR. Mixed-case words are identifiers.

What a port holds

in  question: string required
in  frames:   list[a11.NodeFragment] stream
out audio:    a11.sdk.AudioBuffer stream
out raw:      "application/x-msgpack"

Besides the built-in names — string, text, number, integer, bool, object, json, list, bytes, and any — a type may use an unquoted serialisation tag such as a11.sdk.AudioBuffer. Dotted names are tags and are preserved without importing the module that defines the type. Brackets specify container contents, as in list[string] and list[a11.NodeFragment]. A quoted type is a media type that describes the representation.

Descriptions

Descriptions help callers, including models, decide when to use a flow. They may follow a declaration or use a multiline string:

flow research {
  describe """
    Search the web, read the best hits, and answer from them.

      Costs one search and up to `budget` fetches.
    """

  in  question: string required
    "What to find out — as long as this needs to be, on its own line."
}

A """ string preserves line breaks after dedenting. Dedenting removes a blank first line, a whitespace-only final line and its preceding break, and the common indentation from all remaining lines. Additional indentation on individual lines remains. Escapes match ordinary strings.

A description may also appear alone on the line below a port, header, or describe declaration. A string followed by another token, such as "a literal" -> out, remains a statement.

a11 flow fmt indents a description under its declaration and lines up the columns of a run of declarations around it.

Making a value of a type

A flow can construct a registered type such as Interaction or AudioBuffer from fields with TYPE{...}:

a11.sdk.Interaction{
  role: "user",
  content: [to_chunk({"role": "user", "content": [{"type": "text", "text": said}]})]
}

EXPR as TYPE performs the same conversion and supports generic types such as pieces as list[string]. Both forms validate the value, apply defaults, and report incompatible fields. to_chunk and from_chunk are the two builtins that make and read a Chunk, which is what a content-bearing type is made of.

Which types exist is the host's decision, not the flow's: a tag resolves against the serialisation registries of the process the flow runs in, and a flow cannot import anything. TYPE{...} is unavailable where a { would open a block instead — an if condition, a for's source — so if step.next.done { keeps reading the way it always has; wrap it in brackets if you really need one there.

Action composition

Running a step, and calling one

run some-action(...) executes a handler registered in the local process. call some-action(...) dispatches the action on the flow's attached stream. Choose the verb explicitly:

search = run web-search(query: question)      # ours, here
llm    = call interact_with_llm(...)          # theirs, over there

run requires a local handler. call requires only a local schema for resolution and always dispatches to the peer, even when a local handler has the same name.

The deployment determines which verb is available. flow_actions reports this through each entry's runnable field.

try goes in front of either: try run, try call.

Either verb may also name another flow of the same file, with nothing registered for it:

flow ask-twice {
  in  question: string
  out answers:  string stream
  first  = run ask(question: question)   # `ask` is declared below
  second = run ask(question: question)
  first.answer then second.answer -> answers
}

flow ask {
  in  question: string
  out answer:   string stream
  said = run answer-question(question: question)
  said.text -> answer
}

A program is a set of flows, so declaration order does not affect execution. Compilation checks calls between flows against their declared ports. This keeps reusable compositions in one source file with a single entry point for flow_run and the gateway.

Discarding stream values

An undrained output port stalls its producer. skip page.bytes consumes one value without retaining it. The runtime drains declared outputs that the flow does not reference.

skip n port is a different statement wearing the same word. A Flow stream fans out — every reader sees all of it — so | drop 1 trims only the one reader that says it. A count on skip takes the values off the node itself, before the fan-out, so every reader starts after them:

rows = run read-csv(path: path)
skip 1 rows.lines            # discard the header line
rows.lines | count -> data-rows
rows.lines -> passed-through # both readers start at the second line

Several of them naming one node add up — skip 1 x and skip 2 x leave three values unread, in either order, because the count belongs to the node and is summed while the flow is compiled. It takes a port or a node, not a pipeline: there is no front to take values off a thing each reader derives for itself.

-> _ is the third of these, and the one that does the work. skip says the values were never wanted, and a counted one is taken off the stream before anybody sees it; _ says the result is not wanted, and the pipeline that produced it still runs:

pages | map summarise(it) | logf info "summarised %s" it.url -> _

Every page is summarised and every line is logged, but the result is discarded. _ is valid only as a destination; it cannot be bound or read. _ = node(), _ | count -> n, drain _ and in _: string are each refused while the flow is compiled. It may stand beside a real destination (a -> b, _), where it adds a reader that discards its values.

Putting a stream back together

| group EXPR is batch with a question instead of a count: values gather into a list, and the list closes when the expression holds of the value just added. For example, it can assemble partial utterances into sentences:

pieces | group ends-with(trim(it), [".", "?", "!"]) | map trim(join(it, " "))

Any partial final group is emitted when the stream ends.

| then SOURCE is the other direction: this stream, and then that one.

history then asked -> llm.interactions

then and where may omit the pipe: history then asked and hits where it.ok. Every other stage requires |, which distinguishes stage names from identically named ports.

| flatten is batch backwards: a stream of lists becomes a stream of what they held.

pages | map it.lines | flatten -> lines

Lists are expanded and other values pass through unchanged, so flatten also accepts a mixed stream.

| window N is batch with the lists overlapping: one list of the last N values per value, once N have arrived.

lines | window 2 | where contains(join(it, "\n"), needle) -> hits

Unlike batch, a window can detect patterns that span arbitrary batch boundaries. It retains at most N values, so memory use remains bounded for an unending stream. A stream shorter than N produces no window, while batch may emit a shorter final list.

interleave(a, b, ...) is the other kind of fan-in. Where zip reads its sources in step and gives a tuple per round, this reads them at once and gives each value as it arrives, so a fast stream is not held behind a slow one:

interleave(llm.text_output, tool.progress) -> shown

Values retain their arrival order across sources. The combined stream ends when every source ends; a source failure ends it with that status.

Reducing a stream to one value

Arithmetic reducers calculate one result from a complete stream:

orders | sum it.price -> revenue
runs   | avg it.elapsed -> typical
hits   | max it.score -> best

sum, min, max, and avg read the whole stream and yield one value. With no expression they use each value directly; | sum it.price is equivalent to | map it.price | sum. Durations add and average as durations. For an empty stream, min, max, and avg emit no value, while sum emits 0.

| fold is the general form, for the shape none of those is:

orders | fold 0 as total, total + it.price -> revenue

The name is bound to the previous accumulated value and it to the current input. The starting value is a literal, not an expression: otherwise fold 0 as total could be parsed as a cast of 0 to a type called total. A record literal is allowed because its braces remove this ambiguity.

Carrying state along a stream

| scan is written exactly as fold is, and the difference is where the values go: fold yields one when the stream ends, scan yields one per value as it arrives.

lines | scan 0 as n, n + 1 -> numbered

scan carries state forward for each stream value. repeat also carries state, but rereads its stream from the start on each pass; for reads one value per pass but does not carry state between passes. scan retains one state value, not the complete stream.

The state may also be a record:

lines
  | scan {"inside": false, "line": ""} as s,
      {"inside": starts-with(it, "BEGIN") or (s.inside and not starts-with(it, "END")),
       "line": it}
  | where it.inside and not starts-with(it.line, "BEGIN")
  | map it.line
  -> body

The stage uses constant memory by retaining only one state value.

| sort puts a stream in order:

hits | sort by it.score desc | first 10 -> best

sort buffers the complete stream before emitting values. Comparison follows <; by selects the comparison value, desc reverses the order, and equal values retain their input order.

When a value arrives

Two stages control stream timing.

tokens   | timeout 30s  -> answer
requests | pace 100ms   -> to_api

timeout limits the gap between values. A longer gap ends the flow with deadline_exceeded. Use wait ... timeout to limit an entire step.

pace delays values to enforce a minimum interval without dropping them. The producer blocks when the buffer is full.

Working on several values at once

A per-value stage may say how many values it may have in hand:

urls | map fetch_page(it) parallel 8 -> bodies

Downstream stages still receive values in input order. The parallel stage reorders completed work before emitting it. Add unordered to emit results as soon as they complete:

urls | map fetch_page(it) parallel 8 unordered -> bodies

Use parallel for substantial per-value work such as host round trips, coercions, or large chunks. It adds overhead to simple field access. Stages that gather or order values do not accept parallel.

Text, times, and how long something took

strformat("%s of %s", got, wanted) uses printf conversions: %s for text; %d, %f, and %x for numbers; flags and precision such as %-8s and %06.2f; %2$s for a positional value; and %% for a literal percent. | strformat "fmt" abbreviates | map strformat("fmt", it).

Flow uses printf-style conversions because str.format reads attributes, which could escape sandboxing for untrusted expressions. A printf conversion operates only on supplied values. A conversion with no corresponding value remains unchanged to expose the invalid conversion.

Durations are written the way a timeout is — 500ns, 250ms, 30s, 2m, 1h, and compounded as 1m30s500ms — and are ordinary values. now() is the clock, and the arithmetic is the arithmetic A11's own types allow:

started = node()
now() -> started
work = run slow-thing(input: pages)
done = wait work
let took = now() - started        # instant - instant is a duration
strformat("took %s", took) -> log after done

Steps run concurrently, so source order alone does not delay a clock read. now() -> started needs no barrier because it records the start. A clock read that measures produced work requires after; otherwise the compiler reports flow.barrier.unordered-clock.

An after applies to the complete statement, including arguments. In run act(p: now() - started) after done, the argument is evaluated after done.

+ and - are the only arithmetic the language has, and they exist for this: a composition cannot otherwise say how long it took. A bare number beside a duration counts as seconds; seconds(d) gives the number back. Subtracting in the other order produces a negative duration; it does not use the infinite-timeout convention found elsewhere in A11. - requires spaces because text-upper is an identifier.

Formatting: %s renders a duration as 1m30s and an instant as RFC 3339. A unit in the parenthesised spec gives one number — %(ns)d, %(us)d, %(ms)d, %(s)d, %(m)d, %(h)d — and %(%H:%M:%S)s or %(epoch)d formats an instant.

duration(x) and time(x) are the way back in, and they read exactly what the formatting writes:

deadline = time(header-deadline)          # "2026-08-11T09:14:22Z"
budget   = duration(header-budget)        # "1m30s", or a number of seconds
if now() + budget > deadline { fail deadline_exceeded "not enough time left" }

A timestamp or a timeout that arrived as text — from a header, a JSON field, a model's answer — is a value again, in one call and without a format string to get wrong.

Two statements writing to the same node interleave by arrival. Use then when order matters, such as sending prior conversation turns before the current one.

Reducing data before serialization

| truncate 200 shortens each page before writing it to the summariser. Dropped data is not serialized, sent to a peer, or included in model input. The same applies to | first 3, | where it.ok, | mime "text/*", and | drop 1.

Saying how a value travels

| packb writes a value as application/x-msgpack instead of JSON. Existing MessagePack chunks pass through unchanged, including their type tag. Other representations are re-encoded.

Passing on what the flow was told

Headers carry call metadata such as model selection, identity, and deadlines. Nested actions automatically receive their parent's x-a11- headers. Use forward headers for other headers:

answer = run interact_with_llm(interactions: asked, config: {})
    forward headers "authorization", "x-tenant-*"

Names are forwarded unchanged, and * matches a family of names. Missing optional headers are ignored. Use with "header": expr for computed values. A with value overrides a forwarded header with the same name.

Keeping a step's traffic off the wire

nodes fetched { ... } gives the calls inside it a NodeMap of their own. Their ports are not in the session's node map, so the peer that dispatched the flow neither sees them nor receives their fragments: four fetched pages stay here, one answer goes back. A run step already keeps its nodes off the wire unless it asks for tee; a nodes block is the stronger statement, and it covers call steps too.

Nodes of the flow's own

x = node() creates a stream that several loop passes can write and another step can read. Parentheses distinguish the constructor from an identifier named node.

best = node()

for url in urls {
  page = try run web-fetch(url: url)
  page.text | truncate 120 -> best
}

best | first 1 -> text

The node uses the active node map: the enclosing nodes block's map or the action's map. x = node(existing-id) attaches to an existing node, and x.id passes its identifier to an action that writes to it:

seen = node()
reader = run take-notes(pages: page.text) with "x-a11-progress-node": seen.id
seen -> progress
drain seen after reader          # the flow lent the node; the flow ends it

The final after delays drain until reader finishes writing through seen.id. Without the dependency, the node would close immediately. The compiler reports flow.barrier.wait-lends-node for wait seen in this pattern.

Handle expected failures

A composition that calls four actions will sometimes have one of them fail, and often that is not a reason to abandon the other three. try says so — on either verb — and from there the flow is in charge:

page = try run web-fetch(url: url)
outcome = wait page

if outcome.ok {
  page.text | truncate 120 -> text
} else {
  fail unavailable outcome.message
}

wait holds until its subject is finished — a call, or a node this flow writes -- and bound to a name it is also how the flow reads that outcome, because waiting and finding out are the same moment. status x is the same value where an expression is expected, and drain node is the spelling to use beside the port it is about.

A status is data:

{"ok": false, "code": "NOT_FOUND", "number": 5, "message": "no such page"}

so a flow can branch on it, put it on one of its own outputs, or raise it again. fail takes any of Abseil's canonical codes by name in either case (not_found, NOT_FOUND), a number computed at runtime, or a whole status record — fail outcome re-raises exactly what happened, and fail invalid_argument outcome.message says it again in the caller's terms.

Waiting on something that finished badly ends the flow with that status, unless it was a try: those are the failures the flow said it would handle.

wait first of a, b holds until the first of several calls finishes and leaves the rest running; wait all of a, b holds for every one of them. A race is between calls — a node is finished when whoever writes it says so, which is what wait and drain are for.

A race also produces the zero-based index of the winning call:

won = wait first of primary, backup        # 0 or 1
wait first of primary, backup -> chosen    # ...or straight to a port
let n = wait first of primary, backup      # ...or named

wait all of has no single winner, so it is a barrier only.

A failure one value at a time

A try on a stage is the same idea inside a pipeline: one value the stage cannot do is not a reason to abandon the stream.

docs | try map it as Order -> good

The value is dropped and the failure logged once at warning. Where the failures matter, into sends them somewhere:

docs | try map it as Order into rejected -> good

They arrive as status records — the same shape status x yields — so a stream of failures is an ordinary stream: countable, writable to a port, readable by the caller. Without try, a value a stage cannot do ends the flow, which is the right default for a composition that is not expecting one.

Loops, branches, and state

repeat state = {"round": 0} max 6 {
  step = run triage-step(state: state)
  state <- step.next
  until step.next.confidence >= 0.8

  if step.next.done {
    step.next.verdict -> verdict
  }
}

repeat carries one value from each pass to the next: state starts at the literal and becomes whatever <- names. until (or while) ends the loop, and max bounds it regardless. One of the two is required because repeat has no default bound.

match pulls named fields out of text, as a stage over a stream and as a function over one value: lines | match "name={name} age={age:int}" turns name=Alice age=27 into a record with name and age. Literal text matches itself, a run of spaces or tabs matches any run, and a hole may say what to read itself as (int, number, bool, word, line, rest, duration, time, json). The pattern searches anywhere in the input, so it requires no wildcards. A hole stays on its line unless specified otherwise. The stage drops a value the pattern does not fit and the function answers null. Where the pattern is written out, the fields are known: it.name is completed and a typo is reported.

try also goes in front of a pipe. It converts a source or destination failure into a value instead of ending the flow:

moved = try findings -> seen
status moved | map it.message -> why

Bind it and read it. Unbound, a tolerated pipe that failed leaves its destination closed early and every reader of it sees an ordinary end of stream, with nothing saying why — so the language reports that. This is a different thing from try on a stage: a stage fails once per value and carries on, which is why it has into for the ones it dropped, while a pipe fails once and stops.

A [s =] [try] { ... } block runs its statements as one step. Statements in a flow body run concurrently, while a block groups their outcome. Reading a value blocks only the statements inside the braces. A bound block yields a status like a call. With try, the flow handles a block failure; otherwise the failure ends the flow.

for v in stream runs its block once per value, parallel n runs n passes at a time. A stream read inside a loop or branch is materialised: the runtime buffers it once and replays it to every pass, which is what lets each pass see the same outer value. The buffer grows while it is read, so a pass waits for the value it asks for and not for the stream to finish — a loop reading a stream that is still open is not held up by it, and neither is anything written after the loop.

A loop may be named, and then it reads as its own outcome — the same shape s = try { .. } has:

taken = node()
done = for line in input.lines { line -> taken }
drain taken after done

That last line is what a flow could not say before. The node was already ended when the loop finished — a loop counts as one writer of an outer node for as long as it runs, so the last Release closes it — but nothing in the text said so, and a program whose finished state has to be inferred from writer counting is a program that reads as unfinished. for and repeat also take an after, because a loop is a step like any other.

A for takes until/while too, and it means what it means in a repeat: asked at the tail of a pass, so the body always runs at least once and the value that ended the loop was seen.

for line in input.lines {
  line -> seen
  until line == "quit"
}

That is how a loop over a stream stops before the stream does. It stops reading, exactly as | first n does, and like first n it does not cancel whatever was producing — see below. It cannot be written with parallel: the question is about the pass that just finished, and with several in flight there is no such pass, so which values were seen would depend on scheduling. <- stays a repeat's, because a for takes its value from its stream and has nothing to hand the next pass.

advance is the other way to walk a stream, and it is not a loop: its offset is determined during compilation, so three uses read the first, second, and third values. A name bound outside a loop cannot be advanced inside the loop.

Ending a stream, and ending it badly

drain node writes both of the two facts that end a stream: the node is marked final, so an ordered reader stops, and its writer is closed, so the store admits nothing more. Then it reads what is left, and its name binds the outcome.

abort node is the other ending:

if not status page.ok { abort findings unavailable "the source went away" }

The difference is what a reader is told. Both end the stream; only this one says it went wrong, and without it a stream cut short by something the flow noticed is indistinguishable from one that finished. It takes the code and message a fail takes, and waits for nothing for the same reason, so it belongs in an if or a loop body or carries an after.

Only a node this flow writes can be aborted by it.

Flow unifies node completion into full endings (drain or abort): marking a node final also closes its writer to keep reader semantics consistent.

Ending a step early

cancel x aborts a step, ending the run with status cancelled.

To request graceful completion, send a stop command following standard action conventions:

if tick.number == 3 { {"command": "stop"} -> clock.control_events }

Standard library actions treat {"command": "stop"} on their control port as an end of input, closing their ports normally so downstream readers observe a clean stream termination.

Stage limits (| first n and for with until) stop reading while leaving the upstream producer undisturbed. An active step terminates via its control port, cancel, or an assigned deadline. cancel evaluates immediately, so it belongs within a conditional, loop body, or after clause.

Flow boundaries and sandbox limits

Beyond + and -, Flow provides no arithmetic, function definitions, or direct calls to host code. Expressions read and compare values, access fields with .field and [i], and construct records with built-in functions such as len, lower, join, merge, and default. A flow operates only through declared action streams.

The tables, as data

a11.flow

Flow: a small language for composing A11 actions.

A flow is a composition of existing actions that is itself an action. It declares ports and headers, calls other actions, pipes their streaming ports into one another, loops, branches, and hands its own outputs back. Because it presents an ordinary ActionSchema, anything that can dispatch an action can run one without knowing it is a composition.

Flows are text. That is the point: a gateway, a client, or a model can be handed a composition of actions it has never seen before and run it, with no repository change and no redeploy.

flow shout {
  in  words:   string stream
  out loudest: string

  say = run text-upper(text: words)
  say.upper | first 1 -> loudest
}

Reading a flow

Every statement is one of a handful of shapes, and the whole language fits on a page:

Statement	Means
x = run an-action(port: src)	run an action here, feeding a port
x = call an-action(port: src)	dispatch one on the attached stream
x = node([id]) [in map]	a node of the flow's own, to write and read back
nodes map	declare a node map to keep traffic out of the session
source -> port, port	pipe a stream into one or more node(s)
source \| stage \| stage -> port	reshape it on the way
source \| stage -> _	the same, keeping no result; _ is not a name
skip source	read a stream to its end and discard the values
skip n port	drop a node's first n values, for every reader
s = wait x	hold until x is finished, and say how it went
drain node	end a node: mark it final and close it
abort node [code] [msg]	end a node with a failure, so readers see why
status x	the same outcome, read where a value is expected
[name =] for v in source [parallel n] { }	run a block per value
repeat s = start [max n] { }	repeat a block with carried state
s <- source, until e	what a repeat carries, and when a loop stops
if e { } else { }	run one block or the other
s = try { }	run a block as one step, and say how it went
p = try source -> port	turn a pipe failure into a value
cancel x	ask a called action to stop
fail [code] [message]	end the flow with a status

Every significant word may be written in lower case or upper case — for or FOR, stream or STREAM, not_found or NOT_FOUND. Mixed case is not a keyword, so For is a name.

Steps run concurrently. Statement order does not define execution order. A call starts immediately and receives inputs as they arrive. Use after, wait, or drain when execution order matters.

A11 operations in Flow

Flow provides syntax for these A11 operations:

• run and call, which are two different things. run some-action(...) executes the handler registered where the flow is running; call some-action(...) puts the action on the stream the flow is attached to and lets the peer do it. This matches Action::Run and Action::Call. A composition written against a gateway's actions calls them; one composing actions of its own runs them; a client flow doing retrieval here and inference there does both, in the same flow. run requires a local handler and does not fall back to the session.
• skip, and stages that cut a stream down. skip x.debug reads and discards an output, preventing an undrained output from stalling its producer. skip 1 x.rows takes the first value off the node itself, for every reader of it, which is how a header line stops being everybody's problem — | drop 1 only trims the one reader that says it. Several of them naming the same node add up. -> _ is the complementary form: skip bypasses processing, while _ discards the result after the pipeline runs. pages | map summarise(it) -> _ therefore summarises every page. _ is a destination, not a name, and cannot be read. | first 3, | truncate 4000, | where it.ok and | mime "text/*" throw values away before they reach the next step -- which, when the next step is a model, is the difference between a cheap call and an expensive one. | packb is the other side of the same coin: it says a value travels as MessagePack rather than JSON, and costs nothing when the producer already wrote it that way.
• nodes blocks. Calls inside one get a node map of their own, so their ports are not in the session's map and their fragments are not replicated to the peer that dispatched the flow. A composition that fetches ten pages and sends one summary back should move one summary over the wire, not ten pages. A run step already keeps its nodes off the wire unless it asks for tee.
• wait, status and drain. Completion and having-written are different events in A11, and a composition needs both. Either statement can be bound to a name and named in another statement's after — and a bound wait is also how a flow reads an outcome, because waiting and finding out are the same moment. after also takes a port or a node directly, meaning "once that stream is finished": -> mic.control_events after sentence stops the microphone as soon as there is a sentence, with no barrier to name. It holds the whole statement, arguments included — run act(p: now() - started) after done reads the clock once done has happened — so what a barriered statement reports is what was true by the time it ran. A wait on a node the flow lends rather than writes is the exception that is not a wait at all: it ends the node, and at once, which is why the idiom is drain n after <call>.
• Flow-owned nodes. x = node() makes a stream the flow can write from several places and read back from one; x = node(existing-id) attaches to an existing node, and x.id passes a node identifier to an action that expects to be told where to write. A node lands in the active node map, which is what keeps it off the wire.
• Headers. A11 automatically gives a nested action every x-a11- header of its parent. For other headers, such as an authorization, a tenant id — forward headers "authorization" sends on what the flow was called with, as it arrived, and "x-tenant-*" sends on a family. with "header": expr is the other half, for a value the flow computes; naming both, the with wins, because it is the more specific of the two.

Failures a flow expects

Use try with either verb when the flow will handle a failure:

page = try run web-fetch(url: url)
outcome = wait page                       # wait and read the status
if not outcome.ok {
  fail unavailable outcome.message        # or `fail outcome`, unchanged
}

A status is data — {"ok": .., "code": "NOT_FOUND", "number": 5, "message": ..} — so a flow can branch on it, put it on an output, or raise it again. fail takes any canonical code by name in either case, or a number computed at runtime, or a status record to re-raise as it stands. Waiting on something that finished badly ends the flow with that status, unless it was a try: those are the failures a flow said it would handle.

A file of several flows

A file may declare more than one flow, and a flow may run or call any of the others by name — in whichever order they are written, and with nothing registered for them. This supports extracting reusable or complex sections into named flows while keeping one source document and entry point. Ports are checked between them at compile time, exactly as they are against a registered action, so an incompatible rename is reported before execution.

flow ask {                        # the piece, reusable on its own
  in  question: string
  out answer:   string stream
  said = run answer-question(question: question)
  said.text -> answer
}

flow ask-twice {                  # and a composition of it
  in  question: string
  out answers:  string stream
  first  = run ask(question: question)
  second = run ask(question: question)
  first.answer then second.answer -> answers
}

Grammar

program    := flow+
flow       := "flow" name "{" item* "}"
item       := "describe" description
            | ("in"|"out") name ":" type ["stream"] ["required"] [description]
            | "header" string ["as" name] ["default" literal] [description]
            | statement
statement  := [name "="] call
            | name "=" "node" "(" [expr] ")" ["in" name]
            | [name "="] "wait" reference ["timeout" duration]
            | [name "="] "drain" reference
            | pipeline "->" destination ("," destination)*
            | "skip" (number reference | skip-target ("," skip-target)*)
destination := reference | "_"      # `_` keeps nothing, and is not a name
skip-target := pipeline
            | name ("," name)* "of" name
            | "(" name ("," name)* [ "of" name ] ")"
            | "cancel" name
            | "abort" reference [expr [expr]]
            | "fail" [expr [expr]]
            | "log" [level] expr
            | "logf" [level] string [expr ("," expr)*]
            | "for" name "in" pipeline ["parallel" number] block
            | "repeat" [name "=" expr] ["max" number] block
            | name "<-" pipeline
            | ("until" | "while") expr
            | "if" expr block ["else" (block | if)]
            | "nodes" name [block]
call       := ["try"] ("run" | "call") action "(" [name ":" pipeline, ...] ")"
                  modifier*
modifier   := "tee" | "via" name | "timeout" duration
            | "after" name ("," name)* | "id" expr
            | "with" string ":" expr ("," string ":" expr)*
            | "forward" "headers" string ("," string)*
pipeline   := expr (("|" stage) | bare-stage)*
bare-stage := ("then" source | "where" expr)   # the pipe is optional here
stage      := "first" n | "drop" n | "truncate" n | "batch" n | "window" n
            | "group" expr | "scan" literal "as" name "," expr
            | "match" pattern
            | "then" source | "where" expr | "map" expr | "join" [string]
            | "strformat" string | "mime" string | "collect" | "count"
            | "distinct" | "text" | "json" | "packb"
            | "log" [level] [expr] | "logf" [level] string [expr ("," expr)*]
level      := "debug" | "info" | "warning" | "error" | "critical"
type       := name ("." name)* ["[" type ("," type)* "]"] | string
description := string | newline string   # alone on its line, at any indent
string     := '"' ... '"' | '"""' ... '"""'   # the second may hold line breaks
expr        := literal | name | expr "." name | expr "[" expr "]"
            | builtin "(" expr* ")" | "(" pipeline ")" | "it"
            | "status" reference | name ".id"
            | type "{" [name ":" expr, ...] "}" | expr "as" type
            | expr ("==" | "!=" | "<" | "<=" | ">" | ">=" | "in") expr
            | expr ("+" | "-") expr        # numbers, durations, instants
            | expr ("and" | "or") expr | "not" expr

match pulls named fields out of text, as a stage over a stream and

as a function over one value. Literal text matches itself, a run of spaces or tabs matches any run, and {name} captures up to whatever follows it: lines | match "name={name} age={age:int}" turns name=Alice age=27 into a record with name and age. A hole may say what to read it as: int, number, bool, word, line, rest, duration, time, json; {} captures without a name and is read as it[0]. {{ and }} are literal braces. The pattern searches, so it matches anywhere in the value and needs no leading or trailing wildcards, and a hole stays on its line unless its type says otherwise. The stage drops a value the pattern does not fit, so it is a where and a map at once; the function answers null, which if not obj asks about. Where the pattern is written out rather than computed, the fields it names are known, so it.name is completed and a typo in it is reported. A pattern that cannot be read at all is refused where it is written, because it is a literal almost every time and a silent no-match would hide the typo.

two sources define a value's fields. Missing fields are

reported for both: a port declared with a struct, and a match pattern, whose holes are its fields. Where the file never said -- a port carrying object or json, it without a pattern, or a positional pattern -- nothing is checked, because a value that may hold anything does. One level is checked: a field holding a record of its own says nothing about its keys, so src.meta.title checks meta and stops.

Types are string, text, number, integer, bool, object, json, list, bytes, any, a quoted mimetype, or the tag a serialisation registry knows a type by — a11.sdk.AudioBuffer, written unquoted, and recognised as a tag because it is dotted. A container says what it holds in brackets: list[string], list[a11.NodeFragment]. The type comes first and what the port is like follows it: a port carries one value unless it says stream, and is optional unless it says required. Status codes are Abseil's canonical ones, by name (not_found, NOT_FOUND) or by number. Durations are written 500ns, 250ms, 30s, 5m, 1h, and compound as 1m30s. Comments start with #. The only arithmetic is + and -, which are there for times; there is no way to call out to code, so an expression can read values, compare them, do that arithmetic, take them apart and build new ones, and that is all — which is what makes a flow safe to accept and run.

Prose

A description is prose, and prose does not fit on the line of the declaration it belongs to. Two spellings deal with that, and they compose:

flow documented {
  describe """
    What this flow is for, at the length that actually takes.

      An indented line stays indented, relative to the rest.
    """

  in  question: string required
    "What to find out — as long as it needs to be, on its own line."
}

A """ string may hold line breaks, and its value is dedented: a blank first line goes away, a whitespace-only last line goes away with the break above it, and the indentation every remaining line shares comes off. So a long description sits at the indentation of the flow it describes and still reads as prose. Escapes work as they do in a single-quoted string, and are resolved after the dedent, so a hand-written \\n is a line break and never an indented line.

A description may also stand alone on the line below what it describes, at any indentation or none. That is unambiguous because the string has to be alone: "a literal" -> out is a statement, since something follows the string, and a line holding nothing but a string is not a statement in this language.

A type is also something a value can be made into, which is how a flow feeds a port that wants a real type rather than a bag of keys:

a11.sdk.Interaction{                          # or: {...} as a11.sdk.Interaction
  role: "user",
  content: [to_chunk({"type": "text", "text": said})]
}

Both spellings mean the same thing: take what the expression produced, partial as hand-written things are, and make it that type — filling in what the type defaults and failing where it will not fit. A tag resolves against the serialisation registries of the process the flow runs in, so which types exist is the host's decision: a flow cannot import anything. Tag{...} is not available where a { would open a block instead — an if condition, a for's source — so if step.next.done { keeps meaning what it looks like; brackets lift the restriction, as in if (T{a: 1}).ok {.

Running one

import a11
from a11 import flow

program = flow.loads(source, "shout.flow")
program.register_all(registry)              # now they are actions

result = await program["shout"].invoke(words=["hi", "there"])

a11.flow.plan.FlowPlan.invoke is the convenience path; in a server, register the flows and let the session dispatch them like anything else.

See also a11.flow.plan for the compiled graph, a11.flow.runtime for how it executes, and the REFERENCE constant in this module for a cheat sheet compact enough to put in a prompt.

REFERENCE module-attribute

REFERENCE = 'A11 Flow — a composition of actions that is itself an action.\nEvery keyword may be written in lower case or UPPER CASE, but not Mixed.\n\nflow NAME {\n  describe "what this does"\n  in  PORT: TYPE [stream] [required] "description"     # no `stream` = one value\n  out PORT: TYPE [stream] [required] "description"\n  header "x-header-name" as ALIAS default LITERAL\n\n  X = run some-action(port: SOURCE, ...) MODIFIERS   # a handler registered here\n  X = call some-action(port: SOURCE, ...) MODIFIERS  # on the attached stream\n  let V[, V...] = SOURCE                   # *one* value of that stream, named;\n           # several names take it apart, by field or by position\n  advance V                                # rebind V to the next value of it\n  N = node([ID]) [in MAP]                  # a stream of the flow\'s own\n  nodes MAP [{ ... }]                      # a node map; keeps traffic local\n  SOURCE | STAGE | STAGE -> DEST, DEST     # pipe a stream into node(s)\n  SOURCE | STAGE -> _                      # do the work, keep no result\n  skip SOURCE[, SOURCE...]                 # read to the end, keep nothing\n  skip N PORT                              # drop its first N, for all readers\n  skip X                                   # every output of a call X\n  skip O[, O...] of X                      # just those outputs of X\n           # (also written `skip (O, O...) of X` or `skip (O, O... of X)`)\n  S = wait SUBJECT [timeout 30s]           # finished; S is how it went\n  S = drain NODE                           # end a node, and say how it ended\n  abort NODE [CODE] [MESSAGE]              # end a node with a failure\n  cancel X                                 # ask X to stop\n  fail [CODE] [MESSAGE]                    # end the flow with a status\n  log [LEVEL] WHAT                         # write to the flow\'s own log\n  logf [LEVEL] "fmt" [ARG, ...]            # the same, formatted\n           # `fail`, `cancel`, `abort` and `log` wait for nothing, so they\n           # go in an `if` or a loop body, or carry an `after`: at the top of\n           # a body they race every other statement, and are refused there.\n           # `drain NODE` and `abort NODE` are the two endings a stream can\n           # have: drain marks it final and closes it, which says it is over;\n           # abort says it went wrong. A reader cannot otherwise tell a stream\n           # that finished from one cut short.\n           # `cancel` aborts, and a cancelled run reports `cancelled`. Asking\n           # an action to *finish* instead is not a language construct: it is\n           # `{"command": "stop"} -> X.control_events`, a convention of the\n           # standard library rather than of the language.\n           # The log needs no declared port or manual drain and is created\n           # only when used\n  for V[, V...] in SOURCE [parallel N] { ... }   # once per value; several\n                                           # names take a tuple apart\n  repeat S = START [max N] { ... S <- SOURCE ... until EXPR }\n           # a repeat needs an `until`/`while` or a `max`: there is no\n           # default bound, and nothing ending a loop is refused\n  if EXPR { ... } else { ... }\n  [S =] [try] { ... }                      # these statements as one step;\n           # S is how it went. A condition inside blocks only what is in\n           # the braces, which is what a block is for; without `try` a\n           # failure inside ends the flow, as a call\'s does\n}\n\nstruct NAME {                              # a shape a port may be typed with\n  describe "what these records are"\n  FIELD: TYPE [required] [unique] [A..B] [matching "re"] [one of [..]]\n         [default LITERAL] "description"\n}\n\nA description may be a "..." string, a """...""" one that holds line breaks and\ngives back the indentation the source put in front of it, or either of those\nalone on the line below what it describes, at any indentation. A string with\nanything after it on its line is a value, as it always was. Strings written next\nto each other are one string, so prose that outgrows its line does not need a\n`+`, and `\\"` is a quote inside one. A *keyword\'s* quoted argument — a\n`matching`, a `strformat` — is one literal, since a run there could not be told\nfrom the argument followed by a description.\n\nA `struct` declares a record with named, typed, constrained fields. A port may\nuse the record as its type, and a value may be coerced into it. A declared shape\noutranks a serialisation tag of the same name — what the file says about the\nname is what the file means by it — and it may hold, and be held by, another\nshape. `A..B` bounds a number, a duration or an instant, and the *length* of a\nstring, a byte string or a list; either end may be left off (`1..`, `..200`).\nA shape holding `bytes` anywhere in it cannot go through `| json`, which has\nnothing to carry them in; `| packb` can.\n\nONE VALUE: everything here is a stream, which is the right default for dataflow\n      — but some of what moves through a flow is one value, and `let` gives it\n      a name. `let code = http.status_code` reads one value of that stream and\n      binds it, and the name then stands *where an expression does*:\n      `if code >= 200 and code < 300 { .. }`, `strformat("%d", code)`,\n      `code == other`. It is also a stream of one wherever a SOURCE goes, which\n      is the other direction: `let image = page.body` then\n      `image | chunk 65536 -> upload.parts` cuts that one value into 64 KiB\n      pieces. A `let` is lazy — nothing is read until the name is — so it may\n      be written where it reads best rather than where the value is first\n      needed, and one nothing reads is reported. An empty stream binds nothing,\n      which `if not code` is how to ask about. A value is read, never written.\n\n      Reading a stream where a value belongs *takes* a value off it. Two places\n      that read one stream for a value take turns on the one view of it, so they\n      see two different values rather than two copies of the first: reading the\n      first value and ignoring the rest would lose data silently. Which reader\n      receives each value is undefined; `after` can order separate statements.\n      Within one statement there is no\n      `after` that could order two reads of one node against each other, and the\n      language reports it (`flow.barrier.value-read-twice`). A `let` is the fix:\n      it names a value, and a value is shared. A stream the language can *prove*\n      carries one value is\n      the exception, and is shared rather than taken: a port that did not say\n      `stream`, a header, a status, or a pipeline that reduced with `| collect`,\n      `| count` or `| first 1`. Those promise one value, so a second arriving\n      ends the flow with `invalid_argument` rather than passing unnoticed.\n\n      `advance V` rebinds a `let` value to the *next* value of the same stream,\n      which is how a flow reads several values of one stream one at a time and\n      knows which is which: `let word = words`, use it, `advance word`, use it\n      again. The guarantee is positional rather than an ordering — the *k*th\n      binding of a name is the *k*th value of its stream however the flow is\n      scheduled — so it holds without a barrier. Statements written above an\n      `advance` keep the value they were resolved against, which is what makes\n      the name read top to bottom. Advancing past the end binds nothing.\n\n      Several names take one value apart: `let name, age = user` by field, and\n      `let first, second = pair` by position. They are the same statement, and\n      which one is meant is a question about the value rather than about the\n      text: each name is looked up as a field, and as a position where there is\n      no such field. For example:\n      `let name, age = match("name={name} age={age:int}", line)`\n      reads what a pattern named. A part is not a value of a stream of its own,\n      so `advance` on one says so rather than binding the next whole value.\n\nSOURCE is a port (in-port, X.out-port), a node, a loop variable, a `let` value,\na header alias, a literal, `status SUBJECT`, `N.id`, `zip(SOURCE, ...)`,\n`interleave(SOURCE, ...)`, or any of those with `.field` / `[i]`.\n`interleave(a, b, ...)` reads several streams *at once* and gives one stream of\ntheir values in the order they arrive, so a fast stream is not held behind a\nslow one. `zip` is the other shape: one tuple per round, in step.\n`zip(a, b, ...)` reads several streams in step and gives one stream of tuples,\nread as `it[0]`, `it[1]`, or taken apart by `for x, y in zip(a, b)`. A source\nthat ends *well* contributes a null to every tuple after it, so the longer\nstream is still read to its end; one that ends with an *error* ends the whole\niteration with that status. It stops when every source has, and it is a stream\nlike any other — `wait`, `drain`, `| first n`, `| drop n`, `| count` all work\non one.\nDEST is an out-port, X.in-port, or a node.\nSUBJECT is a call, a node, a port, or a named wait/drain.\n`wait first of a, b` holds until the first of several *calls* finishes and lets\nthe rest carry on; `wait all of a, b` holds for every one of them.\nA race is a value too: which one won, from zero. `wait first of a, b -> n`,\n`let n = wait first of a, b` and `n = wait first of a, b` all name it.\n`wait all of` has no winner, so it is a barrier only.\nMODIFIERS: tee | via MAP | timeout 30s | after X, Y (a step, or a port/node\n           to wait for) |\n           id EXPR | with "header": EXPR, ... |\n           forward headers "x-name", "x-family-*" (send on the headers this\n           flow was called with, as they arrived; `*` matches a family, and an\n           explicit `with` of the same name wins. Every `x-a11-` header already\n           reaches a step, so this is for the others.)\n           ("try run"/"try call" tolerate\n           failure). `run` needs a handler registered where the flow runs and\n           keeps its nodes off the wire; `call` needs none and goes to the peer.\n           Either may name another flow of the same file, in any order, and\n           needs nothing registered for it: a composition can be factored into\n           several flows and still arrive as one text.\nSTAGES: first N | last N | drop N | truncate N | batch N | window N | flatten |\n        chunk N | group EXPR | sort [by EXPR] [desc] | then SOURCE |\n        where EXPR | map EXPR | join "sep" | strformat "fmt" | mime "text/*" |\n        collect | count | sum [EXPR] | min [EXPR] | max [EXPR] | avg [EXPR] |\n        fold LITERAL as NAME, EXPR | scan LITERAL as NAME, EXPR | distinct |\n        text | json | packb | timeout 30s | pace 100ms | log [LEVEL] [EXPR] |\n        logf [LEVEL] "fmt" [ARG, ...]\n      try SOURCE -> DEST is the pipe\'s own form of the same word: a failure\n      arriving from the source, or refused by the destination, becomes a value\n      rather than the end of the flow. Bind it -- `p = try src -> dest` -- and\n      `status p` says how it went; unbound, a failure is silence and the\n      language says so. It differs from `try` on a *stage*: a stage fails once\n      per value and carries on, while a pipe fails once and stops.\n      A loop may be named too: `done = for x in s { .. }` reads as the loop\'s\n      own outcome, so `drain taken after done` is how a flow says "once the\n      loop is over, that node is over". `for`/`repeat` also take an `after`.\n      Any stage may be written `try STAGE` — a value the stage cannot do is\n      dropped and logged instead of ending the flow — and a `try` stage may say\n      `into DEST` to send those failures somewhere as status records:\n      `docs | try map it as Order into bad -> good`.\n      A per-value stage may say `parallel N` to work on N values at once, and\n      what follows still reads them in the order they arrived. `unordered`\n      gives that up for whatever it saves:\n      `urls | map fetch(it) parallel 8 -> bodies`. Use it for substantial\n      per-value work such as a host round trip or coercion.\n      chunk N cuts each value into pieces of at most N *bytes* — the sizes\n      people write are byte counts, because they are about a frame or a buffer.\n      Text stops at a character boundary rather than splitting one. A value\n      with nothing to cut goes through whole; `batch N` is the one that groups\n      several values into one.\n      then and where may drop the `|`: `history then asked`, `hits where\n      it.ok`. Every other stage keeps it.\n      strformat "fmt" is `map strformat(fmt, it)`, the one-value shorthand.\n      log and logf say what is going past and pass every value on unchanged,\n      so a stage may be dropped into a pipeline and taken out again without\n      touching what comes out of it. `| log` with nothing written logs the\n      value itself; otherwise `it` is the value in hand, as in a `map`.\n      then SOURCE reads this stream and then that one, in that order --\n      `history | then asked` is how a conversation keeps its turns straight,\n      which two writers to one node cannot.\n      sum/min/max/avg read the whole stream and give one value; with an\n      expression they read one field of each (`| sum it.price`). min/max/avg of\n      an *empty* stream give nothing, because the smallest of no values is not\n      a value; `| sum` of one is 0. fold is the general form:\n      `| fold 0 as total, total + it.price` binds `total` to what the last\n      value produced and `it` to the value in hand. `+` is arithmetic, not\n      concatenation -- `| join` is what puts strings together.\n      scan is fold with the values published as they are computed rather than\n      only the last: one value out per value in, carrying state forward. That\n      is a state machine over a stream, and it is the only way to write one --\n      `repeat` carries state but reads its stream from the start on every pass,\n      and `for` walks a stream but carries nothing between passes.\n      `| scan 0 as n, n + 1` numbers a stream, and the start may be a record\n      when the state has more than one part:\n      `| scan {"in": false} as s, {"in": starts-with(it, "BEGIN") or s.in}`.\n      window N is batch\'s overlapping form: one list of the last N values per\n      value, once N have arrived. It is what a question about *neighbours*\n      needs — a pattern spanning two lines is invisible to a `batch`, because a\n      boundary falls somewhere and half the matches fall on it. It holds N\n      values and no more, so a window over an endless stream costs nothing that\n      grows, and a stream shorter than N yields nothing.\n      sort reads the whole stream (nothing comes out until it ends), compares\n      the way `<` does, and is stable: values that tie stay in the order they\n      were written. `by` names what to compare and `desc` reverses it.\n      flatten is the inverse of batch: a stream of lists becomes a stream of\n      what they held. A value that is not a list goes through as itself.\n      timeout 30s fails the flow when the *gap* between two values exceeds it,\n      which is what a stalled producer looks like; a whole-step budget is\n      `wait ... timeout` instead. pace 100ms spaces values out and drops\n      nothing -- the producer is held behind the buffer.\n      group EXPR gathers values into a list and closes it when EXPR holds of\n      the one just added — `| group ends-with(it, [".", "?"]) | map join(it)`\n      is how partial pieces become whole sentences. packb writes a value as\n      application/x-msgpack.\nTYPES: string text number integer bool duration time object json list bytes any,\n      a shape this file declares, a quoted mimetype, or a registry tag written\n      unquoted: a11.sdk.AudioBuffer.\n      A container says what it holds: list[string], list[a11.NodeFragment],\n      and `T[]` is the same thing as `list[T]`.\n      A value is made into one with TYPE{field: expr, ...} or EXPR as TYPE --\n      partial in, valid value of that type out. Not `TYPE{` where a `{` would\n      open a block (an if/for header); put it in brackets there.\nEXPR: literals, it (the value a where/map sees), .field, [i], (pipe | count),\n      == != < <= > >= in, and/or/not, + and - (numbers, durations, instants;\n      `-` needs its spaces, since `text-upper` is one name), and the functions\n      len lower upper trim text number bool keys values get join split merge\n      contains starts-with ends-with replace slice default to_chunk from_chunk\n      strformat b64encode b64decode b64urlencode b64urldecode\n      now duration time seconds\n      A list or object literal may spread another in: [...xs, y] and\n      {...it, "tags": [..]}, where a later key wins.\n      b64encode/b64urlencode give text and b64decode/b64urldecode give bytes;\n      the url pair uses the web-safe alphabet and does not insist on padding.\n      starts-with/ends-with take one ending or a list of them; to_chunk(v[,\n      mime]) makes a Chunk, from_chunk(c) reads one back.\n      strformat("%s of %s", a, b) is printf: %s as text, %d %f %x as numbers,\n      printf\'s flags/width/precision (%-8s, %06.2f), %2$s to pick a value by\n      number, %% for a literal percent, and %(SPEC)s to apply a duration unit\n      or a strftime pattern first. Not a Python template: there is nothing for\n      a slot to read into, which is what makes one safe to accept from a model.\nTIME: durations are written 500ns, 250ms, 30s, 2m, 1h, compound as 1m30s500ms,\n      and are values like any other. now() is the clock; instant - instant is a\n      duration; instant +/- duration is an instant; duration +/- duration is a\n      duration, and a bare number on either side counts as seconds. A duration\n      the other way round is below zero and says so.\n      duration(x) and time(x) read a value in: a number of seconds, or the text\n      the language writes: duration("1m30s") or\n      time("2026-08-11T09:14:22Z")\n      — so a duration or an instant that arrived as a string is a value again.\n      seconds(d) is the number of seconds.\n      Formatting: %s gives `1m30s` and `2026-08-11T09:14:22Z`,\n      %(ns)d %(us)d %(ms)d %(s)d %(m)d %(h)d give a duration as one unit, and\n      %(%H:%M:%S)s or %(epoch)d formats an instant.\nSTATUS: a record {"ok": bool, "code": "NOT_FOUND", "number": 5, "message": str}.\n      `try` keeps a failure from ending the flow; `wait`/`status` say what\n      happened; `fail CODE MSG`, `fail NUMBER MSG` or `fail STATUS` ends it.\n      Codes are Abseil\'s: ok cancelled unknown invalid_argument not_found\n      deadline_exceeded already_exists permission_denied resource_exhausted\n      failed_precondition aborted out_of_range unimplemented internal\n      unavailable data_loss unauthenticated.\n\nSteps run concurrently; order comes from the data. Every output of a step is\nread, whether the flow uses it or not. Stages that shrink a stream (first,\ntruncate, where) do so before the next step ever sees it, `skip N PORT` does it\nfor every reader at once, `-> _` performs a pipeline and keeps none of it, and\n`nodes` blocks keep a step\'s traffic off the wire.\n'

EXTENSION module-attribute

EXTENSION = '.flow'

BUILTINS module-attribute

BUILTINS: frozenset[str] = frozenset(_flow.vocabulary()['builtins'])

STAGES module-attribute

STAGES: dict[str, str] = _flow.stages()

FAIL_CODES module-attribute

FAIL_CODES: tuple[str, ...] = tuple(sorted(code.upper() for code in _flow.vocabulary()['status_codes']))

Compiling

loads compiles source that arrived as a string, load a .flow file, and register does both and publishes the result as actions in one call. A problem raises FlowSyntaxError, which carries the line and column and converts to an A11 status.

a11.flow.loads

loads(source: str, source_name: str = '') -> Program

Compile Flow source into a Program.

Parameters:

Name	Type	Description	Default
source	str	The text of one or more flow declarations.	required
source_name	str	A name for error messages, usually a file path.	''

Raises:

Type	Description
FlowSyntaxError	On any problem, with the line and column it was at.

a11.flow.load

load(path: str | PathLike[str]) -> Program

Compile a .flow file.

a11.flow.register

register(source: str, registry: Any, source_name: str = '') -> Program

Compile Flow source and register every flow in it as an action.

The one call a service needs to accept a composition from outside and make it runnable: after this the flows are in the registry, and a session dispatches them like any other action.

a11.flow.diagnostics.FlowSyntaxError

FlowSyntaxError(message: str, line: int, column: int, source_name: str = '')

Bases: Exception

A Flow source file that could not be read.

What a11.flow.loads raises: the strict door onto an engine that otherwise recovers and reports everything. It carries the position, so an author -- or a model writing a flow -- is told exactly where the problem is, and it is built from the first error Diagnostic rather than from a second opinion about what is wrong.

to_status

to_status()

The A11 status a caller sees when a flow will not compile.

Programs and flows

a11.flow.plan.Program

The flows compiled from one Flow source file.

A program is self-contained: its flows may call each other by name, and anything else they call is looked up in the action registry of whatever runtime dispatches them.

flows property

flows: dict[str, FlowPlan]

Every flow, by name, in declaration order.

has_entry property

has_entry: bool

Whether the file declares a flow { ... } -- a program.

Returns a bool because an entry flow has no name and cannot be accessed through program["..."], run, or call. Use this to choose between run_program and a named flow from names.

main property

main: FlowPlan

The first flow declared, which is the one a file is usually about.

names property

names: list[str]

Every named flow, in the order the file declares them. The entry flow -- flow { ... } -- is not here: it has no name, and it is run rather than called.

get

get(name: str) -> FlowPlan | None

The flow of this name, or None.

register_all

register_all(registry: ActionRegistry) -> Program

Register every flow in registry.

a11.flow.plan.FlowPlan

One compiled flow: an action schema, and the graph implementing it.

A handle onto the program it came from, which keeps the program -- and the syntax tree its graph borrows -- alive for as long as anything holds the flow. So a handler taken from one still runs after the program variable has gone.

handler property

handler: ActionHandler | NativeActionHandler | None

The action handler that runs this flow.

headers property

headers: Mapping[str, ActionHeaderSchema]

The declared headers, by name.

inputs property

inputs: Mapping[str, ActionPortSchema]

The declared input ports, by name, as the action schema has them.

outputs property

outputs: Mapping[str, ActionPortSchema]

The declared output ports, by name, as the action schema has them.

schema property

schema: ActionSchema

The ActionSchema a flow presents.

A flow is an action: it has ports, headers and a name, so anything that can dispatch an action can dispatch a composition without being told it is one.

Built through the Python validator rather than taken from the native schema, because a port's typeinfo -- the Python type its JSON schema comes from, and so what a model is shown -- is something only this side can supply.

action

action(**kwargs: Any) -> Action

Build a standalone Action for it.

describe

describe() -> dict[str, Any]

The whole composition as plain data.

The flow.plan/v1 entry for this flow: its ports, headers, node maps and steps, nested bodies and all.

invoke async

invoke(inputs: Mapping[str, Any] | None = None, **kwargs: Any) -> dict[str, Any]

Run the flow once, here, and collect its outputs.

See a11.flow.runtime.invoke for what the keywords mean.

make_handler

make_handler(dispatch_stream: WireStream | None = None) -> ActionHandler | NativeActionHandler | None

The action handler that runs this flow.

dispatch_stream is only for a flow a client runs over a session it already holds: the calls that belong to the peer are bound to that stream, and the flow's own action is not. An action that is run locally and holds a stream ends that stream when it finishes, after which the session can dispatch nothing.

register

register(registry: ActionRegistry, name: str | None = None) -> FlowPlan

Register this flow as an action in registry.

After this the composition is an action like any other: a session dispatches it, another flow calls it, and a model can be offered it as a tool, without any of them knowing it is a composition.

The compiled graph

A compiled flow is data: describe renders the whole composition, which is what makes one reviewable before it is run.

a11.flow.plan

The compiled program a Flow file becomes, as Python holds it.

The language is implemented once, in C++ (cpp/a11/flow/), and this is the Python frontend onto it: Program and FlowPlan are the classes the native extension exports, with the conveniences a Python caller expects attached onto them -- mapping access, registration, and running one. There is no shadow model here and no second resolver: a flow's ports, steps and diagnostics all come from the one implementation, so what a11 flow check says and what flow.loads raises cannot disagree.

Two rules make the compiled graph predictable:

• Steps run concurrently. Order comes from the data, not from the order the statements were written in. A call is dispatched at once and its inputs stream in while it works. Where an order is genuinely needed, after, wait and drain say so.
• A stream read inside a loop or branch is materialised. The runtime buffers it once, in the scope that owns it, and replays the buffer to each reader. Each loop pass sees the same outer value; this is the one case where the language buffers a stream for repeatable reads.

A flow's shape is readable as plain data with FlowPlan.describe, which is the same flow.plan/v1 payload a11 flow describe prints.

TYPE_NAMES module-attribute

TYPE_NAMES: dict[str, type | str] = {'string': str, 'text': str, 'number': float, 'integer': int, 'int': int, 'bool': bool, 'boolean': bool, 'object': dict, 'json': dict, 'list': list, 'array': list, 'bytes': bytes, 'time': 'a11.Time', 'duration': 'a11.Duration', 'any': 'application/json'}

compile_source

compile_source(source: str, source_name: str = '') -> Program

Compile Flow source into a Program.

Raises:

Type	Description
FlowSyntaxError	On any lexical, grammatical or naming problem, with the line and column it was found at.

Running one

a11.flow.runtime

Running a compiled Flow program on A11.

The runtime is native (cpp/a11/flow/runtime.{h,cc}): a flow becomes one action, each call in it becomes a nested action on the same session, and each pipe becomes a fiber copying one node into another as values arrive. What lives here is the way a Python caller starts one and reads it back -- start for the streaming path, invoke for the convenience of collecting the lot at the end -- and nothing about the language itself.

The engine provides three behaviours required by streaming compositions:

• Every output is drained. An output port of a called action that the flow does not read is read and discarded anyway, because an unread output stalls the action producing it. skip is the explicit spelling of the same thing.
• A run step keeps its nodes off the wire. A step that runs in this process is bound to no stream unless it asks for tee, so the intermediate streams between two steps of a composition are never replicated to the peer that dispatched it. A nodes block goes further and keeps them out of the session's node map entirely. A call step, by contrast, is exactly a step put on the stream this flow is attached to.
• Inputs are closed. A port the flow feeds is closed when its last writer finishes, and one it never feeds is closed immediately, so a callee waiting for end-of-input is never left waiting on a port the composition was never going to write.

Running

Running(flow: FlowPlan, action: Action, outputs: Mapping[str, AsyncNode], timeout: Duration | None, inputs: Mapping[str, AsyncNode] | None = None)

A flow that has been started, and the port nodes it is moving values on.

A flow's ports are AsyncNode values like any other action's. Read an output node to receive values as the flow produces them. invoke's dict of lists is a convenience on top of this for the callers that want the lot at the end (a tool call, a test, a script), not the other way round.

Inputs work the same way round when they are asked for: a port named in start's open_inputs is handed back on inputs for whoever wants to fill it while the flow runs.

wait async

wait() -> None

Wait for the flow to finish, raising whatever it finished with.

publish

publish(stream: Any, names: Iterable[str] | None = None) -> dict

Mirror the named outputs to stream, and say where they landed.

Returns the node id of each published port, which is what a peer needs to read it: NodeMap.get(id) on the other side of the stream is the same node. The action is not bound to the stream, preventing local action completion from closing the session transport.

collect async

collect() -> dict[str, Any]

Wait for the flow, and gather every output port into a dict.

One value for a one port, a list for a many one. Reading starts before the wait, because an output nobody reads stalls the flow filling it -- the same rule that applies to any action's ports.

make_handler

make_handler(flow: FlowPlan, dispatch_stream: Any = None) -> Callable

The action handler that runs flow.

Registering this makes the composition an action like any other: a peer can dispatch it, another flow can call it, and an LLM can be offered it as a tool, without any of them knowing it is a composition.

dispatch_stream is only for a flow a client runs itself over a session it already holds; invoke passes it, and nothing that registers a flow as an action needs it.

start async

start(flow: FlowPlan, inputs: Mapping[str, Any] | None = None, *, registry: Any = None, session: Any = None, node_map: Any = None, stream: Any = None, dispatch_stream: Any = None, headers: Mapping[str, Any] | None = None, timeout: Duration | None = None, action_id: str | None = None, open_inputs: Iterable[str] = (), publish_to: Any = None, **keyword_inputs: Any) -> Running

Start flow and return its live input and output ports.

Provided inputs are written and closed before this returns. Read outputs from Running.outputs, then await Running.wait for the final status.

Ports named by open_inputs remain available through Running.inputs; their writer must close them. Their node ids are <action_id>#<port>, so a peer can stream chunks with their original media types into the flow.

publish_to attaches an output stream before execution, ensuring it sees values produced immediately after startup. Use action_id when peers need deterministic port node ids.

See invoke for stream against dispatch_stream.

invoke async

invoke(flow: FlowPlan, inputs: Mapping[str, Any] | None = None, *, registry: Any = None, session: Any = None, node_map: Any = None, stream: Any = None, dispatch_stream: Any = None, headers: Mapping[str, Any] | None = None, timeout: Duration | None = None, **keyword_inputs: Any) -> dict[str, Any]

Run flow once, here, and return its outputs collected.

The convenience path for tests, scripts, and tool calls: it runs the flow to completion and returns every output port keyed by name -- one value for a one port, a list for a many one. Inputs may be keywords or a mapping (which is what a port whose name collides with one of the options needs). registry, session and node_map place the flow in an existing runtime, so its calls resolve and dispatch exactly as they would inside a server.

Collecting is the convenience, not the mechanism: a flow's outputs are nodes, and start hands them over live for a caller that would rather read them as they fill.

The two stream arguments are different questions, and a caller wants exactly one of them:

• stream runs the flow as though a peer had dispatched it over that stream. The flow's own ports are mirrored to the peer, which is what a server's caller is waiting for.
• dispatch_stream runs the flow as the client's own, and gives the stream only to the calls that go to the peer. The flow's ports stay here. This is what a client with a session of its own wants: an action that is run locally and holds a stream ends that stream when it finishes, after which the session can dispatch nothing -- so a client passing stream would find its second flow unable to reach the peer at all.

Running a program

A file with a flow { ... } is a program, and running one is a different call from running a flow: it gets argv, a policy, this process's standard streams, and returns its exit code as a result.

a11 flow run greet.flow -- Helena
a11 flow run --root /var/log --timeout 30s watch.flow -- /var/log/system.log

a11 flow run and the standalone a11-flow-run are the same interpreter, so a program behaves identically whichever started it. What differs is what the host can offer it, and that is the entire reason the Python one exists: a program may only call actions that exist where it runs, the binary has exactly the Flow standard library, and this process has whatever Python has.

a11 flow run ask.flow --allow-llm --allow-net \
    --allow-env ANTHROPIC_API_KEY -- "why is the sky blue"

interact_with_llm needs a provider SDK and a credential, both of which live in Python, so examples/006-flow-programs/ask.flow runs this way and no other. --allow-llm is its own flag and not part of --allow-net because a host-registered action is not bounded by the flow policy: the policy governs what the standard library may do and can say nothing about what a Python handler does. Offering one is therefore a separate decision, and the default is to offer nothing.

From Python directly:

a11.flow.run_program

run_program(source: str, source_name: str = '', *, arguments: Sequence[str] | None = None, roots: Sequence[str] | None = None, allow_write: bool = False, allow_run: bool = False, allow_net: bool = False, allow_local_net: bool = False, allow_env: Sequence[str] | None = None, unrestricted: bool = False, timeout_seconds: float | None = None, standard_streams: bool = True, registry: Any = None, session: Any = None, dispatch_stream: Any = None) -> dict[str, Any]

Run a Flow program's entry flow — the flow { ... } with no name.

The same interpreter a11-flow-run is, called in process. One implementation, so a program behaves identically whichever started it.

The reason to run one from here rather than from a shell is that this process has actions of its own. Hand it a registry that already holds them and the program can call them:

from a11 import flow
from a11.actions import ActionRegistry
from a11.sdk import llm_tools

registry = ActionRegistry()
llm_tools.register(registry)
flow.run_program(
    source,
    "summarise.flow",
    registry=registry,
    arguments=["summarise.flow", "notes.txt"],
)

A name already in registry is never replaced by the standard library's: a host that registered its own read_file meant its own read_file.

Call it off the loop when your actions are async

This runs the program to completion, so it blocks the thread it is called on. A Python action handler written async def has to be driven by an asyncio loop — and if that loop is on this thread, it cannot run while this call is blocking it, and the program waits forever on its own handler. So call it in a thread:

outcome = await asyncio.to_thread(
    flow.run_program, source, "program.flow", registry=registry
)

A program that only uses the standard library needs none of this: those actions are native and take no GIL.

Parameters:

Name	Type	Description	Default
source	str	The program's text.	required
source_name	str	What diagnostics should call it, usually the path.	''
arguments	Sequence[str] | None	The program's argv. By convention arguments[0] is the file, as a C program's is, so what a user passed starts at index 1.	None
roots	Sequence[str] | None	Directories the program may reach. Defaults to the working directory.	None
allow_write	bool	Whether it may write, inside those roots.	False
allow_run	bool	Whether it may run programs, confined by the kernel where the platform can.	False
allow_net	bool	Whether it may reach the network. Loopback, private and link-local addresses stay refused unless allow_local_net.	False
allow_local_net	bool	Also allow loopback, private, and link-local addresses (e.g. cloud metadata services).	False
allow_env	Sequence[str] | None	Environment variables it may read.	None
unrestricted	bool	No filesystem sandbox at all. For a file you wrote.	False
timeout_seconds	float | None	A bound on the whole run, applied as the deadline header every standard-library action honours.	None
standard_streams	bool	Whether to bind this process's stdin/stdout/stderr. Clear it in a host with no useful standard input, so a program reading it fails rather than waits forever.	True
registry	Any	An ActionRegistry whose actions the program may call. One is made when omitted.	None

Returns:

Type	Description
dict[str, Any]	{"exit_code": int, "diagnostics": [...]} — the code the program put on
dict[str, Any]	an out exit_code: integer port (or 0), and whatever the compiler said
dict[str, Any]	that was not an error.

Raises:

Type	Description
Exception	When the source will not compile, declares no entry flow, or the program itself failed. What the flow failed with is the message.

a11.flow.check_program

check_program(source: str, source_name: str = '') -> str

What a program's entry flow is, compiling it and running nothing.

Raises if the source will not compile or declares no flow { ... }.

Call it off the loop when your actions are async

run_program runs the program to completion, so it blocks the thread it is called on. An async def handler needs a loop to drive it, and if that loop is on this thread it cannot run while the call is blocking it -- so the program waits forever on its own handler. await asyncio.to_thread(...) is the pattern, and it is what a11 flow run does.

Diagnostics

Everything that reports on a flow -- the CLI, an editor, a CI job -- renders the one Diagnostic shape. See Checking flows from a toolchain for the envelopes it travels in.

a11.flow.diagnostics

Flow diagnostics, and the formats external tools read them in.

One problem found in a flow is a Diagnostic: a stable code, a severity, a family, a range that carries both byte offsets and line/column, a message, and an optional single-edit fix. The CLI, editors, and CI integrations render this shared shape.

The formats are versioned and additive: a reader checks format and ignores fields it does not know, and a new field is not a version change. They are produced by cpp/a11/flow/emit_json.{h,cc}, which provides one writer per envelope. This module supplies Python dataclasses plus text and SARIF renderers. testdata/flow/codes.json is generated from the C++ table.

DIAGNOSTICS_FORMAT module-attribute

DIAGNOSTICS_FORMAT = 'flow.diagnostics/v1'

CODES_FORMAT module-attribute

CODES_FORMAT = 'flow.codes/v1'

TOKENS_FORMAT module-attribute

TOKENS_FORMAT = 'flow.tokens/v1'

PLAN_FORMAT module-attribute

PLAN_FORMAT = 'flow.plan/v1'

SYNTAX_FORMAT module-attribute

SYNTAX_FORMAT = 'flow.syntax/v1'

Diagnostic dataclass

Diagnostic(code: str, message: str, range: Range, severity: Severity = ERROR, family: Family = SYNTAX, flow: str = '', fixes: tuple[Fix, ...] = ())

One problem found in a flow.

as_text

as_text(source: str = '') -> str

The line editors and compilers have printed for decades.

from_payload classmethod

from_payload(value: dict[str, Any]) -> 'Diagnostic'

A diagnostic read back from the wire shape as_json writes.

What the native engine hands over, and what a frontend reading the JSON envelope -- a CI script, the IDE plugin -- turns back into an object. Unknown fields are ignored and missing fields use their defaults for compatibility with newer producers.

Severity

Bases: StrEnum

How much a diagnostic matters.

The distinction that earns its keep is between "this cannot work" and "this does nothing": the first stops a flow from compiling, the second is the greyed-out unused symbol every editor already knows how to show.

Family

Bases: StrEnum

The kind of problem, which is the grouping a reader thinks in.

Editors may expose each family as a switchable inspection, and CI can gate on selected families. The family is explicit in the output.

Position dataclass

Position(offset: int = 0, line: int = 1, column: int = 1)

One place in the source: the byte offset, and the line and column at it.

All three travel because each consumer wants a different one -- offsets to edit with, line and column to display. Converting between them requires the source text, which may not accompany a JSON diagnostic. Lines and columns are 1-based, as the lexer has always reported them.

Range dataclass

Range(start: Position, end: Position)

Half-open span of source, [start, end).

Edit dataclass

Edit(start: int, end: int, text: str = '')

One replacement of a span of source. Empty text is a deletion.

Fix dataclass

Fix(label: str, edits: tuple[Edit, ...] = ())

Edits that would fix a diagnostic, where one obvious edit exists.

A frontend applies these blind -- it never re-derives what the fix should be -- so a fix that guessed would be a fix that corrupted somebody's file.

CodeInfo dataclass

CodeInfo(code: str, family: Family, severity: Severity, summary: str)

What a diagnostic code means, from the generated table.

LineIndex

LineIndex(source: str)

Line and column lookup over one source text.

Built once per file and shared by everything that reports a position, so a diagnostic never costs a scan of the source to locate.

Over source bytes, not characters. The lexer, diagnostics, and edits use byte offsets, while Python strings use code-point indexes. For example, § occupies one code point and two UTF-8 bytes. Use characters_of when a character count is required.

length property

length: int

The byte length used to bound offsets.

at

at(offset: int) -> Position

The position at a byte offset, clamped to the end of the source.

between

between(start: int, end: int) -> Range

A range from two byte offsets.

offset_of

offset_of(line: int, column: int) -> int

The byte offset of a 1-based line and column, clamped to the source.

The inverse of at, for the errors the compiler reports by line and column rather than by offset.

characters_of

characters_of(offset: int) -> int

How many characters of the source precede a byte offset.

For the one consumer that is specified in characters rather than bytes: SARIF's charOffset.

known_codes cached

known_codes() -> tuple[CodeInfo, ...]

Every diagnostic code the language publishes, sorted by code.

Read from testdata/flow/codes.json, which the C++ table generates: this is a reader of that contract, never a second copy of it.

find_code

find_code(code: str) -> CodeInfo | None

The entry for code, or None if nothing publishes it.

sort_diagnostics

sort_diagnostics(diagnostics: Iterable[Diagnostic]) -> list[Diagnostic]

The order every frontend presents them in: by position, then by code.

diagnostics_envelope

diagnostics_envelope(source: str, diagnostics: Sequence[Diagnostic]) -> dict[str, Any]

The flow.diagnostics/v1 envelope.

counts is there so a gate can be written without walking the list.

codes_envelope

codes_envelope() -> dict[str, Any]

The published code table, as flow.codes/v1.

sarif_log

sarif_log(source: str, diagnostics: Sequence[Diagnostic], index: LineIndex | None = None) -> dict[str, Any]

A SARIF 2.1.0 log for one file's diagnostics.

Code-scanning services and CI annotators consume SARIF directly. Every rule in the log is documented from the published code table.

index contains the source text used by the diagnostics. SARIF specifies charOffset in characters while a diagnostic carries bytes, and the two differ after a non-ASCII character. Without an index, character-offset fields are omitted while line and column remain available.