Flow language

Flow composes actions into a new action. A host loads a document with application code or receives and checks it at runtime, then resolves it against the current action registry. Documents cannot 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

Complete flow

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
    }
  }

  brief.summary -> answer
}

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.

Port values

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.

Constructing typed values

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 create and read a Chunk. Content-bearing types store their content in chunks.

Available types come from the host's serialisation registries. A flow cannot import types. TYPE{...} is unavailable where { opens a block, including an if condition or a for source. Parentheses permit a typed construction in these positions: if (T{done: true}).done {.

Action composition

Local runs and remote calls

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

search = run web-search(query: question)
llm    = call interact_with_llm(...)

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 name another flow in the same file without a registry entry:

flow ask-twice {
  in  question: string
  out answers:  string stream
  first  = run ask(question: question)
  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. The runtime drains declared call outputs that the flow does not reference.

skip n port removes values from the node before fan-out. Every reader starts after those values. In comparison, | drop 1 trims only its pipeline:

rows = run read-csv(path: path)
skip 1 rows.lines
rows.lines | count -> data-rows
rows.lines -> passed-through

Counts on one node are additive. skip 1 x and skip 2 x remove three values, independent of statement order, because compilation sums the counts. The target must be a port or node; pipelines are reader-specific.

-> _ executes a pipeline and discards its result:

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 concatenates this stream with the source that follows.

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 expands a stream of lists into their values.

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, ...) reads all sources concurrently and emits each value on arrival. zip reads its sources in step and emits one tuple per round:

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 defines a custom reduction:

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 uses the same syntax as fold. fold yields one value when the stream ends; scan yields the current accumulator for each input value.

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.

Stream timing

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 can bound its concurrent work:

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 and time

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.

utf8encode(text) produces bytes for an application/octet-stream input. utf8decode(bytes) produces text for a text/plain input and fails with invalid_argument when the bytes are malformed. These operations preserve the distinction between binary ports and text ports; use b64encode and b64decode when the representation itself is Base64.

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
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 language's only arithmetic operators. A bare number beside a duration counts as seconds; seconds(d) returns the numeric value. Subtraction can produce a negative duration and does not use A11's infinite-timeout convention. - 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) parse the formats described above:

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" }

These functions accept text from headers, JSON fields, or model output without a separate format string.

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.

Value representation

| 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.

Header propagation

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.

Local step traffic

nodes fetched { ... } gives its calls a private NodeMap. Their ports and fragments remain outside the session's node map. For example, four fetched pages can remain local while one answer returns to the peer. A run step keeps its nodes local unless it specifies tee; a nodes block also applies to call steps.

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

try converts an expected step failure into a status the flow can handle. It applies to either call verb:

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 a call or flow-written node finishes. A bound wait also returns the outcome. status x returns the same value in an expression, and drain node reads the outcome while ending the named node.

A status is data:

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

A flow can branch on a status, write it to an output, or raise it again. fail accepts an Abseil canonical code in lower or upper case (not_found, NOT_FOUND), a number computed at runtime, or a complete status record. fail outcome preserves the status; fail invalid_argument outcome.message changes the code while retaining its message.

Waiting on a failed step ends the flow with that status unless the step used try.

wait first of a, b holds until the first call finishes and leaves the others running. wait all of a, b holds until every call finishes. These forms accept calls; node completion uses wait or drain on the node.

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

won = wait first of primary, backup
wait first of primary, backup -> chosen
let n = wait first of primary, backup

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 drops values that the stage cannot process while allowing the stream to continue.

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

Failures sent through into use the same status-record shape as status x. They can be counted, written to a port, or read by the caller. Without try, a stage failure ends the flow.

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 a tried pipe to retain its status. An unbound failed pipe closes its destination early without exposing the failure to a reader, so the compiler reports it. A tried stage can fail once per value and uses into for dropped values. A tried 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 up to n passes concurrently. A stream read inside a loop or branch is materialised once and replayed to every pass. Its buffer grows during reads, so each pass waits only for the requested value. It does not wait for the source stream to finish.

A named loop returns its outcome in the same shape as s = try { .. }:

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

The loop counts as one writer of an outer node for its lifetime. Its final release closes taken; the explicit drain records that dependency in the source. for and repeat also accept after.

A for can use until or while. The condition runs after each pass, so the body runs at least once and includes the value that ends the loop.

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

The loop stops reading without cancelling the producer, matching | first n. This condition cannot be combined with parallel, because several passes in flight would make the stopping point depend on scheduling. <- applies only to repeat; a for obtains each value from its source stream.

advance reads successive values without creating a loop. Compilation assigns its offset, so three uses read the first, second, and third values. A name bound outside a loop cannot be advanced inside it.

Stream completion and failure

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 ends a node with a failure:

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

Readers receive the failure status. abort accepts the same code and message as fail and does not wait. Place it in an if, a loop body, or a statement with after when ordering matters.

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.

Language tables

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.

Flow documents can arrive through a gateway, client, or model and run without a repository change or deployment.

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 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. Unbound call outputs are drained automatically, so they cannot stall their 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: _ 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. This bounds the input sent to a model. | packb writes a value as MessagePack. An existing MessagePack chunk passes through without re-encoding.
• 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 to another step ends the node immediately. Use drain n after <call> to wait for that step first.
• 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
            | "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]
destination := reference | "_"      # `_` keeps nothing, and is not a name
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

Pattern matching

match extracts named fields from text as a stream stage or value function. Literal text matches itself, whitespace matches any run of spaces or tabs, and {name} captures text up to the next literal. For example, lines | match "name={name} age={age:int}" converts name=Alice age=27 into a record with name and age fields.

A hole can specify int, number, bool, word, line, rest, duration, time, or json. {} creates a positional capture read as it[0]; {{ and }} match literal braces. Patterns search anywhere in a value without leading or trailing wildcards. A hole remains on its line unless its type permits line breaks.

The stage drops values that do not match; the function returns null. Literal patterns expose their named fields to completion and diagnostics. Invalid literal patterns produce a diagnostic at their source location.

Field checking

Fields come from declared struct types and named match holes. Missing fields produce diagnostics for both sources. Values typed as object or json, an it value without a pattern, and positional patterns have no named-field check. Checking stops after one level: src.meta.title checks meta and does not validate title.

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.

Descriptions and docstrings

Descriptions document flows, ports, and headers. A quoted string may follow a declaration or appear on the next line:

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. A long description therefore aligns with its declaration while retaining its internal layout. 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 flow constructs registered values when an input port requires a concrete type:

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 N PORT                              # drop its first N, for all readers\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`. Graceful\n           # completion uses the standard-library control-port convention:\n           # `{"command": "stop"} -> X.control_events`.\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 use a "..." string or a dedented """...""" string with line\nbreaks. Either form may appear alone below its declaration at any indentation.\nA string followed by another token is a value. Adjacent strings concatenate;\n`\\"` inserts a quote. A keyword argument such as `matching` or `strformat`\naccepts one literal to distinguish it from a following 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\ntakes precedence over a serialisation tag with the same name and may contain or\nbe contained by another shape. `A..B` bounds a number, duration, instant, or the\n*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: every source is a stream. `let` reads and names one value.\n      `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 also acts as a one-value stream wherever a SOURCE is\n      accepted: `let image = page.body` then\n      `image | chunk 65536 -> upload.parts` cuts that one value into 64 KiB\n      pieces. A `let` reads lazily on first use. The compiler reports unused\n      bindings. An empty stream binds nothing, which `if not code` tests. A\n      value is read-only.\n\n      Reading a stream where a value belongs consumes one value. Two value reads\n      from one stream receive different values from their shared view. Which\n      reader receives each value is undefined; `after` orders 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: 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`.\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: the *k*th binding of a name is the\n      *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. The value determines the form:\n      each name is looked up as a field, then as a position when no such field\n      exists. For example:\n      `let name, age = match("name={name} age={age:int}", line)`\n      reads what a pattern named. A part has no stream of its own, so `advance`\n      reports an error on it.\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, ... (steps or ports/nodes\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 status\n      value. 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 chunks stop at character boundaries. 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 publishes each computed accumulator; fold publishes only the last.\n      It produces one value per input while carrying state forward, defining a\n      state machine over the stream. `repeat` carries state but reads its stream\n      from the start on every pass; `for` walks a stream without carrying state\n      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      utf8encode utf8decode\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      utf8encode gives bytes and utf8decode gives text; decoding rejects\n      malformed UTF-8 rather than replacing it.\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.

Runtime API

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.
• 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 use the same interpreter. Their hosts expose different actions: the standalone binary provides the Flow standard library, while the Python process can also provide registered Python actions.

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

interact_with_llm requires a provider SDK and credential from Python, so examples/006-flow-programs/ask.flow runs under the Python host. --allow-llm is separate from --allow-net: flow policy does not bound host-registered actions. Policy governs the standard library; registered Python handlers retain their own capabilities. The host exposes no optional action by default.

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.

This calls the a11-flow-run interpreter in process. Pass a registry to expose actions from the host process:

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.

Run off-loop for async action handlers

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. Use only with trusted source.	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. A read then fails instead of waiting indefinitely.	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 { ... }.

Run off-loop for async action handlers

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

The CLI, editors, and CI integrations use 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.