Let a model compose its own tools¶

A model that calls tools one at a time must read each intermediate result and include it in a later request. Large pages, transcripts, and file listings then consume the context window, add input tokens, and require the model to copy values between calls.

a11.sdk.flow_tools offers the alternative as three tools. The model writes a flow — a composition of the actions available for that turn — and runs it as one step. No new handler is generated or deployed. The host checks the document against its current registry and allow-list, then the values moving between steps bypass the model.

Flow does not add a persistent graph or another agent loop. The runtime starts the declared actions, pipes their named ports, and lets data arrival coordinate them. The model still chooses the composition, but the deterministic transfer of intermediate values runs outside its context.

Use skills for knowledge and Flow for a checked procedure¶

The Agent Skills specification defines a portable folder containing SKILL.md instructions and optional scripts, references, and assets. Compatible harnesses use progressive disclosure: they advertise each skill's name and description, then load its instructions when a task appears to match.

That format is useful for domain knowledge, judgment, and procedures whose details vary with the task. A skill may bundle tested scripts, but its SKILL.md procedure is still interpreted by the model. The model decides whether the skill applies, selects each tool, and copies each tool result into a later request. More detailed instructions can improve consistency, but they do not guarantee that every described step occurs.

Flow is the stronger form when the procedure can be expressed through actions:

flow_check resolves action names, ports, and types before execution, and flow_run checks the model turn's action permissions before dispatch;
branches, loops, ordering, concurrency, and failure handling have defined runtime semantics;
an output pipes directly to the next input without entering model context;
large intermediate values can remain in local nodes and off the wire;
only the Flow source and declared results need to occupy model context.

This distinction matters in a research task. A skill can tell a model to search, fetch several pages, trim them, and summarize them. The model must still perform that loop and observe the page contents. A Flow expresses the same procedure once:

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

for hit in search.hits parallel 3 {
  page = run web-fetch(url: hit.url)
  page.text | truncate 2000 -> brief.pages
}

brief.summary -> answer

The runtime performs every declared pipe and control-flow construct. It does not depend on the model remembering the next instruction after each tool call. Model calls and external services remain variable, and concurrent streams may arrive in different orders. The model sees the compact composition and final answer, not every fetched page or transfer step.

Skills and Flow can work together. a11.sdk.flow_tools publishes its compact language reference as an Agent Skill, helping a model decide when and how to write a composition. flow_check then validates the document, and flow_run executes the checked semantics. The skill supplies judgment; Flow supplies the data path and control flow.

When a model chooses whether to call flow_run, that choice is still model judgment. Once the document is submitted, completing its declared procedure no longer depends on further skill activation or instruction following.

from a11.actions import ActionRegistry
from a11.sdk import bash, flow_tools

registry = ActionRegistry()
bash.register(registry)  # the tools to compose
flow_tools.register(registry)  # and the ability to compose them

system_prompt = "\n\n".join(
    [bash.get_system_prompt(), flow_tools.get_system_prompt()]
)

Register them on the registry that holds the actions they are meant to compose: a flow resolves its calls by name through the registry it runs under. In a11 chat, for instance, the shell tools are the client's — so the flow tools belong there too, not on the gateway.

On the gateway¶

The gateway serves them by default, after everything else it registers, so a composition can reach its shell, audio and conversation actions:

a11 gateway run                    # flow_actions, flow_check, flow_run served
a11 gateway run --no-flow-tools    # not served

Clients can send a flow that connects capture_transcription directly to interact_with_llm, avoiding transcript round trips through the client and model. scripts/flow_playground.py demonstrates this workflow: it discovers gateway actions, checks a flow, captures a spoken sentence, and sends the model reply history into the next turn with then.

A flow can also run on the client and call the gateway's actions, which is what that script does with its own microphone-to-model composition. For that, register the actions' schemas locally with no handler: a flow resolves every call against its registry to learn the port names, and dispatches it to the session precisely when it finds no handler to run it with. The SDK ships the schemas (audio.actions.CAPTURE_TRANSCRIPTION_SCHEMA, INTERACT_WITH_LLM_SCHEMA), so this costs one registry.register(schema.name, schema) per action.

The three tools¶

Tool	What the model does with it
`flow_actions`	Asks what it may compose. Returns each action with its input and output ports — the part a tool definition cannot carry, and exactly what a pipe needs on both sides of a `->`.
`flow_check`	Compiles a flow and describes what it resolves to, without dispatching anything. A syntax error comes back with its line and column.
`flow_run`	Compiles, checks the call targets, runs the flow, and returns its declared outputs as one object.

Client-side streaming with `flow_run`¶

Models typically consume the collected object result. Application clients can instead stream inputs into a running flow and receive incremental outputs:

call = a11.Action(flow_tools.FLOW_RUN_SCHEMA)
await call.call()

# Access flow inputs and outputs through their deterministic node IDs.
said = session.node_map.get(flow_tools.flow_output_node_id(call.get_id(), "said"))
words = session.node_map.get(flow_tools.flow_input_node_id(call.get_id(), "words"))
words.attach_stream(stream)

# Declare streamed inputs and finalize static parameters.
await call["input_streams"].finalize(["words"])
await call["source"].finalize(SOURCE)

# Stream inputs and read incremental responses.
await words.put("one")
response = await said.next_object()
await words.finalize()

Finalize or close every port named in input_streams so the receiving action can observe the end of input.

What the model is not allowed to compose¶

A flow dispatches through the registry, underneath the layer that decides which actions a model may call. So flow_run and flow_check make that decision again themselves, against the same x-a11-allowed-llm-actions header a11.sdk.llm_tools.runner.collect_tools reads: every call in the submitted source is checked, and one naming an action the caller may not reach is refused with PERMISSION_DENIED before anything runs. A flow may not call the flow tools either — that would be a way around the same check, and a way to recurse.

With no such header there is no restriction from this layer: a script or a test driving these handlers is not a model being held to an allow-list.

The instructions¶

The model needs to be told the capability exists, and it needs the language. Both are one text, available in whichever shape the host prefers:

flow_tools.get_system_prompt()  # for composing into a larger system prompt
flow_tools.get_skill()  # the same words as an a11.sdk.skill.Skill

The text embeds a11.flow.REFERENCE, the compact language reference, so prompts use the implemented syntax. a11/sdk/flow_tools/SKILL.md is generated from the same constants and checked in, for a host that loads skills from disk; a test fails when the file and the code disagree.

What a composed step looks like¶

Given web-search, web-fetch and summarize, a model asked a research question can send this as one flow_run call:

flow answer-from-the-web {
  in  question: string required
  out answer:   string
  out sources:  string stream

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

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

  brief.summary -> answer
  skip search.debug
}

The result is {"answer": ..., "sources": [...]}. The nodes fetched block keeps fetched and trimmed pages off the wire, while the model receives only the declared answer and source values. Keep declared outputs small enough for the model to read.

a11.sdk.flow_tools ¶

Flow as a capability you hand to a model, or to a client.

Three Actions -- flow_actions, flow_check and flow_run -- let an LLM compose the tools it already has into an A11 Flow and run the composition as a single tool call. Register them on the registry that holds those tools:

from a11.actions import ActionRegistry
from a11.sdk import flow_tools

registry = ActionRegistry()
flow_tools.register(registry)
system_prompt = flow_tools.get_system_prompt()

The registry matters: a flow calls actions by name through the registry it is run under, so these belong beside the actions they are meant to compose.

Values passed between composed steps do not enter the model context. A flow that fetches four pages and summarises them returns one summary; the model does not read the source pages. Separate model tool calls would include each intermediate in both a result and the next request.

The instructions to put in front of the model come with it, as either get_system_prompt text or an a11.sdk.skill.Skill -- the same words, in whichever shape the host prefers.

A client is the other kind of caller, and it wants flow_run too -- with the flow's ports as nodes rather than an object of values. Naming a port on input_streams leaves it open to write while the flow runs, and every output is readable as it fills; the ids are derived from the call's own (flow_input_node_id, flow_output_node_id). That is the same tool serving a caller that has a session, not a second one.

describe_composable_actions ¶

describe_composable_actions(registry: Any, patterns: list[str]) -> list[dict[str, Any]]

The actions a flow may name, each with its ports and its verb.

A thin projection of the written schema (a11.actions.describe) into the words a flow uses. The written schema is shared with tool discovery. This projection only translates Flow's stream and action terminology from unary and name.

Output port names support pipe expressions, but tool definitions only describe inputs.

runnable distinguishes actions for run from actions for call. Actions registered with a handler are runnable; schema-only registrations are callable through the peer.

Autofilled inputs are omitted because the runtime supplies them before the handler runs. Action logs use :meth:a11.actions.action.Action.log and are not schema ports.

flow_actions `async` ¶

flow_actions(action: Action) -> None

Report the actions this caller may compose, and their ports.

flow_check `async` ¶

flow_check(action: Action) -> None

Compile a flow and describe it, without dispatching anything.

flow_input_node_id ¶

flow_input_node_id(call_id: str, port: str) -> str

Where a caller writes an input port it named on input_streams.

The mirror image of flow_output_node_id, and the same derivation -- a port's node id does not depend on its direction, so the two functions agree by construction and exist separately to say which one a caller means. This also prevents a flow from declaring an input and output with the same name because they would share one node. Input ports are in the session's node map like its outputs, so the caller fills one by writing to its id -- and keeps writing, for as long as it has values.

call = a11.Action(FLOW_RUN_SCHEMA)...
await call.call()
await call["input_streams"].finalize(["words"])  # this port is mine
words = session.node_map.get(flow_input_node_id(call.get_id(), "words"))
words.attach_stream(stream)
await (await words.put("one"))  # the flow reads it now, not later
await words.finalize()          # and this is what ends the port

Arity is not a second mechanism: a port declared stream takes as many values as the caller has, an ordinary port takes the one it carries, and an empty port is a port that carried none. The caller owns the close -- nothing else will do it, and a flow reading a port nobody closes waits, bounded only by the call's own x-a11-deadline (with no such header, not at all).

flow_output_node_id ¶

flow_output_node_id(call_id: str, port: str) -> str

Where a flow's output port lands, for the caller that dispatched it.

A flow's outputs are nodes, and flow_run mirrors them onto the stream the call arrived on, so a caller does not have to wait for the whole composition to see what it is producing. The id is worked out from the call's own id, which the caller chose, so both ends know it without anything being announced:

call = a11.Action(FLOW_RUN_SCHEMA)...
await call.call()
replies = session.node_map.get(flow_output_node_id(call.get_id(), "reply"))
async for token in replies:      # arrives as the model writes it
    ...

The collected result still lands at the end, for callers -- a model making a tool call, most of all -- that only want the lot.

flow_run `async` ¶

flow_run(action: Action) -> None

Compile a flow, check what it calls, run it, and return its outputs.

A port is filled one of two ways, and which one is the caller's to choose. A value in inputs is written and the port closed, which is all a model can express. A port named on input_streams is left open for the caller to write as a node while the flow runs -- which is how a value reaches a flow that has already started, and how a port carrying a real type is fed at all. Neither way looks at how many values the port carries.

verify_calls ¶

verify_calls(program: Program, patterns: list[str]) -> None

Check every call in program against what the caller may call.

Raises:

Type	Description
`StatusException`	`PERMISSION_DENIED`, naming the first action the caller is not allowed to reach, before any of them is dispatched.

get_skill ¶

get_skill() -> 'Skill'

The same instructions as an a11.sdk.skill.Skill.

Imported on demand because a11.sdk.skill needs PyYAML for frontmatter, while generating prompt text does not.

get_system_prompt ¶

get_system_prompt() -> str

Return instructions telling the model it can compose its tools.

The text explains when a flow beats a sequence of tool calls, the order to use the three tools in, and the handful of rules that decide whether a composition works -- and embeds a11.flow.REFERENCE, so the language it describes is the language that is implemented.

register ¶

register(registry: ActionRegistry) -> None

Register all three Flow Actions on registry.

a11.sdk.flow_tools.prompt ¶

The instructions that teach a model when and how to compose its tools.

One text, in two shapes: get_system_prompt for a host that composes a system prompt out of parts (the way a11 chat does with the shell tools), and get_skill for one that loads a11.sdk.skill.Skills. The language reference is not written out again here -- it is a11.flow.REFERENCE, the same cheat sheet the language ships -- so the instructions cannot describe a Flow that the compiler does not accept.

SKILL.md beside this module is generated from these constants and checked in, so a host that reads skills off disk gets the same words. The test suite fails when the file and the constants disagree.

get_system_prompt ¶

get_system_prompt() -> str

Return instructions telling the model it can compose its tools.

The text explains when a flow beats a sequence of tool calls, the order to use the three tools in, and the handful of rules that decide whether a composition works -- and embeds a11.flow.REFERENCE, so the language it describes is the language that is implemented.

get_skill ¶

get_skill() -> 'Skill'

The same instructions as an a11.sdk.skill.Skill.

Imported on demand because a11.sdk.skill needs PyYAML for frontmatter, while generating prompt text does not.

a11.sdk.flow_tools.handlers ¶

Handlers for the three Flow Actions.

Each is a thin adapter over a11.flow: read the action's inputs, compile, and -- for flow_run -- invoke the composition in the caller's own runtime, so its calls dispatch exactly as a nested action's would.

Two rules are enforced here rather than by the language, because they are about who is asking rather than about what a flow means:

A flow may only call actions the caller may call. A composition dispatches through the registry, which is under the layer that checks the x-a11-allowed-llm-actions header, so the check is made again here -- against every call in the submitted source, before anything runs. With no header there is no restriction from this layer: a script or a test that invokes these handlers directly is not a model being kept to an allow-list.
A flow may not call the flow tools. Composing flow_run into a flow is a way to run something the check above just refused, and a way to recurse.

Each handler narrates its run through :meth:a11.actions.action.Action.log. The LLM tool runner reads that separately from the action's outputs, so the narration cannot reach the model's result. No schema here declares a log port.

verify_calls ¶

verify_calls(program: Program, patterns: list[str]) -> None

Check every call in program against what the caller may call.

Raises:

Type	Description
`StatusException`	`PERMISSION_DENIED`, naming the first action the caller is not allowed to reach, before any of them is dispatched.

describe_composable_actions ¶

describe_composable_actions(registry: Any, patterns: list[str]) -> list[dict[str, Any]]

The actions a flow may name, each with its ports and its verb.

A thin projection of the written schema (a11.actions.describe) into the words a flow uses. The written schema is shared with tool discovery. This projection only translates Flow's stream and action terminology from unary and name.

Output port names support pipe expressions, but tool definitions only describe inputs.

runnable distinguishes actions for run from actions for call. Actions registered with a handler are runnable; schema-only registrations are callable through the peer.

Autofilled inputs are omitted because the runtime supplies them before the handler runs. Action logs use :meth:a11.actions.action.Action.log and are not schema ports.

flow_actions `async` ¶

flow_actions(action: Action) -> None

Report the actions this caller may compose, and their ports.

flow_check `async` ¶

flow_check(action: Action) -> None

Compile a flow and describe it, without dispatching anything.

flow_output_node_id ¶

flow_output_node_id(call_id: str, port: str) -> str

Where a flow's output port lands, for the caller that dispatched it.

A flow's outputs are nodes, and flow_run mirrors them onto the stream the call arrived on, so a caller does not have to wait for the whole composition to see what it is producing. The id is worked out from the call's own id, which the caller chose, so both ends know it without anything being announced:

call = a11.Action(FLOW_RUN_SCHEMA)...
await call.call()
replies = session.node_map.get(flow_output_node_id(call.get_id(), "reply"))
async for token in replies:      # arrives as the model writes it
    ...

The collected result still lands at the end, for callers -- a model making a tool call, most of all -- that only want the lot.

flow_input_node_id ¶

flow_input_node_id(call_id: str, port: str) -> str

Where a caller writes an input port it named on input_streams.

The mirror image of flow_output_node_id, and the same derivation -- a port's node id does not depend on its direction, so the two functions agree by construction and exist separately to say which one a caller means. This also prevents a flow from declaring an input and output with the same name because they would share one node. Input ports are in the session's node map like its outputs, so the caller fills one by writing to its id -- and keeps writing, for as long as it has values.

call = a11.Action(FLOW_RUN_SCHEMA)...
await call.call()
await call["input_streams"].finalize(["words"])  # this port is mine
words = session.node_map.get(flow_input_node_id(call.get_id(), "words"))
words.attach_stream(stream)
await (await words.put("one"))  # the flow reads it now, not later
await words.finalize()          # and this is what ends the port

Arity is not a second mechanism: a port declared stream takes as many values as the caller has, an ordinary port takes the one it carries, and an empty port is a port that carried none. The caller owns the close -- nothing else will do it, and a flow reading a port nobody closes waits, bounded only by the call's own x-a11-deadline (with no such header, not at all).

flow_run `async` ¶

flow_run(action: Action) -> None

Compile a flow, check what it calls, run it, and return its outputs.

A port is filled one of two ways, and which one is the caller's to choose. A value in inputs is written and the port closed, which is all a model can express. A port named on input_streams is left open for the caller to write as a node while the flow runs -- which is how a value reaches a flow that has already started, and how a port carrying a real type is fed at all. Neither way looks at how many values the port carries.

Let a model compose its own tools¶

Use skills for knowledge and Flow for a checked procedure¶

On the gateway¶

The three tools¶

Client-side streaming with flow_run¶

What the model is not allowed to compose¶

The instructions¶

What a composed step looks like¶

a11.sdk.flow_tools ¶

describe_composable_actions ¶

flow_actions async ¶

flow_check async ¶

flow_input_node_id ¶

flow_output_node_id ¶

flow_run async ¶

verify_calls ¶

get_skill ¶

get_system_prompt ¶

register ¶

a11.sdk.flow_tools.prompt ¶

get_system_prompt ¶

get_skill ¶

a11.sdk.flow_tools.handlers ¶

verify_calls ¶

describe_composable_actions ¶

flow_actions async ¶

flow_check async ¶

flow_output_node_id ¶

flow_input_node_id ¶

flow_run async ¶

Client-side streaming with `flow_run`¶

flow_actions `async` ¶

flow_check `async` ¶

flow_run `async` ¶

flow_actions `async` ¶

flow_check `async` ¶

flow_run `async` ¶