☁️ 技術調査 - Cloudflare Dynamic Workflows

# npx 経由（インストール不要）
npx wrangler --version

# グローバルインストールする場合
npm install -g wrangler

npm create cloudflare@latest -- my-workflow

{
  "$schema": "node_modules/wrangler/config-schema.json",
  "name": "my-workflow",
  "main": "src/index.ts",
  "compatibility_date": "2026-05-01",
  "observability": {
    "enabled": true
  },
  "workflows": [
    {
      "name": "my-workflow",
      "binding": "MY_WORKFLOW",
      "class_name": "MyWorkflow"
    }
  ]
}

npx wrangler types

npm install @cloudflare/dynamic-workflows

import { WorkflowEntrypoint, WorkflowStep } from "cloudflare:workers";
import type { WorkflowEvent } from "cloudflare:workers";

type Params = { name?: string };

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    const data = await step.do("fetch data", async () => {
      const response = await fetch("https://api.cloudflare.com/client/v4/ips");
      return await response.json();
    });

    await step.sleep("pause", "20 seconds");

    const result = await step.do(
      "process data",
      { retries: { limit: 3, delay: "5 seconds", backoff: "linear" } },
      async () => {
        return { name: event.payload.name ?? "World" };
      },
    );
    return result;
  }
}

from workers import WorkflowEntrypoint

class MyWorkflow(WorkflowEntrypoint):
    async def run(self, event, step):
        @step.do('fetch-data')
        async def fetch_data():
            return {"status": "ok"}

        result = await fetch_data()

        await step.sleep("pause", "20 seconds")

        @step.do('process-data', config={"retries": {"limit": 3, "delay": "5 seconds"}})
        async def process_data():
            return {"name": event.payload.get("name", "World")}

        return await process_data()

const result = await step.do(
  "step-with-retry",
  { retries: { limit: 10, delay: "10 seconds", backoff: "exponential" }, timeout: "30 minutes" },
  async () => {
    return { value: 42 };
  },
);

await step.sleep("wait 1 hour", "1 hour");
await step.sleepUntil("wait until deadline", new Date("2026-12-31T00:00:00Z"));

let stripeEvent = await step.waitForEvent<IncomingStripeWebhook>(
  "receive invoice paid webhook from Stripe",
  { type: "stripe-webhook", timeout: "1 hour" },
);

import { NonRetryableError } from "cloudflare:workflows";

await step.do("validate payload", async () => {
  if (!event.payload.data) {
    throw new NonRetryableError("event.payload.data did not contain the expected payload");
  }
});

const instance = await env.MY_WORKFLOW.create({
  id: crypto.randomUUID(),
  params: { name: "Alice" },
});

curl https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/workflows/$WORKFLOW_NAME/instances \
    -X POST \
    -H "Authorization: Bearer $CLOUDFLARE_API_TOKEN" \
    -H "Content-Type: application/json" \
    -d '{"params": {"name": "Alice"}}'

npx wrangler workflows trigger my-workflow '{"name": "Alice"}' --id my-custom-id

const instance = await env.MY_WORKFLOW.get(instanceId);
await instance.sendEvent({
  type: "stripe-webhook",
  payload: webhookPayload,
});

import {
  createDynamicWorkflowEntrypoint,
  wrapWorkflowBinding,
  DynamicWorkflowBinding,
} from '@cloudflare/dynamic-workflows';

function loadTenant(env: Env, tenantId: string) {
  return env.LOADER.get(tenantId, async () => ({
    compatibilityDate: '2026-01-01',
    mainModule: 'index.js',
    modules: { 'index.js': await fetchTenantCode(tenantId) },
    env: { WORKFLOWS: wrapWorkflowBinding({ tenantId }) },
  }));
}

export const DynamicWorkflow = createDynamicWorkflowEntrypoint<Env>(
  async ({ env, metadata }) => {
    const stub = loadTenant(env, metadata.tenantId);
    return stub.getEntrypoint('TenantWorkflow');
  }
);

// Dynamic Worker 経由で Workflow instance を作成するために必須
export { DynamicWorkflowBinding };

import { WorkflowEntrypoint } from "cloudflare:workers";

export class TenantWorkflow extends WorkflowEntrypoint {
  async run(event, step) {
    return step.do('greet', async () => `Hello, ${event.payload.name}!`);
  }
}

import { dispatchWorkflow } from '@cloudflare/dynamic-workflows';

export class MyDynamicWorkflow extends WorkflowEntrypoint {
  async run(event, step) {
    return dispatchWorkflow(
      { env: this.env, ctx: this.ctx },
      event,
      step,
      ({ metadata, env }) => loadRunnerForTenant(env, metadata),
    );
  }
}

# デプロイ
wrangler deploy

# 特定環境へデプロイ
wrangler deploy --env production

# 一覧
wrangler workflows list
wrangler workflows instances list MY_WORKFLOW

# 詳細（ステップ・リトライ・エラーログ）
wrangler workflows describe MY_WORKFLOW
wrangler workflows instances describe MY_WORKFLOW <instanceId>

# トリガー
wrangler workflows trigger MY_WORKFLOW '{"name":"test"}'

# ライフサイクル
wrangler workflows instances pause MY_WORKFLOW <instanceId>
wrangler workflows instances resume MY_WORKFLOW <instanceId>
wrangler workflows instances restart MY_WORKFLOW <instanceId>
wrangler workflows instances terminate MY_WORKFLOW <instanceId>

# イベント注入（waitForEvent 待機中インスタンスへ）
wrangler workflows instances send-event MY_WORKFLOW <instanceId> \
  --type stripe-webhook \
  --payload '{"amount": 1000}'

const instance = await env.MY_WORKFLOW.get(instanceId);

await instance.status();
await instance.pause();
await instance.resume();
await instance.restart();
await instance.terminate();

await instance.sendEvent({
  type: "stripe-webhook",
  payload: webhookPayload,
});

# リアルタイム
wrangler tail --format pretty
wrangler tail --format json

# インスタンス単位
wrangler workflows instances describe MY_WORKFLOW <instanceId>

{
  viewer {
    accounts(filter: { accountTag: "YOUR_ACCOUNT_ID" }) {
      workflowsAdaptiveGroups(
        filter: {
          workflowName: "my-workflow"
          datetimeGeq: "2026-05-01T00:00:00Z"
        }
        limit: 100
      ) {
        dimensions {
          workflowName
          instanceId
          stepName
          eventType
        }
        count
      }
    }
  }
}

await step.do("send notification", async () => {
  const already = await db.get(`notified:${userId}`);
  if (already) return { skipped: true };
  await sendEmail(userId);
  await db.set(`notified:${userId}`, true);
  return { sent: true };
});

class ApprovalWorkflow extends WorkflowEntrypoint {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    const analysis = await step.do("analyze request", async () => {
      return await callLLM(event.payload.request);
    });

    const approval = await step.waitForEvent("wait for human approval", {
      type: "approval-decision",
      timeout: "7 days",
    });

    if (approval.payload.approved) {
      await step.do("execute approved action", async () => {
        return await executeAction(analysis);
      });
    }
  }
}

const instance = await env.MY_WORKFLOW.get(instanceId);
await instance.sendEvent({
  type: "approval-decision",
  payload: { approved: true, reviewer: "alice@example.com" },
});

// 1. ツール呼び出しと実副作用を分離する基本形
const toolResult = await step.do("collect order data via tool", async () => {
  return await llmToolCall({ tool: "get_order_details", orderId });
});

await step.do("write order to DB", async () => {
  await db.insert("orders", toolResult);
});

// 2. 複数副作用の補償トランザクション例
async function placeOrder(event, step) {
  const reservation = await step.do("reserve inventory", async () => {
    return await inventory.reserve(event.payload.orderId);
  });

  try {
    const charge = await step.do("charge payment", async () => {
      return await payment.charge(event.payload.userId, event.payload.amount);
    });
    await step.do("send confirmation", async () => {
      return await notify.send(event.payload.userId, { charge, reservation });
    });
  } catch (e) {
    await step.do("compensate inventory", async () => {
      return await inventory.release(reservation.id);
    });
    // 補償完了後にワークフロー全体を失敗として確定させる
    throw new NonRetryableError(`order placement failed: ${e.message}`);
  }
}

-- wrangler d1 execute <DB> --command "..." で実行
CREATE TABLE IF NOT EXISTS idempotency (
  request_id TEXT PRIMARY KEY,
  result     TEXT NOT NULL,
  created_at INTEGER NOT NULL
);

await step.do("call non-idempotent API", async () => {
  const requestId = `${event.instanceId}:send-email`;
  const cached = await env.DB.prepare(
    "SELECT result FROM idempotency WHERE request_id = ?",
  ).bind(requestId).first();
  if (cached) return JSON.parse(cached.result);

  const result = await externalApi.send(event.payload);
  await env.DB.prepare(
    "INSERT INTO idempotency (request_id, result, created_at) VALUES (?, ?, ?)",
  ).bind(requestId, JSON.stringify(result), Date.now()).run();
  return result;
});

const childId = crypto.randomUUID();
await step.do("spawn child workflow", async () => {
  await env.CHILD_WORKFLOW.create({ id: childId, params: { task: subtask } });
  return { childId };
});

const childResult = await step.waitForEvent("await child completion", {
  type: "child-complete",
  timeout: "1 hour",
});

const plan = await step.do("generate research plan", async () => {
  return await llm.generatePlan(event.payload.topic);
});

for (const subtask of plan.subtasks) {
  await step.do(`execute subtask: ${subtask.id}`, async () => {
    return await llm.executeSubtask(subtask);
  });
}

// good: 関係のない呼び出しは分割
const a = await step.do("call api1", () => api1());
const b = await step.do("call api2", () => api2());

[[durable_objects.bindings]]
name = "TenantDB"
class_name = "TenantStorage"

[[migrations]]
tag = "v1"
new_sqlite_classes = ["TenantStorage"]

await env.WORKFLOW.create({
  id: `${tenantId}-${crypto.randomUUID()}`,
  params: { tenantId, ...userParams },
});

console.log(JSON.stringify({
  level: "info",
  tenantId: event.payload.tenantId,
  instanceId: event.instanceId,
  step: "process-data",
  durationMs: elapsed,
}));

[observability]
enabled = true

// bad: 再実行で異なる ID が生成される
await step.do("create record", async () => {
  return await db.insert({ id: crypto.randomUUID(), ... });
});

// good: ID をステップ前に確定する
const id = await step.do("generate id", async () => crypto.randomUUID());
await step.do("create record", async () => {
  return await db.insert({ id, ... });
});

// good: step.do() の戻り値として保存
const imageList = await step.do("fetch images", async () => {
  return await fetch("...").then(r => r.json());
});