Workflows
Un workflow, ce sont des données que le daemon détient et des corps que le daemon ne voit jamais. Cette page traite des données : ce qu’une définition peut dire, et ce que chaque construction fait faire au daemon.
La répartition du travail
Une exécution durable a deux propriétaires. Le daemon possède l’exécution entre les étapes : le curseur, le journal, le budget de nouvelles tentatives, les timers et les attentes. Votre processus possède tout ce qui se passe à l’intérieur d’une étape : le modèle que vous appelez, le prompt que vous envoyez, les outils que vous exécutez, les secrets que vous détenez. Rien de tout cela n’est remis au daemon, et il n’exécute jamais un modèle à votre place.
Un workflow arrive donc en deux parties depuis un seul fichier. Le graphe va au daemon sous forme de définition, un document que le daemon parcourt. Les corps d’étape restent dans votre processus, dans le langage où vous les avez écrits, pour qu’un worker les serve. Il n’y a rien à déployer sur le daemon, et un corps peut changer sans que la définition change en quoi que ce soit.
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.
En écrire un
Le builder produit les deux parties. document() est la
définition, et chaque appel à step ou à agent
garde son corps de côté pour le worker. Chaque SDK a le même builder avec
les mêmes neuf constructions : un workflow écrit dans un langage se lit de
la même façon dans un autre, et le daemon stocke le même document quel que
soit celui qui l’a écrit.
un workflow de support
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}"})
Trois traits de cet exemple donnent la forme de tout workflow. L’entrée d’une
étape est un template que le daemon évalue à chaque tentative, si bien que
draft lit ce que classify a répondu sans qu’aucun
des deux corps connaisse l’autre. Une garde décide si un nœud s’exécute :
une question part telle quelle, un bug attend une personne. Et la dernière
étape est en at_most_once avec une politique de nouvelles
tentatives, la combinaison qu’exige un effet de bord : le daemon réessaie
après une défaillance passagère, et il n’exécute jamais un corps deux fois
sans qu’on le lui ait dit.
Les constructions
Neuf, et chaque SDK les a toutes les neuf. Deux sont réclamables, c’est-à-dire servies par un worker ; les autres appartiennent au daemon.
| Construction | Qui l’exécute | Ce qu’elle fait |
|---|---|---|
task | Un worker | Du travail réclamable. Le worker exécute le code qu’il a enregistré sous cet identifiant. |
agent | Un worker | La même chose sur le réseau, consignée comme un tour d’agent : votre propre boucle, avec autant d’appels de modèle qu’il faut, comme une seule étape durable. La trace et l’interface la distinguent d’une tâche ordinaire. |
sleep | Le daemon | Un délai d’un certain nombre de millisecondes. Aucun processus n’est vivant pendant ce temps. |
sleep_until | Le daemon | Un délai jusqu’à un instant évalué à partir d’un template, par exemple une date de facturation portée par l’entrée. |
wait_event | Le daemon | Une pause jusqu’à l’arrivée d’un événement portant ce nom et cette clé, ou jusqu’au passage de l’échéance. La clé est évaluée par exécution, si bien qu’une attente vise une commande et une seule. |
parallel | Les deux | Des branches nommées, chacune une étape réclamable, toutes sur disque avant qu’aucune ne soit proposée. L’exécution repart quand la dernière branche se termine. |
foreach | Les deux | Une étape réclamable par élément du tableau situé à un chemin, bornée par max_items et, si vous le demandez, par le nombre d’étapes simultanées. Les sorties reviennent en tableau, dans l’ordre des éléments. |
loop | Les deux | Une boucle bornée sur une étape réclamable. until est évalué après chaque itération et peut y mettre fin plus tôt. |
subworkflow | Le daemon | Une exécution fille d’une autre définition, avec son propre journal. Sa sortie devient la sortie de cette étape, et annuler le parent l’annule. |
Une branche, un corps de foreach et un corps de loop sont chacun une étape réclamable. Un corps qui a besoin de plusieurs étapes est un subworkflow, ce qui garde chaque journal plat : une exécution est une liste, jamais un arbre de listes.
Templates et gardes
Un template est du JSON portant des références $${path}
dans ses chaînes. Le daemon l’évalue à la création d’une tentative, à
partir de ce que l’exécution détient à cet instant : input,
l’entrée de l’exécution ; steps.<id>, ce qu’une étape
antérieure a répondu, avec .output, .status et
.error ; run, l’identifiant et la corrélation de
l’exécution ; et, à l’intérieur d’un foreach ou d’un loop,
item, index et iter.index. Une
référence qui occupe toute la chaîne garde son type, si bien que
"$${steps.classify.output}" remet à l’étape suivante un
objet, et non une chaîne qui en contient un.
Une garde est un prédicat sur les mêmes chemins : eq,
ne, lt, lte, gt,
gte, in, exists et
matches, combinés par all, any et
not. Une garde fausse écrit le nœud comme skipped
et l’exécution repart. Un chemin absent rend toute comparaison fausse au
lieu de lever une erreur, parce qu’une exécution qui meurt sur un champ
manquant est une exécution que personne ne peut mener à terme.
Le daemon valide le document au moment où il est stocké. Une garde qui nomme
un chemin que rien ne peut produire, un template qui lit une étape déclarée
plus loin, un foreach sur autre chose qu’un chemin, un identifiant qui ne
respecte pas ^[a-z][a-z0-9_]*$ : chacun est refusé avec un 400
qui nomme le champ. Les champs inconnus sont refusés eux aussi, si bien
qu’une garde mal orthographiée est une erreur et non une exécution qui saute
sa garde.
Chaque durée a un plafond aussi bien qu’un plancher.
run_timeout_ms, step_timeout_ms, les
initial_ms et max_ms d’une politique de nouvelles
tentatives, le ms d’un sleep et le timeout_ms
d’une attente valent chacun au plus 315360000000, soit
3650 jours, et une valeur plus longue est refusée avec un 400 qui nomme le
champ, le plafond et la valeur que vous avez envoyée. Une échéance plus
lointaine ne tiendrait pas dans l’horloge sur laquelle le daemon la stocke.
Nouvelles tentatives, effets et échec
Un corps qui échoue dit au daemon comment, avec une classe.
retryable justifie une nouvelle tentative, et le daemon en fait
une tant que dure le budget retry du nœud, en respectant le
backoff du nœud et en attendant au moins le retry_after_ms que
le corps a demandé, lui-même soumis au même plafond de 3650 jours que toute
autre durée. fatal et invalid_input ne sont pas
réessayés. unservable veut dire que ce worker ne détient
pas la définition ou l’étape, et la tentative retourne dans la file pour un
autre worker sans dépenser le budget. Un corps qui lève autre chose est
fatal.
max_attempts compte toutes les tentatives, y compris la
première : la valeur par défaut de un veut donc dire aucune nouvelle
tentative. Une fois le budget épuisé, on_error tranche.
fail_run, la valeur par défaut, met fin à l’exécution avec le
message de l’étape comme erreur ; continue marque l’étape en
échec et repart, de sorte qu’une garde ultérieure peut lire
steps.<id>.status et la contourner.
effect dit ce que coûterait un corps exécuté en double. Sous
at_least_once, la valeur par défaut, un bail perdu remet la
tentative en file et le corps peut tourner une nouvelle fois. Sous
at_most_once, le worker doit poster started avant
le corps, et une tentative démarrée dont le bail est perdu échoue au lieu de
tourner à nouveau. Réservez-le à l’étape qui envoie l’e-mail ou débite la
carte, et donnez à ce corps l’idempotency_key du bail à
transmettre au tiers : elle est stable d’une tentative à l’autre de la même
étape, si bien qu’une nouvelle tentative après une réponse perdue est
reconnue de l’autre côté aussi.
Les versions
Une définition est immuable une fois stockée. Postez le même corps à nouveau et vous récupérez la même version ; postez un corps différent sous une version qui existe et le daemon refuse avec un 409, car une exécution qui a épinglé cette version verrait sinon sa définition changer. Omettez la version et le daemon attribue le numéro suivant, ce que fait un worker au démarrage. Une exécution épingle la version avec laquelle elle a commencé et ne la résout jamais à nouveau : un déploiement qui stocke la version 3 termine les exécutions en version 2 telles qu’elles avaient été écrites.
Les limites
Une définition peut déclarer max_steps,
run_timeout_ms et max_cost_usd. Le daemon impute
chaque tentative à la première, l’âge de l’exécution à la deuxième, et la
consommation tarifée que les étapes déclarent à la troisième, puis il
termine l’exécution en timed_out ou en failed au
plafond, en nommant le plafond comme erreur. Une
boucle qui ne converge pas et une étape d’agent qui ne cesse pas de dépenser
sont le même problème pour le daemon, et la même limite arrête les deux.
Ce que le daemon garde
L’entrée d’une exécution, son état, sa sortie, et l’entrée et la sortie de chaque étape sont gardés dans le stockage des exécutions exactement tels qu’ils ont été envoyés. Ils ne sont pas masqués : ce sont les données de travail du workflow lui-même, une étape ultérieure les relit, et une valeur que le masquage aurait réécrite changerait ce que fait l’exécution. Les spans qu’une exécution écrit dans le stockage des traces sont masqués comme tout autre span. La conservation et l’effacement atteignent aussi le stockage des exécutions, pour les exécutions terminées. Exécutions dit exactement ce que chacun supprime.
Ensuite
- Servir les corps, et le protocole que parle un worker : Workers.
- Démarrer, observer, signaler et annuler une exécution : Exécutions.
- Le document lui-même, champ par champ : definition.schema.json, et le contrat qui l’engage dans run-contract-v1.md.