워크플로 · mcpd

역할의 분담

지속성 실행에는 주인이 둘 있습니다. 데몬은 단계와 단계 사이의 실행을 소유합니다. 커서, 저널, 재시도 예산, 타이머, 대기가 그것입니다. 직접 돌리는 프로세스는 단계 안의 모든 것을 소유합니다. 호출하는 모델, 보내는 프롬프트, 돌리는 도구, 들고 있는 자격 증명이 그것입니다. 데몬은 그중 아무것도 넘겨받지 않으며, 대신 모델을 돌리는 일도 없습니다.

그래서 워크플로는 파일 하나에서 두 부분으로 나뉘어 도착합니다. 그래프는 정의로, 데몬이 따라 걷는 문서로 데몬에게 갑니다. 단계의 본문은 작성한 언어 그대로 직접 돌리는 프로세스에 남아 워커가 처리하기를 기다립니다. 데몬에 배포할 것은 아무것도 없고, 정의를 전혀 바꾸지 않은 채로 본문만 바꿀 수 있습니다.

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가 같은 아홉 구문을 가진 같은 빌더를 갖추고 있어서, 한 언어로 쓴 워크플로가 다른 언어에서도 같게 읽히고, 어느 쪽이 썼든 데몬은 같은 문서를 저장합니다.

지원 업무 워크플로

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

저 예제에서 세 가지가 모든 워크플로의 모양입니다. 단계의 입력은 데몬이 시도마다 렌더링하는 템플릿이므로, draft는 두 본문이 서로를 알지 못한 채로 classify가 내놓은 답을 읽습니다. 가드는 노드가 아예 돌지 말지를 결정합니다. 질문은 그대로 보내고, 버그는 사람을 기다립니다. 그리고 마지막 단계는 재시도 정책이 붙은 at_most_once인데, 부수 효과가 있는 일에 필요한 조합이 바로 이것입니다. 데몬은 일시적인 실패를 재시도하고, 그러라는 말 없이 본문을 두 번 돌리는 일은 절대 없습니다.

구문

아홉이고, 모든 SDK가 아홉 개를 다 갖추고 있습니다. 둘은 가져갈 수 있는 것, 즉 워커가 처리하는 것이고, 나머지는 데몬 자신의 몫입니다.

구문	돌리는 쪽	하는 일
`task`	워커	가져갈 수 있는 일입니다. 워커는 이 id로 등록해 둔 코드를 그대로 돌립니다.
`agent`	워커	통신상으로는 같고, 에이전트의 한 차례로 기록됩니다. 직접 만든 루프가 모델을 몇 번 호출하든 지속성 있는 단계 하나가 됩니다. 트레이스와 인터페이스는 이것을 보통의 task와 구별합니다.
`sleep`	데몬	정해진 밀리초만큼의 지연입니다. 그동안 살아 있는 프로세스는 없습니다.
`sleep_until`	데몬	템플릿에서 렌더링된 시각까지의 지연입니다. 입력에 실려 온 청구일 같은 것입니다.
`wait_event`	데몬	이 이름과 키를 가진 이벤트가 도착하거나 마감 시각이 지날 때까지의 멈춤입니다. 키는 실행마다 렌더링되므로, 대기 하나는 주문 하나만을 가리키고 다른 것은 가리키지 않습니다.
`parallel`	양쪽	이름이 붙은 브랜치들이며, 각각이 가져갈 수 있는 단계 하나입니다. 어느 하나를 내주기 전에 전부가 디스크에 있습니다. 마지막 브랜치가 끝나면 실행이 앞으로 나아갑니다.
`foreach`	양쪽	어떤 경로에 있는 배열의 원소마다 가져갈 수 있는 단계 하나씩입니다. `max_items`로, 그리고 원한다면 동시에 몇 개가 도는지로 한계를 둡니다. 출력은 원소 순서대로 배열로 돌아옵니다.
`loop`	양쪽	가져갈 수 있는 단계 하나에 대한 한계가 있는 루프입니다. `until`은 반복이 끝날 때마다 평가되어 루프를 일찍 끝낼 수 있습니다.
`subworkflow`	데몬	다른 정의의 자식 실행이며 자기 저널을 갖습니다. 그 출력이 이 단계의 출력이 되고, 부모를 취소하면 함께 취소됩니다.

브랜치 하나, foreach의 본문 하나, loop의 본문 하나는 각각 가져갈 수 있는 단계 하나입니다. 여러 단계가 필요한 본문은 subworkflow이며, 그 덕분에 모든 저널이 평평하게 유지됩니다. 실행은 목록이지, 목록의 트리가 아닙니다.

템플릿과 가드

템플릿은 문자열 안에 $${path} 참조가 들어 있는 JSON입니다. 데몬은 시도가 만들어질 때, 그 순간 실행이 들고 있는 것으로 템플릿을 렌더링합니다. 실행의 입력인 input, 앞선 단계가 내놓은 답인 steps.<id>와 그 아래의 .output, .status, .error, 실행 자신의 id와 상관 정보인 run, 그리고 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를 선언할 수 있습니다. 데몬은 첫 번째에 대해 모든 시도를, 두 번째에 대해 실행의 나이를, 세 번째에 대해 단계들이 보고한 값 매겨진 사용량을 세다가, 상한에서 실행을 timed_out 또는 failed로 끝내고 그 상한을 오류로 이름 붙입니다. 수렴하지 않는 루프와 지출을 멈추지 않는 에이전트 단계는 데몬에게 같은 문제이고, 같은 플래그가 둘 다 멈춥니다.

데몬이 보관하는 것

실행의 입력, 상태, 출력, 그리고 모든 단계의 입력과 출력은 보낸 그대로 실행 저장소에 보관됩니다. 이들은 마스킹되지 않습니다. 이들은 워크플로 자신의 작업 상태이고, 뒤의 단계가 이것을 다시 읽으며, 마스킹기가 고쳐 쓴 값은 실행이 하는 일을 바꿀 것입니다. 실행이 트레이스 저장소에 쓰는 스팬은 다른 모든 스팬처럼 마스킹됩니다. 보존 기간과 삭제 요청은 실행 저장소에도 닿으며, 끝난 실행이 대상입니다. 각각이 정확히 무엇을 지우는지는 실행에 있습니다.

본문을 처리하는 일과, 워커가 말하는 프로토콜: 워커.
실행을 시작하고, 지켜보고, 신호를 보내고, 취소하기: 실행.
문서 자체를 필드 하나하나까지: definition.schema.json, 그리고 구속력 있는 계약인 run-contract-v1.md.