Building workflows

Introduction

A workflow definition is plain JSON: { id, name, nodes, edges }. You can write it by hand, as the examples here do, or draw it in the visual builder. @pilotoo/workflow/definition is browser-safe (it needs only zod) and exports workflowDefinitionSchema and validateWorkflow, so a builder UI can validate without the engine.

Wiring two nodes

Every node has input ports and output ports. An edge connects an output port of one node to an input port of another.

{
  "id": "greet",
  "name": "Greet a user",
  "nodes": [
    { "id": "start", "type": "trigger.manual", "config": {} },
    { "id": "fetch", "type": "http", "config": { "url": "https://api.example.com/users/{{ $input.userId }}" } },
    { "id": "done", "type": "output", "config": { "name": "message", "value": "Hello {{ $node.fetch.body.name }}" } }
  ],
  "edges": [
    { "from": { "node": "start", "port": "out" }, "to": { "node": "fetch", "port": "in" } },
    { "from": { "node": "fetch", "port": "out" }, "to": { "node": "done", "port": "in" } }
  ]
}

Run it with input { "userId": 7 } and a finished run’s output is { "message": "Hello Ada" }: the object built by the output nodes that ran.

Part	Shape
Workflow	{ id, name, description?, nodes, edges }
Node	{ id, type, config, retry?, position? } (config defaults to {})
Edge	{ from: { node, port }, to: { node, port } }

position is for the builder and ignored by the engine. An edge carries no condition: branching is done by nodes with several output ports. Over REST the server mints the id, so you leave it out of POST /workflows.

Reading data with expressions

Any config value may hold expressions.

Expression	Reads
{{ $input.x }}	The run input
{{ $node.<id>.<path> }}	A node’s output; items[0].name and items.0.name both work
{{ $vars.y }}	A value a set node stored with as
{{ $item }}, {{ $index }}	The current item and its index, inside a forEach body

A lone expression keeps the type of what it reads. Inside a longer string it is stringified (objects as JSON, nothing as an empty string). An unknown root, or a node that has not run, fails the node. A missing leaf is undefined.

Tip

There is no eval. Logic beyond reading a path goes in a code node.

Branching

A node leaves through one output port, the first by default. Condition nodes choose the port:

• if leaves through true or false.
• switch leaves through the name of the matching case, or default.
• forEach leaves through each (per item) and done.
• approve leaves through approved or rejected.

An if node with two conditions:

{
  "id": "check",
  "type": "if",
  "config": {
    "combine": "and",
    "conditions": [
      { "left": "{{ $node.fetch.body.active }}", "operator": "truthy" },
      { "left": "{{ $node.fetch.body.plan }}", "operator": "eq", "right": "pro" }
    ]
  }
}

Each condition is { left, operator, right? }. Operators: eq, neq, gt, gte, lt, lte, contains, startsWith, endsWith, matches (regular expression), exists, empty, truthy. Ordering operators compare as numbers when both sides are numeric, otherwise as strings. combine is and (default) or or.

A switch node compares value with each case by JSON equality:

{
  "id": "route",
  "type": "switch",
  "config": { "value": "{{ $input.kind }}", "cases": [{ "name": "bug", "equals": "bug" }, { "name": "billing", "equals": "billing" }] }
}

Edges then leave route from the ports bug, billing and default.

Calling an HTTP API

{
  "id": "create",
  "type": "http",
  "config": {
    "url": "https://api.example.com/orders",
    "method": "POST",
    "headers": { "x-source": "workflow" },
    "body": { "sku": "{{ $input.sku }}" }
  }
}

The output is { status, headers, body }. body is parsed when the response is JSON, otherwise it is text. An object body is sent as JSON with a content-type header unless you set one.

Option	Default
method	GET (also POST, PUT, PATCH, DELETE, HEAD)
timeoutMs	30000
failOnStatus	true: a non-2xx response fails the node
query, headers	{}

The node needs the fetch port. The server supplies it.

Calling a model

An llm node makes one call on a model profile. Without schema its output is the reply text. With schema (JSON Schema) the output is a structured object you can read with expressions.

{
  "id": "classify",
  "type": "llm",
  "config": {
    "profileId": "default",
    "system": "You triage support messages.",
    "prompt": "Is this urgent? {{ $input.text }}",
    "schema": { "type": "object", "properties": { "urgent": { "type": "boolean" } }, "required": ["urgent"] }
  }
}

profileId must be a model profile the caller can read; otherwise the node fails with “Model profile … was not found”. See Models.

Calling a tool

A tool node calls a tool of the server’s registry by name, with an explicit input and no model.

{ "id": "page", "type": "tool", "config": { "tool": "notify_oncall", "input": { "message": "{{ $node.classify.summary }}" } } }

A string result that is valid JSON is parsed. An unregistered name fails the node. The tool runs as the run’s caller, so a tool that reads the caller, such as connections_execute, works here as it does in a conversation. See Tools.

Using a connected account

When the server has connections, every tool of the caller’s connected services is a node kind of its own, typed connection. and the tool’s qualified name. The builder lists them with the others, such as Gmail · Send message, and draws the arguments form from the tool’s parameters.

{
  "id": "report",
  "type": "connection.gmail__send_message",
  "config": { "arguments": { "to": "me@example.com", "subject": "Daily report", "body": "{{ $node.summary }}" } }
}

account picks one of the caller’s accounts by id or label; leave it out for the default. The node leaves by out with the tool’s result, and by failed with the error the connection answered, such as { "code": "not_connected", "message": "..." }. Tools that write, such as sending mail, run like the others: add an approve node before one when a person should confirm it.

Tip

A run needs only the node’s type to call the tool, so a stored workflow keeps running when the catalog cannot be read. The builder and validation read the caller’s catalog, and an unknown tool there is an unknown type.

Running an agent

An agent node runs an agent definition with a prompt. With outputSchema the answer is parsed as JSON and validated.

{
  "id": "research",
  "type": "agent",
  "config": { "agentId": "researcher", "prompt": "Summarize {{ $input.topic }}", "outputSchema": { "type": "object", "properties": { "summary": { "type": "string" } } } }
}

If the agent asks the user something, the run pauses. See Pausing for answers.

Running a code snippet

{ "id": "total", "type": "code", "config": { "code": "return nodes.fetch.body.items.length;" } }

The snippet runs in the host’s sandbox with input, nodes, vars, item and index in scope, and its return value is the node’s output. timeoutMs defaults to 5000. The host decides the isolation and the snippet syntax its sandbox accepts, and without a sandbox the node fails; see Server integration.

Looping over a list

forEach takes items, an expression resolving to an array (otherwise the node fails). It leaves through each once per item and through done once all items are through.

{ "id": "loop", "type": "forEach", "config": { "items": "{{ $input.orders }}" } }

• The body is the nodes reachable from each. It must end with an edge back into the forEach node’s in port.
• Items run in parallel, each with $item and $index in scope. The output of the node that wires back is that item’s result.
• After the loop, $node.<forEachId> is the array of results in item order. Body nodes’ outputs are per item and are not readable outside the body.

Joining branches

A join is a node that several edges lead into. It matters when branches reconverge.

• It waits for all its incoming edges only when they are reachable by exactly the same paths through branch nodes (an unconditional fan-out, or a fan-out under one arm of an if).
• Otherwise each arrival fires it: arms of an if that reconverge run the next node once per arm that ran.
• Path analysis is capped at 64 distinct paths per node. Past that, the node and everything downstream fire per arrival. A run never hangs because of the cap.

Storing values and naming results

set emits value. With as, it also stores the value in $vars for the rest of the run. output names a value the run returns: several output nodes build one object, and the same name written twice keeps the later write.

{ "id": "keep", "type": "set", "config": { "value": "{{ $node.fetch.body.email }}", "as": "email" } }

Retrying a failing node

{ "id": "fetch", "type": "http", "config": { "url": "https://flaky.example.com" }, "retry": { "maxAttempts": 3, "backoffMs": 2000 } }

maxAttempts is an integer of at least 1, and backoffMs defaults to 1000. Each failed attempt emits NODE_FAILED.

Validation

validateWorkflow(workflow, kinds) returns a list of issues, each with a code and a message. The server runs it on every create, update and start, and exposes it as POST /workflows/validate.

Code	Meaning
unknown_node_type	No registered kind has that type
invalid_config	The node’s config fails its kind’s schema
duplicate_node_id	Two nodes share an id
dangling_edge	An edge points to a node that does not exist
unknown_port	An edge leaves from or enters a port the node does not have
missing_trigger	There is no trigger node
multiple_triggers	More than one trigger node
unreachable_node	The trigger cannot reach the node

A broken node yields one issue, not a cascade. Starting a run on a workflow that does not validate throws WorkflowValidationError with its issues, and over REST answers 400.

Node kinds at a glance

Type	Config	Ports (in / out)
trigger.manual	none	none / out
trigger.webhook	none	none / out
trigger.schedule	none	none / out
set	value, as?	in / out
if	conditions, combine?	in / true, false
switch	value, cases	in / each case, default
output	name?, value	in / out
approve	message, data?	in / approved, rejected
http	url, method, headers, query, body?, timeoutMs, failOnStatus	in / out
code	code, timeoutMs	in / out
forEach	items	in / each, done
tool	tool, input	in / out
llm	profileId, system?, prompt, schema?	in / out
agent	agentId, prompt, outputSchema?	in / out
connection.<tool>	account?, arguments	in / out, failed

The server registers tool, llm and agent because they use its tool registry, model profiles and agent runner, and the connection. kinds because they use its connections. The approve node is covered in Running workflows.

Writing your own node kind

A kind is a descriptor (type, category, label, a zod config schema, input and output ports) plus execute(context, config, input). Outputs may be a function of the config. defineNodeKind types it from its schema.

src/kinds/slugify.ts

import { defineNodeKind } from '@pilotoo/workflow';
import { z } from 'zod';

export const slugify = defineNodeKind({
  type: 'slugify',
  category: 'transform',
  label: 'Slugify',
  config: z.object({ text: z.string() }),
  ports: { inputs: ['in'], outputs: ['out'] },
  execute: (_context, config) => ({
    output: config.text.toLowerCase().replace(/[^a-z0-9]+/g, '-'),
  }),
});

config arrives with expressions already resolved and parsed. input is the output of the single upstream node (undefined when several edges enter). execute returns { output, port?, vars?, outputs? }. Register kinds with workflowKinds on the server, or with kinds on the runtime.

Next steps

• Running workflows: start a run and follow it.
• Triggers: how the trigger node gets fired.
• Server integration: register kinds, the sandbox and the REST routes.