Workflow
Un workflow è dati che il daemon tiene e corpi che il daemon non vede mai. Questa pagina sono i dati: che cosa può dire una definizione, e che cosa ogni costrutto fa fare al daemon.
La divisione del lavoro
Un’esecuzione durevole ha due proprietari. Il daemon possiede l’esecuzione fra un passo e l’altro: il cursore, il diario, il budget dei nuovi tentativi, i timer e le attese. Il tuo processo possiede tutto quello che sta dentro un passo: il modello che chiami, il prompt che mandi, gli strumenti che usi, le credenziali che tieni. Al daemon non viene consegnato niente di tutto questo, e non esegue mai un modello per tuo conto.
Così un workflow arriva in due parti da un solo file. Il grafo va al daemon come definizione, un documento che il daemon percorre. I corpi dei passi restano nel tuo processo, nel linguaggio in cui li hai scritti, perché li serva un worker. Non c’è niente di cui fare il deploy sul daemon, e un corpo può cambiare senza che la definizione cambi affatto.
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.
Scriverne uno
Il builder produce tutte e due le parti. document() è la
definizione, e ogni chiamata a step o a agent
trattiene il proprio corpo per il worker. Ogni SDK ha lo stesso builder con
gli stessi nove costrutti, così un workflow scritto in un linguaggio si
legge uguale in un altro, e il daemon archivia lo stesso documento
qualunque linguaggio l’abbia scritto.
un workflow di assistenza
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}"})
Tre cose di quell’esempio sono la forma di ogni workflow. L’input di un
passo è un template che il daemon rende a ogni tentativo, così
draft legge quello che ha risposto classify senza
che nessuno dei due corpi sappia dell’altro. Una guardia decide se un nodo
gira o no: una domanda viene mandata così com’è, un bug aspetta una persona.
E l’ultimo passo è at_most_once con una politica di nuovi
tentativi, che è la combinazione di cui ha bisogno un effetto collaterale:
il daemon riprova dopo un guasto passeggero, e non esegue mai un corpo due
volte senza che gli sia stato detto.
I costrutti
Nove, e ogni SDK li ha tutti e nove. Due si prendono in carico, cioè li serve un worker; gli altri sono del daemon.
| Costrutto | Chi lo esegue | Che cosa fa |
|---|---|---|
task | Un worker | Lavoro da prendere in carico. Il worker esegue il codice che ha registrato sotto questo id. |
agent | Un worker | La stessa cosa a livello di protocollo, registrata come un turno di agente: il tuo ciclo, con quante chiamate al modello vuoi, come un solo passo durevole. La traccia e l’interfaccia lo distinguono da un task semplice. |
sleep | Il daemon | Un ritardo di tanti millisecondi. Durante il ritardo non è vivo nessun processo. |
sleep_until | Il daemon | Un ritardo fino a un istante reso da un template, per esempio una data di fatturazione che arriva nell’input. |
wait_event | Il daemon | Una pausa finché non arriva un evento con questo nome e questa chiave, oppure finché non passa la scadenza. La chiave è resa per ogni esecuzione, così un’attesa si rivolge a un solo ordine e a nessun altro. |
parallel | Tutti e due | Rami con un nome, ognuno un passo da prendere in carico, tutti su disco prima che ne venga offerto uno. L’esecuzione va avanti quando finisce l’ultimo ramo. |
foreach | Tutti e due | Un passo da prendere in carico per ogni elemento dell’array a un percorso, limitato da max_items e, se lo dici, da quanti ne girano insieme. Gli output tornano come array nell’ordine degli elementi. |
loop | Tutti e due | Un ciclo limitato su un solo passo da prendere in carico. until è valutato dopo ogni giro e lo può chiudere in anticipo. |
subworkflow | Il daemon | Un’esecuzione figlia di un’altra definizione, con un diario suo. Il suo output diventa l’output di questo passo, e annullare il genitore annulla anche questa. |
Un ramo, un corpo di foreach e un corpo di loop sono un passo da prendere in carico ciascuno. Un corpo che ha bisogno di più passi è un subworkflow, ed è questo che tiene piatto ogni diario: un’esecuzione è una lista, mai un albero di liste.
Template e guardie
Un template è JSON con riferimenti $${path} dentro le sue
stringhe. Il daemon lo rende quando crea un tentativo, a partire da quello
che l’esecuzione tiene in quel momento: input, l’input
dell’esecuzione; steps.<id>, quello che ha risposto un
passo precedente, con .output, .status e
.error; run, l’id e la correlazione
dell’esecuzione stessa; e dentro un foreach o un loop, item,
index e iter.index. Un riferimento che occupa
tutta la stringa conserva il proprio tipo, così
"$${steps.classify.output}" consegna al passo successivo un
oggetto, non una stringa che ne contiene uno.
Una guardia è un predicato sugli stessi percorsi: eq,
ne, lt, lte, gt,
gte, in, exists e
matches, messi insieme con all, any e
not. Una guardia falsa scrive il nodo come
skipped e l’esecuzione va avanti. Un percorso assente rende
falso ogni confronto invece di sollevare un errore, perché un’esecuzione che
muore su un campo mancante è un’esecuzione che nessuno può portare a
termine.
Il daemon convalida il documento quando lo archivia. Una guardia che nomina
un percorso che nessuno può produrre, un template che legge un passo
dichiarato più tardi, un foreach su qualcosa che non è un percorso, un
id che non corrisponde a ^[a-z][a-z0-9_]*$: ognuno
di questi è rifiutato con un 400 che nomina il campo. Anche i campi
sconosciuti sono rifiutati, così una guardia scritta male è un errore e non
un’esecuzione che salta la sua guardia.
Ogni durata ha un tetto oltre che un minimo. run_timeout_ms,
step_timeout_ms, l’initial_ms e il
max_ms di un nuovo tentativo, il ms di uno sleep
e il timeout_ms di un’attesa valgono ciascuno al massimo
315360000000, cioè 3650 giorni, e uno più lungo viene
rifiutato con un 400 che nomina il campo, il tetto e il valore che hai
mandato. Una scadenza più lontana di così non entrerebbe nell’orologio su
cui il daemon la archivia.
Nuovi tentativi, effetti e fallimento
Un corpo che fallisce dice al daemon come, con una classe.
retryable vale un altro tentativo, e il daemon ne fa uno
finché dura il budget retry del nodo, con il backoff del nodo,
aspettando almeno il retry_after_ms che il corpo ha chiesto,
che è tenuto allo stesso tetto di 3650 giorni di ogni altra durata.
fatal e invalid_input non vengono ritentati.
unservable vuol dire che questo worker non tiene la definizione
o il passo, e il tentativo torna in coda per un altro worker senza consumare
il budget. Un corpo che solleva qualunque altra cosa è fatal.
max_attempts conta ogni tentativo compreso il primo, quindi il
valore predefinito di uno vuol dire nessun nuovo tentativo. Finito il
budget, decide on_error: fail_run, il valore
predefinito, chiude l’esecuzione con il messaggio del passo come errore;
continue segna il passo come fallito e va avanti, così una
guardia successiva può leggere steps.<id>.status e
aggirarlo.
effect dice quanto costerebbe un corpo eseguito due volte.
Sotto at_least_once, il valore predefinito, un lease perso
rimette il tentativo in coda e il corpo può girare di nuovo. Sotto
at_most_once il worker deve scrivere started prima
del corpo, e un tentativo già partito che perde il lease fallisce invece di
girare di nuovo. Usalo per il passo che manda l’email o addebita la carta, e
dai a quel corpo l’idempotency_key del lease da passare alla
terza parte: è stabile fra i nuovi tentativi dello stesso passo, così un
tentativo dopo una risposta persa viene riconosciuto anche dall’altro lato.
Le versioni
Una definizione è immutabile una volta archiviata. Manda di nuovo lo stesso corpo e ti torna indietro la stessa versione; manda un corpo diverso sotto una versione che esiste già e il daemon rifiuta con un 409, perché altrimenti un’esecuzione che ha fissato quella versione se la vedrebbe cambiare sotto i piedi. Lascia fuori la versione e il daemon assegna il numero successivo, che è quello che fa un worker all’avvio. Un’esecuzione fissa la versione con cui è partita e non la risolve mai una seconda volta, così un deploy che archivia la versione 3 porta a termine le esecuzioni sulla versione 2 come erano state scritte.
I limiti
Una definizione può dichiarare max_steps,
run_timeout_ms e max_cost_usd. Il primo conta ogni
tentativo, il secondo l’età dell’esecuzione, il terzo il consumo valorizzato
che i passi riferiscono; al tetto il daemon chiude l’esecuzione come
timed_out o failed, con il tetto nominato come
errore. Un loop che non converge e un passo di agente che non smette di
spendere sono lo stesso problema per il daemon, e lo stesso flag li ferma
tutti e due.
Che cosa tiene il daemon
L’input di un’esecuzione, il suo stato, il suo output e l’input e l’output di ogni passo sono tenuti nello storage delle esecuzioni esattamente come sono stati mandati. Non sono oscurati: sono lo stato di lavoro del workflow stesso, un passo successivo li rilegge, e un valore che l’oscuratore avesse riscritto cambierebbe quello che fa l’esecuzione. Gli span che un’esecuzione scrive nello storage delle tracce sono oscurati come ogni altro span. Anche la conservazione e la cancellazione raggiungono lo storage delle esecuzioni, per le esecuzioni finite. Esecuzioni dice esattamente che cosa cancella ciascuna.
Poi
- Servire i corpi, e il protocollo che parla un worker: Worker.
- Avviare, seguire, segnalare e annullare un’esecuzione: Esecuzioni.
- Il documento in sé, campo per campo: definition.schema.json, e il contratto vincolante in run-contract-v1.md.