Workflows
A workflow is data the daemon holds and bodies the daemon never sees. This page is the data: what a definition can say, and what each construct makes the daemon do.
The division of labour
A durable run has two owners. The daemon owns the run between steps: the cursor, the journal, the retry budget, the timers and the waits. Your process owns everything inside a step: the model you call, the prompt you send, the tools you run, the credentials you hold. The daemon is never handed any of that, and it never runs a model on your behalf.
So a workflow arrives in two parts from one file. The graph goes to the daemon as a definition, a document the daemon walks. The step bodies stay in your process, in whatever language you wrote them, for a worker to serve. Nothing has to be deployed to the daemon, and a body can change without the definition changing at all.
The cursor
The daemon holds the run between steps: where it is, what each step answered, what it is waiting for and until when. A worker holds one step at a time, and only while it runs. During a wait no process is alive at all.
Writing one
The builder produces both parts. document() is the definition,
and each step or agent call keeps its body behind
for the worker. Every SDK has the same builder with the same nine
constructs, so a workflow written in one language reads the same in
another, and the daemon stores the same document whichever wrote it.
a support workflow
const support = workflow({ id: 'support_ticket' })
.step('classify', async (ctx, input: { ticket: string }) => ({
kind: input.ticket.startsWith('Q') ? 'question' : 'bug',
}))
.agent(
'draft',
async (ctx, input: { kind: string }) => {
// Your own model call, recorded under this step in the run's trace. Its
// usage adds up into the step's, which the daemon prices.
return await ctx.withSpan('model_call', async (span) => {
span.setModel('claude-sonnet-4', 'anthropic')
span.setUsage({ inputTokens: 812, outputTokens: 37 })
return `an answer to a ${input.kind}`
})
},
{ input: { kind: '${steps.classify.output.kind}' } },
)
.waitFor('approval', {
event: 'approved',
key: '${input.ticket}',
timeoutMs: 7 * 24 * 3_600_000,
when: when('steps.classify.output.kind', 'eq', 'bug'),
})
.step(
'send',
async (ctx, input: { text: string }) => {
if (!(await deliver(input.text, ctx.idempotencyKey))) {
throw new RetryableError('mail relay busy', { retryAfterMs: 2_000 })
}
return { sent: true }
},
{
input: { text: '${steps.draft.output}' },
effect: 'at_most_once',
retry: { maxAttempts: 3, backoff: 'exponential', initialMs: 500 },
},
)
.output({ sent: '${steps.send.output.sent}' }) support = workflow("support_ticket")
@support.step("classify")
def classify(ctx, input, prior):
return {"kind": "question" if input["ticket"].startswith("Q") else "bug"}
@support.agent("draft", input={"kind": "${steps.classify.output.kind}"})
def draft(ctx, input, prior):
# Your own model call, recorded under this step in the run's trace. What
# the span spends is the step's usage: the daemon prices it from here.
with ctx.span("model_call", kind="model_generation") as span:
span.set_model("claude-sonnet-4", "anthropic")
span.set_usage(input_tokens=812, output_tokens=37)
text = f"an answer to a {input['kind']}"
return text
support.wait_for("approval", event="approved", key="${input.ticket}",
timeout_ms=7 * 24 * 3_600_000,
when=when("steps.classify.output.kind", "eq", "bug"))
@support.step("send", input={"text": "${steps.draft.output}"}, effect="at_most_once",
retry=RetryPolicy(max_attempts=3, backoff="exponential", initial_ms=500))
def send(ctx, input, prior):
if not deliver(input["text"], ctx.idempotency_key):
raise StepFailure("mail relay busy", class_="retryable", retry_after_ms=2_000)
return {"sent": True}
support.output({"sent": "${steps.send.output.sent}"}) let support = Workflow::new("support_ticket")
.step(
"classify",
|ctx: StepContext| async move {
let input: Value = ctx.input()?;
let ticket = input["ticket"].as_str().unwrap_or("");
let kind = if ticket.starts_with('Q') { "question" } else { "bug" };
Ok::<_, StepFailure>(json!({ "kind": kind }))
},
StepOptions::new(),
)
.agent(
"draft",
|ctx: StepContext| async move {
// Your own model call, recorded under this step in the run's trace.
// Its usage is the step's usage: the daemon prices what the span spent.
let span = ctx.span("model_call", SpanKind::ModelGeneration);
span.set_model("claude-sonnet-4", "anthropic");
span.set_tokens(812, 37);
span.end();
Ok::<_, StepFailure>("an answer")
},
StepOptions::new().input(draft_input),
)
.wait_for(
"approval",
"approved",
"${input.ticket}",
Duration::from_secs(7 * 24 * 3600),
StepOptions::new().when(when("steps.classify.output.kind", "eq", "bug")),
)
.step(
"send",
|ctx: StepContext| async move {
let input: Value = ctx.input()?;
let text = input["text"].as_str().unwrap_or("");
deliver(text, ctx.idempotency_key()).await.map_err(busy)?;
Ok::<_, StepFailure>(json!({ "sent": true }))
},
StepOptions::new()
.input(json!({ "text": "${steps.draft.output}" }))
.at_most_once()
.retry(backoff),
)
.output(json!({ "sent": "${steps.send.output.sent}" })); support := mcpd.NewWorkflow("support_ticket").
Step("classify", func(ctx context.Context, step *mcpd.StepContext) (any, error) {
var in struct {
Ticket string `json:"ticket"`
}
if err := step.Input(&in); err != nil {
return nil, err
}
kind := "bug"
if strings.HasPrefix(in.Ticket, "Q") {
kind = "question"
}
return map[string]any{"kind": kind}, nil
}).
Agent("draft", func(ctx context.Context, step *mcpd.StepContext) (any, error) {
// Your own model call. A span started from ctx lands under this step,
// and what it spent is the step's usage: the daemon prices the sum.
_, span := m.Start(ctx, "model_call", mcpd.WithModel("claude-sonnet-4", "anthropic"))
defer span.End()
span.SetTokens(812, 37)
return "an answer", nil
}, mcpd.Input(map[string]any{"kind": "${steps.classify.output.kind}"})).
WaitFor("approval", "approved", "${input.ticket}", 7*24*time.Hour,
mcpd.Guard(mcpd.When("steps.classify.output.kind", "eq", "bug"))).
Step("send", func(ctx context.Context, step *mcpd.StepContext) (any, error) {
if err := deliver("the draft", step.IdempotencyKey); err != nil {
return nil, mcpd.Retryable(err, 2*time.Second)
}
return map[string]any{"sent": true}, nil
}, mcpd.Input(map[string]any{"text": "${steps.draft.output}"}), mcpd.AtMostOnce(),
mcpd.Retry(mcpd.RetryPolicy{
MaxAttempts: 3, Backoff: mcpd.BackoffExponential, Initial: 500 * time.Millisecond,
})).
Output(map[string]any{"sent": "${steps.send.output.sent}"}) val support = Workflow("support_ticket")
.step("classify") { ctx ->
val ticket = ctx.input<JsonObject>()["ticket"]!!.jsonPrimitive.content
mapOf("kind" to if (ticket.startsWith("Q")) "question" else "bug")
}
.agent("draft", StepOptions(input = mapOf("kind" to "\${steps.classify.output.kind}"))) {
ctx ->
// Your own model call, recorded under this step in the run's trace.
// Its usage is summed into the step, so the daemon prices the run.
ctx.span("model_call", SpanKind.MODEL_GENERATION) { span ->
span.setModel("claude-sonnet-4", "anthropic")
span.setUsage(inputTokens = 812, outputTokens = 37)
}
"an answer to a " + ctx.input<JsonObject>()["kind"]!!.jsonPrimitive.content
}
.waitFor(
"approval", "approved", "\${input.ticket}", 7.days,
StepOptions(guard = predicate("steps.classify.output.kind", "eq", "bug")),
)
.step(
"send",
StepOptions(
input = mapOf("text" to "\${steps.draft.output}"),
effect = Effect.AT_MOST_ONCE,
retry = RetryPolicy(
maxAttempts = 3, backoff = Backoff.EXPONENTIAL, initial = 500.milliseconds,
),
),
) { ctx ->
val text = ctx.input<JsonObject>()["text"]!!.jsonPrimitive.content
if (!deliver(text, ctx.idempotencyKey)) {
throw RetryableException("mail relay busy", retryAfter = 2.seconds)
}
mapOf("sent" to true)
}
.output(mapOf("sent" to "\${steps.send.output.sent}")) support =
Workflow.new("support_ticket")
|> Workflow.step("classify", fn _ctx, input, _prior ->
kind = if String.starts_with?(input["ticket"], "Q"), do: "question", else: "bug"
%{"kind" => kind}
end)
|> Workflow.agent(
"draft",
fn ctx, input, _prior ->
# Your own model call, recorded under this step in the run's trace. Its
# usage is the step's: the daemon prices what the trace shows.
opts = [kind: :model_generation, model: "claude-sonnet-4", provider: "anthropic",
usage: %{"input_tokens" => 812, "output_tokens" => 37}]
StepContext.span(ctx, "model_call", opts, fn -> :ok end)
"an answer to a #{input["kind"]}"
end,
input: %{"kind" => "${steps.classify.output.kind}"}
)
|> Workflow.wait_for("approval",
event: "approved",
key: "${input.ticket}",
timeout_ms: 7 * 24 * 3_600_000,
when: where("steps.classify.output.kind", :eq, "bug")
)
|> Workflow.step(
"send",
fn ctx, input, _prior ->
unless deliver.(input["text"], ctx.idempotency_key) do
raise Mcpd.StepException,
message: "mail relay busy", class: :retryable, retry_after_ms: 2_000
end
%{"sent" => true}
end,
input: %{"text" => "${steps.draft.output}"},
effect: :at_most_once,
retry: [max_attempts: 3, backoff: :exponential, initial_ms: 500]
)
|> Workflow.output(%{"sent" => "${steps.send.output.sent}"})
Three things about that example are the shape of every workflow. A step's
input is a template rendered by the daemon per attempt, so
draft reads what classify answered without either
body knowing about the other. A guard decides whether a node runs at all:
a question is sent as it is, a bug waits for a person. And the last step is
at_most_once with a retry policy, which is the combination a
side effect needs: the daemon retries a transient failure, and it never
runs a body twice without being told.
The constructs
Nine, and every SDK has all nine. Two are claimable, which means a worker serves them; the rest are the daemon's own.
| Construct | Who runs it | What it does |
|---|---|---|
task | A worker | Claimable work. The worker runs whatever code it registered under this id. |
agent | A worker | The same on the wire, recorded as an agent turn: your own loop, however many model calls, as one durable step. The trace and the interface tell it from a plain task. |
sleep | The daemon | A delay of so many milliseconds. No process is alive during it. |
sleep_until | The daemon | A delay until an instant rendered from a template, such as a billing date carried in the input. |
wait_event | The daemon | A pause until an event with this name and key arrives, or until the deadline passes. The key is rendered per run, so one wait addresses one order and no other. |
parallel | Both | Named branches, each one claimable step, all on disk before any is offered. The run moves on when the last branch ends. |
foreach | Both | One claimable step per element of the array at a path, bounded by max_items and, if you say so, by how many run at once. The outputs come back as an array in item order. |
loop | Both | A bounded loop over one claimable step. until is evaluated after each iteration and may end it early. |
subworkflow | The daemon | A child run of another definition, with its own journal. Its output becomes this step's output, and cancelling the parent cancels it. |
A branch, a foreach body and a loop body are one claimable step each. A body that needs several steps is a subworkflow, which keeps every journal flat: a run is a list, never a tree of lists.
Templates and guards
A template is JSON with $${path} references in its strings.
The daemon renders it when an attempt is created, from what the run holds
at that moment: input, the run's input;
steps.<id>, what an earlier step answered, with
.output, .status and .error;
run, the run's own id and correlation; and inside a foreach or
a loop, item, index and iter.index.
A reference that is the whole string keeps its type, so
"$${steps.classify.output}" hands the next step an object,
not a string of one.
A guard is a predicate on the same paths: eq, ne,
lt, lte, gt, gte,
in, exists and matches, combined with
all, any and not. A guard that is
false writes the node as skipped and the run moves on. A path
that is absent makes every comparison false rather than raising, because a
run that dies on a missing field is a run nobody can finish.
The daemon validates the document when it is stored. A guard that names a
path nothing can produce, a template that reads a step declared later, a
foreach over something that is not a path, an id that does not match
^[a-z][a-z0-9_]*$: each is refused with a 400 naming the
field. Unknown fields are refused too, so a guard spelled wrong is an error
and not a run that skips its guard.
Every duration has a ceiling as well as a floor. run_timeout_ms,
step_timeout_ms, a retry's initial_ms and
max_ms, a sleep's ms and a wait's
timeout_ms are each at most 315360000000, which
is 3650 days, and a longer one is refused with a 400 naming the field, the
ceiling and the value you sent. A deadline further out than that would not
fit the clock the daemon stores it on.
Retries, effects and failure
A body that fails tells the daemon how, with a class. retryable
is worth another attempt, and the daemon makes one while the node's
retry budget lasts, at the node's backoff, waiting at least
the retry_after_ms the body asked for, which is held to the
same 3650-day ceiling as every other duration. fatal and
invalid_input are not retried. unservable means
this worker does not hold the definition or the step, and the attempt goes
back to the queue for another worker without spending the budget. A body
that throws anything else is fatal.
max_attempts counts every attempt including the first, so the
default of one means no retry. After the budget, on_error
decides: fail_run, the default, ends the run with the step's
message as its error; continue marks the step failed and
moves on, so a later guard can read steps.<id>.status
and route around it.
effect says what a duplicated body would cost. Under
at_least_once, the default, a lease that is lost requeues the
attempt and the body may run again. Under at_most_once the
worker must post started before the body, and a started
attempt whose lease is lost fails rather than runs again. Use it for the
step that sends the email or charges the card, and give that body the
lease's idempotency_key to hand to the third party: it is
stable across retries of the same step, so a retry after a lost reply is
recognised on the far side too.
Versions
A definition is immutable once stored. Post the same body again and you get the same version back; post a different body under a version that exists and the daemon refuses with a 409, because a run that pinned that version would otherwise change under it. Leave the version out and the daemon assigns the next number, which is what a worker does at start-up. A run pins the version it started with and never re-resolves it, so a deploy that stores version 3 finishes the runs on version 2 as they were written.
Limits
A definition may declare max_steps, run_timeout_ms
and max_cost_usd. The daemon counts every attempt against the
first, the run's age against the second, and the priced usage the steps
report against the third, and ends the run timed_out or
failed at the ceiling with the ceiling named as the error. A
loop that will not converge and an agent step that will not stop spending
are the same problem to the daemon, and the same flag stops both.
What the daemon keeps
A run's input, its state, its output and every step's input and output are kept in the run store exactly as they were sent. They are not redacted: they are the workflow's own working state, a later step reads them back, and a value the redactor had rewritten would change what the run does. The spans a run writes into the trace store are redacted like any other span. Retention and erasure reach the run store as well, for runs that have ended. Runs says exactly what each one deletes.
Next
- Serving the bodies, and the protocol a worker speaks: Workers.
- Starting, watching, signalling and cancelling a run: Runs.
- The document itself, field by field: definition.schema.json, and the binding contract in run-contract-v1.md.