İş akışları
Bir iş akışı, daemon'un tuttuğu veri ile daemon'un hiç görmediği gövdelerdir. Bu sayfa veri kısmıdır: bir tanım neler söyleyebilir ve her yapı daemon'a ne yaptırır.
İş bölümü
Kalıcı bir çalıştırmanın iki sahibi vardır. Adımlar arasında çalıştırma daemon'undur: imleç, günlük, yeniden deneme bütçesi, zamanlayıcılar ve beklemeler. Bir adımın içindeki her şey sizin sürecinizindir: çağırdığınız model, gönderdiğiniz prompt, çalıştırdığınız araçlar, elinizdeki kimlik bilgileri. Bunların hiçbiri daemon'a verilmez ve daemon sizin adınıza hiçbir zaman model çalıştırmaz.
Bu yüzden bir iş akışı tek bir dosyadan iki parça hâlinde çıkar. Graf, daemon'a bir tanım olarak, yani daemon'un yürüdüğü bir belge olarak gider. Adım gövdeleri, hangi dilde yazdıysanız o dilde sizin sürecinizde kalır ve onları bir worker sunar. Daemon'a dağıtılması gereken bir şey yoktur ve bir gövde, tanım hiç değişmeden değişebilir.
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.
İş akışı yazmak
Builder her iki parçayı da üretir. document() tanımdır; her
step ya da agent çağrısı da gövdesini worker için
geride bırakır. Her SDK'da aynı dokuz yapıya sahip aynı builder vardır, yani
bir dilde yazılmış bir iş akışı başka bir dilde de aynı okunur ve hangisi
yazmış olursa olsun daemon aynı belgeyi saklar.
bir destek iş akışı
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}"})
O örnekteki üç şey, her iş akışının biçimini verir. Bir adımın girdisi,
daemon'un her deneme için işlediği bir şablondur; böylece
draft, iki gövde birbirinden habersizken classify
ne yanıtladıysa onu okur. Bir
guard, bir düğümün hiç çalışıp çalışmayacağına karar verir: bir soru olduğu
gibi gönderilir, bir hata bir insanı bekler. Son adım ise yeniden deneme
politikası olan bir at_most_once adımıdır; bir yan etkinin
ihtiyaç duyduğu birleşim de budur: daemon geçici bir hatayı yeniden dener ve
söylenmedikçe bir gövdeyi asla iki kez çalıştırmaz.
Yapılar
Dokuz tane ve her SDK'da dokuzu da var. İkisi talep edilebilir, yani onları bir worker sunar; geri kalanı daemon'un kendi işidir.
| Yapı | Kim çalıştırır | Ne yapar |
|---|---|---|
task | Bir worker | Talep edilebilir iş. Worker, bu kimlik altında kaydettiği kodu çalıştırır. |
agent | Bir worker | Hat üzerinde aynısı, bir ajan turu olarak kaydedilir: kendi döngünüz, kaç model çağrısı olursa olsun, tek bir kalıcı adım olarak. Trace ve arayüz bunu düz bir görevden ayırt eder. |
sleep | Daemon | Şu kadar milisaniyelik gecikme. O sırada ayakta hiçbir süreç yoktur. |
sleep_until | Daemon | Bir şablondan işlenen bir ana kadar gecikme; örneğin girdide taşınan bir fatura tarihine kadar. |
wait_event | Daemon | Bu ad ve anahtarla bir olay gelene ya da süre sonu geçene kadar duraklama. Anahtar her çalıştırma için işlenir, yani bir bekleme tek bir siparişi hedefler, başkasını değil. |
parallel | İkisi de | Adlandırılmış dallar; her biri talep edilebilir tek bir adım, herhangi biri sunulmadan önce hepsi diskte. Son dal bittiğinde çalıştırma ilerler. |
foreach | İkisi de | Bir yoldaki dizinin her öğesi için talep edilebilir bir adım; max_items ile ve siz söylerseniz aynı anda kaç tanesinin çalışacağıyla sınırlanır. Çıktılar öğe sırasına göre bir dizi olarak döner. |
loop | İkisi de | Talep edilebilir tek bir adım üzerinde sınırlı bir döngü. until her yinelemeden sonra değerlendirilir ve döngüyü erken bitirebilir. |
subworkflow | Daemon | Başka bir tanımın, kendi günlüğü olan bir alt çalıştırması. Çıktısı bu adımın çıktısı olur ve üst çalıştırmayı iptal etmek onu da iptal eder. |
Bir dal, bir foreach gövdesi ve bir loop gövdesi,
her biri talep edilebilir tek bir adımdır. Birkaç adıma ihtiyaç duyan bir
gövde bir alt iş akışıdır; her günlüğü düz tutan da budur: bir çalıştırma bir
listedir, hiçbir zaman listelerden oluşan bir ağaç değil.
Şablonlar ve guard'lar
Bir şablon, dizgilerinde $${path} başvuruları olan JSON'dur.
Daemon onu, bir deneme oluşturulduğunda, çalıştırmanın o andaki içeriğinden
işler: input, çalıştırmanın girdisi;
steps.<id>, daha önceki bir adımın verdiği yanıt,
.output, .status ve .error ile;
run, çalıştırmanın kendi kimliği ve ilişkilendirmesi; ve bir
foreach ya da loop içinde item,
index ve iter.index. Dizginin tamamı olan bir
başvuru türünü korur, yani
"$${steps.classify.output}" bir sonraki adıma bir nesne
verir, onun dizgiye çevrilmiş hâlini değil.
Bir guard, aynı yollar üzerinde bir yüklemdir: eq,
ne, lt, lte, gt,
gte, in, exists ve
matches; bunlar all, any ve
not ile birleşir. Yanlış çıkan bir guard düğümü
skipped olarak yazar ve çalıştırma ilerler. Bulunmayan bir yol,
hata fırlatmak yerine her karşılaştırmayı yanlış yapar; çünkü eksik bir alan
yüzünden ölen bir çalıştırma, kimsenin bitiremeyeceği bir çalıştırmadır.
Daemon belgeyi saklarken doğrular. Hiçbir şeyin üretemeyeceği bir yolu
adlandıran bir guard, sonradan bildirilen bir adımı okuyan bir şablon, bir
yol olmayan bir şey üzerinde foreach,
^[a-z][a-z0-9_]*$ ile eşleşmeyen bir kimlik: her biri, alanı
adlandıran bir 400 ile reddedilir. Bilinmeyen alanlar da reddedilir, yani
yanlış yazılmış bir guard bir hatadır, guard'ını atlayan bir çalıştırma
değil.
Her sürenin bir tabanı olduğu gibi bir tavanı da vardır.
run_timeout_ms, step_timeout_ms, bir yeniden
denemenin initial_ms ve max_ms değerleri, bir
uykunun ms değeri ve bir beklemenin timeout_ms
değeri, her biri en çok 315360000000, yani 3650 gündür; daha
uzun olanı, alanı, tavanı ve gönderdiğiniz değeri adlandıran bir 400 ile
reddedilir. Bundan daha ileri bir süre sonu, daemon'un onu sakladığı saate
sığmazdı.
Yeniden denemeler, etkiler ve başarısızlık
Başarısız olan bir gövde, daemon'a bunun nasıl olduğunu bir sınıfla söyler.
retryable, bir deneme daha yapmaya değer demektir; düğümün
retry bütçesi sürdükçe daemon, düğümün beklemesine uyarak ve en
az gövdenin istediği retry_after_ms kadar bekleyerek bir deneme
daha yapar; bu değer de diğer her süreyle aynı 3650 günlük tavana bağlıdır.
fatal ve invalid_input yeniden denenmez.
unservable, bu worker'ın tanımı ya da adımı tutmadığı anlamına
gelir; deneme, bütçeyi harcamadan başka bir worker için kuyruğa
döner. Başka bir şey fırlatan bir gövde fatal sayılır.
max_attempts ilki dahil her denemeyi sayar, yani varsayılan olan
bir, hiç yeniden deneme yok demektir. Bütçe bittikten sonra
on_error karar verir: varsayılan olan fail_run,
adımın mesajını hata olarak alıp çalıştırmayı bitirir; continue
ise adımı başarısız olarak işaretler ve devam eder, böylece sonraki bir guard
steps.<id>.status değerini okuyup etrafından dolaşabilir.
effect, tekrarlanmış bir gövdenin neye mal olacağını söyler.
Varsayılan olan at_least_once altında, kaybedilen bir lease
denemeyi kuyruğa geri koyar ve gövde yeniden çalışabilir.
at_most_once altında worker gövdeden önce started
göndermek zorundadır ve lease'i kaybedilen, başlamış bir deneme yeniden
çalışmak yerine başarısız olur. Bunu e-postayı gönderen ya da karttan çeken
adım için kullanın ve o gövdeye, üçüncü tarafa iletmesi için lease'in
idempotency_key değerini verin: aynı adımın yeniden denemeleri
boyunca sabittir, yani kaybolan bir yanıttan sonraki yeniden deneme karşı
tarafta da tanınır.
Sürümler
Bir tanım saklandıktan sonra değişmezdir. Aynı gövdeyi yeniden gönderirseniz aynı sürümü geri alırsınız; var olan bir sürüm altında farklı bir gövde gönderirseniz daemon 409 ile reddeder, çünkü aksi hâlde o sürümü sabitlemiş bir çalıştırmanın altından tanım değişirdi. Sürümü yazmazsanız daemon bir sonraki numarayı verir; bir worker'ın açılışta yaptığı da budur. Bir çalıştırma başladığı sürümü sabitler ve onu bir daha çözmez, yani sürüm 3'ü saklayan bir dağıtım, sürüm 2 üzerindeki çalıştırmaları yazıldıkları gibi bitirir.
Sınırlar
Bir tanım max_steps, run_timeout_ms ve
max_cost_usd bildirebilir. Daemon her denemeyi birinciye,
çalıştırmanın yaşını ikinciye, adımların bildirdiği fiyatlanmış kullanımı da
üçüncüye sayar; tavanda çalıştırmayı timed_out ya da
failed olarak bitirir ve hata olarak tavanın adını verir.
Yakınsamayan bir döngü ile harcamayı kesmeyen bir ajan adımı daemon için aynı
sorundur ve ikisini de aynı ayar durdurur.
Daemon'un sakladıkları
Bir çalıştırmanın girdisi, durum verisi, çıktısı ve her adımın girdisi ve çıktısı, çalıştırma deposunda gönderildikleri gibi tutulur. Maskelenmezler: bunlar iş akışının kendi çalışma durumudur, sonraki bir adım onları geri okur ve maskeleyicinin yeniden yazdığı bir değer çalıştırmanın yaptığı şeyi değiştirirdi. Bir çalıştırmanın trace deposuna yazdığı span'ler, diğer her span gibi maskelenir. Saklama ve silme, sona ermiş çalıştırmalar için çalıştırma deposunu da kapsar. Her birinin tam olarak neyi sildiğini Çalıştırmalar sayfası söyler.
Sonraki
- Gövdeleri sunmak ve bir worker'ın konuştuğu protokol: Worker'lar.
- Bir çalıştırmayı başlatmak, izlemek, sinyal göndermek ve iptal etmek: Çalıştırmalar.
- Belgenin kendisi, alan alan: definition.schema.json, bir de bağlayıcı sözleşme: run-contract-v1.md.