ワークフロー

役割の分担

永続実行には、持ち主が 2 つあります。ステップとステップの間の実行はデーモンが持ちます。カーソル、ジャーナル、リトライの予算、タイマー、待機です。ステップの中身はすべてあなたのプロセスが持ちます。呼び出すモデル、送るプロンプト、動かすツール、保持している資格情報です。デーモンにはそのどれも渡らず、デーモンがあなたの代わりにモデルを動かすこともありません。

そのためワークフローは、1 つのファイルから 2 つの部分になって届きます。グラフは定義としてデーモンに渡り、デーモンはその文書をたどります。ステップの本体は、あなたが書いた言語のまま、あなたのプロセスに残り、ワーカーがそれを処理します。デーモンにデプロイするものは何もなく、定義をまったく変えずに本体だけを変えられます。

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.

書き方

ビルダーは両方の部分を作ります。document() が定義で、 step や agent の呼び出しはそれぞれの本体をワーカーのために手元に残します。どの SDK にも同じ 9 つの構成要素を持つ同じビルダーがあるので、ある言語で書いたワークフローは別の言語でも同じように読め、どの言語で書いてもデーモンが保存する文書は同じです。

サポートのワークフロー

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}"})

この例に見える 3 つの点が、どのワークフローにも共通する形です。ステップの入力は試行ごとにデーモンが展開するテンプレートなので、draft は classify が答えたものを読めますが、どちらの本体も相手を知りません。ガードはノードをそもそも実行するかどうかを決めます。質問はそのまま送られ、不具合は人を待ちます。そして最後のステップは at_most_once にリトライ方針を組み合わせたもので、これが副作用のあるステップに必要な組み合わせです。デーモンは一時的な失敗をリトライしますが、指示されない限り本体を二度実行することはありません。

構成要素

9 つあり、どの SDK にも 9 つすべてがあります。そのうち 2 つは取得可能で、つまりワーカーが処理します。残りはデーモン自身のものです。

構成要素	実行する側	すること
`task`	ワーカー	取得可能な作業。ワーカーは、この id で登録したコードを実行します。
`agent`	ワーカー	通信の上では同じで、エージェントのターンとして記録されます。あなた自身のループが、モデル呼び出しが何回あっても、1 つの永続ステップになります。トレースとインターフェースは、これを素の task と区別します。
`sleep`	デーモン	指定したミリ秒だけの遅延。その間、生きているプロセスはありません。
`sleep_until`	デーモン	テンプレートから展開された時刻までの遅延。入力に入っている請求日などです。
`wait_event`	デーモン	この名前とキーを持つイベントが届くまで、または期限を過ぎるまでの一時停止。キーは実行ごとに展開されるので、1 つの待機は 1 つの注文だけを指し、他は指しません。
`parallel`	両方	名前付きの分岐。それぞれが 1 つの取得可能なステップで、どれかが提示される前にすべてがディスクに載ります。最後の分岐が終わると実行は先へ進みます。
`foreach`	両方	あるパスにある配列の要素ごとに 1 つの取得可能なステップ。`max_items` で上限を決め、指定すれば同時に走る数でも制限できます。出力は要素の順に並んだ配列として返ります。
`loop`	両方	1 つの取得可能なステップに対する、上限付きのループ。`until` は各反復のあとに評価され、途中で終わらせることがあります。
`subworkflow`	デーモン	別の定義による子の実行で、自分のジャーナルを持ちます。その出力がこのステップの出力になり、親をキャンセルすると子もキャンセルされます。

分岐、foreach の本体、loop の本体は、それぞれ 1 つの取得可能なステップです。複数のステップが必要な本体は subworkflow にします。そうすることでどのジャーナルも平らなままです。実行はリストであり、決してリストの木にはなりません。

テンプレートとガード

テンプレートとは、文字列の中に $${path} 参照を持つ JSON です。デーモンは試行が作られた時点で、その時点の実行が持つものから展開します。 input は実行の入力、steps.<id> は先のステップが答えたもので、.output、.status、.error を持ちます。run は実行自身の id と相関です。foreach や loop の中では item、index、iter.index が使えます。文字列全体が参照であるときは型が保たれるので、 "$${steps.classify.output}" は次のステップに、文字列ではなくオブジェクトを渡します。

ガードは同じパスに対する述語です。eq、ne、 lt、lte、gt、gte、 in、exists、matches があり、 all、any、not で組み合わせます。偽になったガードは、そのノードを skipped として書き、実行は先へ進みます。存在しないパスは、例外を投げるのではなくすべての比較を偽にします。フィールドがないだけで死ぬ実行は、誰にも終わらせられない実行だからです。

デーモンは文書を保存するときに検証します。何も生み出せないパスを名指しするガード、あとで宣言されるステップを読むテンプレート、パスではないものに対する foreach、^[a-z][a-z0-9_]*$ に合わない id。いずれもフィールドを名指しした 400 で拒否されます。未知のフィールドも拒否されるので、綴りを間違えたガードはエラーになり、ガードを飛ばして走る実行にはなりません。

どの期間にも、下限だけでなく上限があります。run_timeout_ms、 step_timeout_ms、リトライの initial_ms と max_ms、スリープの ms、待機の timeout_ms は、いずれも最大で 315360000000、つまり 3650 日です。それより長い値は、フィールド、上限、送った値を名指しした 400 で拒否されます。それより先の期限は、デーモンがそれを保存する時計に収まらないからです。

リトライ、副作用、失敗

失敗した本体は、その分類でデーモンにどう失敗したかを伝えます。 retryable はもう一度試す価値があるという意味で、デーモンはノードの retry 予算が続く限り、ノードのバックオフに従って、本体が求めた retry_after_ms 以上は待ってから、もう一度試行します。この retry_after_ms にも、ほかのすべての期間と同じ 3650 日の上限がかかります。fatal と invalid_input はリトライされません。 unservable は、このワーカーがその定義かステップを持っていないという意味で、その試行は予算を使わずに別のワーカーのためにキューへ戻ります。それ以外を投げた本体は fatal です。

max_attempts は最初の試行も含めて数えるので、既定の 1 はリトライなしという意味です。予算を使い切ったあとは on_error が決めます。既定の fail_run は、そのステップのメッセージをエラーとして実行を終わらせます。continue はステップを失敗と印を付けて先へ進むので、あとのガードが steps.<id>.status を読んで迂回できます。

effect は、本体が二重に実行された場合の代償を宣言します。既定の at_least_once では、失われたリースは試行をキューに戻し、本体はもう一度実行されることがあります。at_most_once では、ワーカーは本体の前に started を送らなければならず、開始済みの試行のリースが失われた場合は、もう一度実行するのではなく失敗します。メールを送るステップやカードに請求するステップに使ってください。そのとき本体には、第三者に渡すための idempotency_key をリースから渡してください。これは同じステップのリトライをまたいで安定するので、応答が失われたあとのリトライは相手側でも認識されます。

バージョン

定義は一度保存されると変更できません。同じ本文をもう一度送れば同じバージョンが返ります。既存のバージョンに対して違う本文を送ると、デーモンは 409 で拒否します。そうしないと、そのバージョンを固定した実行が足元で変わってしまうからです。バージョンを省くとデーモンが次の番号を割り当てます。ワーカーが起動時にするのはこれです。実行は開始したときのバージョンを固定し、二度と解決し直さないので、バージョン 3 を保存するデプロイをしても、バージョン 2 で走っている実行は書かれたとおりに終わります。

上限

定義は max_steps、run_timeout_ms、 max_cost_usd を宣言できます。デーモンは 1 つ目に対してすべての試行を数え、2 つ目に対して実行の経過時間を数え、3 つ目に対してステップが報告した値付け済みの使用量を数えます。そして上限に達したところで、その上限をエラーとして名指しながら実行を timed_out か failed で終わらせます。収束しないループと、支出が止まらないエージェントのステップは、デーモンにとって同じ問題であり、同じ設定が両方を止めます。

デーモンが保持するもの

実行の入力、状態、出力、そして各ステップの入力と出力は、送られたとおりに実行ストアに保持されます。これらは秘匿化されません。それらはワークフロー自身の作業状態であり、あとのステップがそれを読み返すので、秘匿化処理が書き換えた値は実行の動きを変えてしまいます。実行がトレースのストアに書くスパンは、ほかのスパンと同じように秘匿化されます。保持期間と消去も、終わった実行については実行ストアに及びます。それぞれが何を削除するかは実行に正確に書いてあります。

次に

本体を処理する側と、ワーカーが話すプロトコル。ワーカー。
実行の開始、監視、シグナル送信、キャンセル。実行。
文書そのものをフィールドごとに見るなら definition.schema.json、そして拘束力のある契約は run-contract-v1.md。