Workflows API

Workflows orchestrate multi-step agent execution across six patterns. Each workflow contains a definition with steps, optional loop or swarm configuration, and persisted execution state. Workflows manage their own task lifecycle — creating, executing, and monitoring step tasks automatically.

Quick Start

Create a workflow with sequential steps, execute it, poll for status, and check the results:

// 1. Create a sequence workflow with three steps
interface Workflow {
id: string;
name: string;
status: string;
runNumber: number;
}

const wfRes: Response = await fetch("http://localhost:3000/api/workflows", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
  name: "Content Pipeline",
  projectId: "proj-8f3a-4b2c",
  definition: {
    pattern: "sequence",
    steps: [
      {
        id: "research",
        name: "Research",
        prompt: "Research competitor pricing pages and summarize patterns",
        assignedAgent: "claude-code",
        agentProfile: "researcher",
      },
      {
        id: "draft",
        name: "Draft",
        prompt: "Write a pricing page draft based on the research findings",
        dependsOn: ["research"],
      },
      {
        id: "review",
        name: "Review",
        prompt: "Review the draft for clarity, accuracy, and tone",
        requiresApproval: true,
        dependsOn: ["draft"],
      },
    ],
  },
}),
});
const workflow: Workflow = await wfRes.json();
// → { id: "wf-4a7c-e2d1", status: "draft", runNumber: 0, ... }

// 2. Execute the workflow — returns 202 immediately
await fetch(`http://localhost:3000/api/workflows/${workflow.id}/execute`, { method: "POST" });

// 3. Poll for status until all steps are done
interface StepState { stepId: string; status: string }
interface WorkflowStatus { status: string; steps: StepState[] }

const poll = async (): Promise<string> => {
const statusRes: Response = await fetch(`http://localhost:3000/api/workflows/${workflow.id}/status`);
const status: WorkflowStatus = await statusRes.json();

console.log(`Workflow: ${status.status}`);
for (const step of status.steps) {
  console.log(`  ${step.stepId}: ${step.status}`);
}
return status.status;
};

// 4. Check output documents produced by the workflow
const docsRes: Response = await fetch(`http://localhost:3000/api/workflows/${workflow.id}/documents`);
const docs: { documentId: string }[] = await docsRes.json();

console.log(`${docs.length} documents produced`);

import requests

# 1. Create a sequence workflow with three steps
wf_res = requests.post("http://localhost:3000/api/workflows", json={
  "name": "Content Pipeline",
  "projectId": "proj-8f3a-4b2c",
  "definition": {
      "pattern": "sequence",
      "steps": [
          {
              "id": "research",
              "name": "Research",
              "prompt": "Research competitor pricing pages and summarize patterns",
              "assignedAgent": "claude-code",
              "agentProfile": "researcher",
          },
          {
              "id": "draft",
              "name": "Draft",
              "prompt": "Write a pricing page draft based on the research findings",
              "dependsOn": ["research"],
          },
          {
              "id": "review",
              "name": "Review",
              "prompt": "Review the draft for clarity, accuracy, and tone",
              "requiresApproval": True,
              "dependsOn": ["draft"],
          },
      ],
  },
})
workflow: dict = wf_res.json()
# → { "id": "wf-4a7c-e2d1", "status": "draft", "runNumber": 0, ... }

# 2. Execute the workflow — returns 202 immediately
requests.post(f"http://localhost:3000/api/workflows/{workflow['id']}/execute")

# 3. Poll for status until all steps are done
status_res = requests.get(f"http://localhost:3000/api/workflows/{workflow['id']}/status")
status: dict = status_res.json()

print(f"Workflow: {status['status']}")
for step in status["steps"]:
  print(f"  {step['stepId']}: {step['status']}")

# 4. Check output documents produced by the workflow
docs_res = requests.get(f"http://localhost:3000/api/workflows/{workflow['id']}/documents")
docs: list[dict] = docs_res.json()

print(f"{len(docs)} documents produced")

Base URL

/api/workflows

Patterns

Pattern	Description
sequence	Steps execute in order, each waits for the previous to complete
parallel	Steps execute concurrently with dependency tracking
loop	Single step repeats for configurable iterations with time/signal budget
planner-executor	Planning phase followed by execution phase
checkpoint	Steps with approval gates between execution stages
swarm	Concurrent worker agents with configurable concurrency limit

Endpoints

List Workflows

GET /api/workflows

Retrieve all workflows with task count and output document count aggregation, ordered newest first.

Response Body (Array)

Field	Type	Req	Description
id	string (UUID)	*	Workflow identifier
name	string	*	Workflow name
projectId	string (UUID)	—	Associated project
definition	string (JSON)	*	Serialized WorkflowDefinition
status	enum	*	draft, active, paused, completed, or failed
runNumber	number	*	Execution iteration counter
taskCount	number	*	Total step tasks created
outputDocCount	number	*	Output documents produced
createdAt	ISO 8601	*	Creation timestamp
updatedAt	ISO 8601	*	Last modification

Fetch all workflows to build a pipeline overview — includes task and document counts:

// List all workflows with aggregated counts
const response: Response = await fetch("http://localhost:3000/api/workflows");
const workflows: Workflow[] = await response.json();

workflows.forEach((wf) => {
const def: { pattern: string } = JSON.parse(wf.definition);
console.log(`${wf.name} [${wf.status}] — ${def.pattern}, ${wf.taskCount} tasks, ${wf.outputDocCount} docs`);
});

import requests
import json

# List all workflows with aggregated counts
response = requests.get("http://localhost:3000/api/workflows")
workflows: list[dict] = response.json()

for wf in workflows:
  defn = json.loads(wf["definition"])
  print(f"{wf['name']} [{wf['status']}] — {defn['pattern']}, {wf['taskCount']} tasks, {wf['outputDocCount']} docs")

Example response:

[
  {
    "id": "wf-4a7c-e2d1",
    "name": "Content Pipeline",
    "projectId": "proj-8f3a-4b2c",
    "status": "completed",
    "runNumber": 2,
    "taskCount": 3,
    "outputDocCount": 1,
    "createdAt": "2026-04-01T08:00:00.000Z",
    "updatedAt": "2026-04-01T09:15:00.000Z"
  }
]

Create Workflow

POST /api/workflows

Create a new workflow in draft state. The definition must include a valid pattern and at least one step.

Request Body

Field	Type	Req	Description
name	string	*	Workflow name (non-empty)
projectId	string (UUID)	—	Project to associate with
definition	WorkflowDefinition	*	Pattern, steps, and optional config

WorkflowDefinition

Field	Type	Req	Description
pattern	enum	*	sequence, parallel, loop, planner-executor, checkpoint, or swarm
steps	WorkflowStep[]	*	At least one step
loopConfig	LoopConfig	—	Required for loop pattern
swarmConfig	SwarmConfig	—	Optional concurrency settings for swarm
sourceTaskId	string	—	Task that originated this workflow

WorkflowStep

Field	Type	Req	Description
id	string	*	Step identifier (unique within workflow)
name	string	*	Step display name
prompt	string	*	Agent prompt for this step
requiresApproval	boolean	—	Pause for human approval before executing
dependsOn	string[]	—	Step IDs that must complete first
assignedAgent	string	—	Override agent runtime for this step
agentProfile	string	—	Override profile for this step
documentIds	string[]	—	Documents available to this step

LoopConfig

Field	Type	Req	Description
maxIterations	number	*	Maximum loop iterations
timeBudgetMs	number	—	Total time budget in milliseconds
assignedAgent	string	—	Agent runtime for loop iterations
agentProfile	string	—	Profile for loop iterations
completionSignals	string[]	—	Agent output signals that end the loop early

Response 201 Created — Workflow with status: "draft", runNumber: 0

Errors: 400 — Invalid name, definition, or assignment validation failure

Create a parallel workflow where independent steps run concurrently, with a final step that depends on both:

// Create a parallel workflow — pricing and features run concurrently,
// synthesis waits for both to finish
const response: Response = await fetch("http://localhost:3000/api/workflows", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
  name: "Competitive Analysis",
  projectId: "proj-8f3a-4b2c",
  definition: {
    pattern: "parallel",
    steps: [
      { id: "pricing", name: "Pricing Research", prompt: "Analyze competitor pricing models" },
      { id: "features", name: "Feature Comparison", prompt: "Compare feature sets across top 5 competitors" },
      { id: "synthesis", name: "Synthesis", prompt: "Combine pricing and feature analysis into a final report", dependsOn: ["pricing", "features"] },
    ],
  },
}),
});
const workflow: Workflow = await response.json();

console.log(workflow.id);     // "wf-4a7c-e2d1"
console.log(workflow.status); // "draft"

import requests

# Create a parallel workflow — pricing and features run concurrently,
# synthesis waits for both to finish
response = requests.post("http://localhost:3000/api/workflows", json={
  "name": "Competitive Analysis",
  "projectId": "proj-8f3a-4b2c",
  "definition": {
      "pattern": "parallel",
      "steps": [
          {"id": "pricing", "name": "Pricing Research", "prompt": "Analyze competitor pricing models"},
          {"id": "features", "name": "Feature Comparison", "prompt": "Compare feature sets across top 5 competitors"},
          {"id": "synthesis", "name": "Synthesis", "prompt": "Combine pricing and feature analysis into a final report", "dependsOn": ["pricing", "features"]},
      ],
  },
})
workflow: dict = response.json()

print(workflow["id"])     # "wf-4a7c-e2d1"
print(workflow["status"]) # "draft"

Example response:

{
  "id": "wf-4a7c-e2d1",
  "name": "Competitive Analysis",
  "projectId": "proj-8f3a-4b2c",
  "status": "draft",
  "runNumber": 0,
  "taskCount": 0,
  "outputDocCount": 0,
  "createdAt": "2026-04-03T10:00:00.000Z",
  "updatedAt": "2026-04-03T10:00:00.000Z"
}

Update Workflow

PATCH /api/workflows/{id}

Edit a draft workflow's name or definition, or pause an active workflow. Active/paused workflows cannot be edited — only paused.

Mode A — Edit (draft, completed, or failed workflows only):

Request Body

Field	Type	Req	Description
name	string	—	Updated name
definition	WorkflowDefinition	—	Updated definition

Mode B — Pause (active workflows only):

{ "status": "paused" }

Editing a completed or failed workflow resets it to draft and clears execution state.

Errors: 404 — Not found, 409 — Cannot edit active/paused workflow, invalid status transition

Update the name and add a step to a draft workflow:

// Rename a draft workflow before execution
await fetch("http://localhost:3000/api/workflows/wf-4a7c-e2d1", {
method: "PATCH",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ name: "Competitive Analysis v2" }),
});

import requests

# Rename a draft workflow before execution
requests.patch("http://localhost:3000/api/workflows/wf-4a7c-e2d1", json={
  "name": "Competitive Analysis v2",
})

Delete Workflow

DELETE /api/workflows/{id}

Delete a workflow and all related resources. Active workflows cannot be deleted — pause first.

Response 200 — { "deleted": true }

Cascade deletes: usage ledger, agent logs, notifications, documents, learned context, tasks, then workflow.

Errors: 404 — Not found, 409 — Cannot delete active workflow

// Delete a workflow — must not be in active state
const res: Response = await fetch("http://localhost:3000/api/workflows/wf-4a7c-e2d1", { method: "DELETE" });

if (res.status === 409) {
console.log("Pause the workflow first before deleting");
}

import requests

# Delete a workflow — must not be in active state
res = requests.delete("http://localhost:3000/api/workflows/wf-4a7c-e2d1")

if res.status_code == 409:
  print("Pause the workflow first before deleting")

Execute Workflow

POST /api/workflows/{id}/execute

Start workflow execution. Returns 202 immediately while the workflow engine runs steps asynchronously. Increments runNumber on each execution.

Response 202 Accepted — { "status": "started", "workflowId": "..." }

For completed or failed workflows, re-execution resets the internal state and starts fresh.

Errors: 404 — Not found, 409 — Already running

Execute a workflow and track its progress — the engine creates tasks for each step automatically:

// Start the workflow — returns immediately while steps run in background
const res: Response = await fetch("http://localhost:3000/api/workflows/wf-4a7c-e2d1/execute", {
method: "POST",
});

if (res.status === 202) {
const { workflowId }: { workflowId: string } = await res.json();
console.log("Workflow started — poll /status to track progress");
} else if (res.status === 409) {
console.log("Workflow is already running");
}

import requests

# Start the workflow — returns immediately while steps run in background
res = requests.post("http://localhost:3000/api/workflows/wf-4a7c-e2d1/execute")

if res.status_code == 202:
  data: dict = res.json()
  print("Workflow started — poll /status to track progress")
elif res.status_code == 409:
  print("Workflow is already running")

Example response:

{
  "status": "started",
  "workflowId": "wf-4a7c-e2d1"
}

Get Workflow Status

GET /api/workflows/{id}/status

Get detailed execution status including per-step states, loop progress, documents, and run history.

The response shape depends on the workflow pattern:

Non-loop patterns return steps[] with per-step state:

Step State Fields

Field	Type	Req	Description
stepId	string	*	Step identifier
status	enum	*	pending, running, completed, failed, waiting_approval, or waiting_dependencies
taskId	string	—	Associated task ID (once created)
result	string	—	Step execution result
error	string	—	Error message if failed
startedAt	ISO 8601	—	Step start time
completedAt	ISO 8601	—	Step completion time

Loop patterns return loopState with iteration history:

Loop State Fields

Field	Type	Req	Description
currentIteration	number	*	Current iteration index
iterations	array	*	Per-iteration status, result, timing
status	enum	*	running, completed, paused, or failed
stopReason	enum	—	max_iterations, time_budget, agent_signaled, human_cancel, human_pause, or error
totalDurationMs	number	—	Total execution time

Both shapes include stepDocuments (per-task output files), parentDocuments (workflow-level inputs), and runHistory (per-run aggregation).

Errors: 404 — Not found

Poll workflow status to build a real-time progress tracker:

// Poll workflow status to display step-by-step progress
const response: Response = await fetch("http://localhost:3000/api/workflows/wf-4a7c-e2d1/status");
const status: WorkflowStatus = await response.json();

console.log(`Workflow: ${status.status}, Run #${status.runNumber}`);

// Display per-step progress
for (const step of status.steps) {
const duration: string = step.completedAt && step.startedAt
  ? ((new Date(step.completedAt).getTime() - new Date(step.startedAt).getTime()) / 1000).toFixed(1) + "s"
  : "—";
console.log(`  ${step.stepId}: ${step.status} (${duration})`);
}

import requests
from datetime import datetime

# Poll workflow status to display step-by-step progress
response = requests.get("http://localhost:3000/api/workflows/wf-4a7c-e2d1/status")
status: dict = response.json()

print(f"Workflow: {status['status']}, Run #{status['runNumber']}")

# Display per-step progress
for step in status["steps"]:
  if step.get("completedAt") and step.get("startedAt"):
      started = datetime.fromisoformat(step["startedAt"])
      completed = datetime.fromisoformat(step["completedAt"])
      duration = f"{(completed - started).total_seconds():.1f}s"
  else:
      duration = "—"
  print(f"  {step['stepId']}: {step['status']} ({duration})")

Example response for a running sequence workflow:

{
  "status": "active",
  "runNumber": 1,
  "steps": [
    {
      "stepId": "research",
      "status": "completed",
      "taskId": "task-b3e1-7f2a",
      "result": "Found 5 competitor pricing pages with common patterns...",
      "startedAt": "2026-04-03T10:01:00.000Z",
      "completedAt": "2026-04-03T10:03:22.000Z"
    },
    {
      "stepId": "draft",
      "status": "running",
      "taskId": "task-d9c4-1a8e",
      "startedAt": "2026-04-03T10:03:23.000Z"
    },
    {
      "stepId": "review",
      "status": "waiting_dependencies"
    }
  ],
  "stepDocuments": {},
  "parentDocuments": [],
  "runHistory": []
}

List Workflow Documents

GET /api/workflows/{id}/documents

List all document bindings for a workflow, including step-scoped and workflow-level documents.

Response Body (Array)

Field	Type	Req	Description
bindingId	string (UUID)	*	Binding record ID
documentId	string (UUID)	*	Document ID
stepId	string	—	Step scope (null = all steps)
document	object	—	Document metadata (name, MIME type, size, direction, status)

List all documents attached to a workflow, including step-scoped and workflow-wide bindings:

// List workflow documents to see inputs and outputs
const response: Response = await fetch("http://localhost:3000/api/workflows/wf-4a7c-e2d1/documents");
const docs: { bindingId: string; documentId: string; stepId?: string }[] = await response.json();

// Separate workflow-wide vs step-scoped documents
const globalDocs = docs.filter((d) => !d.stepId);
const scoped = docs.filter((d) => d.stepId);
console.log(`${globalDocs.length} global, ${scoped.length} step-scoped documents`);

import requests

# List workflow documents to see inputs and outputs
response = requests.get("http://localhost:3000/api/workflows/wf-4a7c-e2d1/documents")
docs: list[dict] = response.json()

# Separate workflow-wide vs step-scoped documents
global_docs = [d for d in docs if not d.get("stepId")]
scoped = [d for d in docs if d.get("stepId")]
print(f"{len(global_docs)} global, {len(scoped)} step-scoped documents")

Attach Documents

POST /api/workflows/{id}/documents

Attach documents to a workflow. Optionally scope to a specific step. Duplicate bindings are silently ignored.

Request Body

Field	Type	Req	Description
documentIds	string[]	*	Document UUIDs to attach (non-empty)
stepId	string	—	Scope to a specific step (null = all steps)

Response 201 Created — { "attached": 2, "workflowId": "...", "stepId": null }

Errors: 400 — Empty array, 404 — Workflow or documents not found

Attach reference documents to a specific step — only that step’s agent will see these files:

// Attach documents scoped to the research step only
const response: Response = await fetch("http://localhost:3000/api/workflows/wf-4a7c-e2d1/documents", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
  documentIds: ["doc-competitor-data-q4", "doc-pricing-spreadsheet"],
  stepId: "research",
}),
});
const { attached }: { attached: number } = await response.json();

console.log(`Attached ${attached} documents to research step`);

import requests

# Attach documents scoped to the research step only
response = requests.post("http://localhost:3000/api/workflows/wf-4a7c-e2d1/documents", json={
  "documentIds": ["doc-competitor-data-q4", "doc-pricing-spreadsheet"],
  "stepId": "research",
})
data: dict = response.json()

print(f"Attached {data['attached']} documents to research step")

Remove Documents

DELETE /api/workflows/{id}/documents

Remove document bindings. Pass specific IDs to remove selectively, or omit to remove all bindings.

Request Body (optional)

Field	Type	Req	Description
documentIds	string[]	—	Specific documents to unbind
stepId	string	—	Filter removal by step

Response 200 — { "ok": true }

// Remove a specific document binding
await fetch("http://localhost:3000/api/workflows/wf-4a7c-e2d1/documents", {
method: "DELETE",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ documentIds: ["doc-competitor-data-q4"] }),
});

import requests

# Remove a specific document binding
requests.delete("http://localhost:3000/api/workflows/wf-4a7c-e2d1/documents", json={
  "documentIds": ["doc-competitor-data-q4"],
})

Retry Step

POST /api/workflows/{id}/steps/{stepId}/retry

Retry a failed workflow step. Returns 202 while the retry runs asynchronously.

Response 202 Accepted — { "status": "retry_started", "workflowId": "...", "stepId": "..." }

Retry a failed step without re-running the entire workflow:

// Retry a failed step — the workflow engine picks up from here
const res: Response = await fetch("http://localhost:3000/api/workflows/wf-4a7c-e2d1/steps/research/retry", {
method: "POST",
});

if (res.status === 202) {
console.log("Retry started — downstream steps will re-execute when this completes");
}

import requests

# Retry a failed step — the workflow engine picks up from here
res = requests.post("http://localhost:3000/api/workflows/wf-4a7c-e2d1/steps/research/retry")

if res.status_code == 202:
  print("Retry started — downstream steps will re-execute when this completes")

Create from AI Assist

POST /api/workflows/from-assist

Create a workflow with pre-built step tasks in a single atomic operation. Optionally execute immediately.

Request Body

Field	Type	Req	Description
name	string	*	Workflow name
projectId	string (UUID)	—	Project association
definition	WorkflowDefinition	*	Pattern and steps
priority	number	—	Default priority for step tasks(default: `2`)
assignedAgent	string	—	Default agent for steps
executeImmediately	boolean	—	Start execution after creation

Response Body

Field	Type	Req	Description
workflow	object	*	Created workflow
taskIds	string[]	*	IDs of created step tasks
status	enum	*	created or started

Errors: 400 — Validation failure

Create a workflow and start it immediately in one call — useful for AI-assisted task decomposition:

// Create and execute in one call — no need for a separate execute request
const response: Response = await fetch("http://localhost:3000/api/workflows/from-assist", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
  name: "Quick Analysis",
  definition: {
    pattern: "sequence",
    steps: [
      { id: "analyze", name: "Analyze", prompt: "Analyze the Q4 revenue dataset and produce a summary" },
    ],
  },
  executeImmediately: true,
}),
});
const { workflow, taskIds, status }: { workflow: Workflow; taskIds: string[]; status: string } = await response.json();

console.log(`Workflow ${workflow.id}: ${status}`); // "started"
console.log(`Task IDs: ${taskIds.join(", ")}`);

import requests

# Create and execute in one call — no need for a separate execute request
response = requests.post("http://localhost:3000/api/workflows/from-assist", json={
  "name": "Quick Analysis",
  "definition": {
      "pattern": "sequence",
      "steps": [
          {"id": "analyze", "name": "Analyze", "prompt": "Analyze the Q4 revenue dataset and produce a summary"},
      ],
  },
  "executeImmediately": True,
})
data: dict = response.json()

print(f"Workflow {data['workflow']['id']}: {data['status']}")  # "started"
print(f"Task IDs: {', '.join(data['taskIds'])}")

Example response:

{
  "workflow": {
    "id": "wf-7b2e-c4a9",
    "name": "Quick Analysis",
    "status": "active",
    "runNumber": 1
  },
  "taskIds": ["task-e5f1-8d3c"],
  "status": "started"
}

Get Debug Info

GET /api/workflows/{id}/debug

Analyze a failed workflow and return a structured debug report with root cause classification, event timeline, per-step errors, and tiered fix suggestions.

Response 200 — DebugAnalysis object

DebugAnalysis Object

Field	Type	Req	Description
rootCause.type	enum	*	Failure category: budget_exceeded, timeout, transient, or unknown
rootCause.summary	string	*	Human-readable summary of the root cause
timeline	TimelineEvent[]	*	Ordered list of agent log events with severity tagging
suggestions	FixSuggestion[]	*	Tiered remediation suggestions (quick, better, best)
stepErrors	object[]	*	Per-step errors extracted from workflow state

TimelineEvent

Field	Type	Req	Description
timestamp	ISO 8601	*	When the event occurred
event	string	*	Event type from agent logs
severity	enum	*	success, warning, or error
details	string	*	Error message, result snippet, or step name
stepId	string	—	Step associated with this event (if any)

FixSuggestion

Field	Type	Req	Description
tier	enum	*	Effort level: quick, better, or best
title	string	*	Short suggestion title
description	string	*	Detailed description of the fix
action	string	—	Machine-readable action identifier (e.g. raise_budget, reduce_docs)

Errors: 500 — Workflow not found or analysis failed

Fetch debug info after a workflow fails — use the root cause type and suggestions to build a guided recovery UI:

// Fetch debug analysis for a failed workflow
interface DebugAnalysis {
rootCause: { type: string; summary: string };
timeline: { timestamp: string; event: string; severity: string; details: string; stepId?: string }[];
suggestions: { tier: string; title: string; description: string; action?: string }[];
stepErrors: { stepId: string; stepName: string; error: string }[];
}

const response: Response = await fetch("http://localhost:3000/api/workflows/wf-4a7c-e2d1/debug");
const debug: DebugAnalysis = await response.json();

console.log(`Root cause: ${debug.rootCause.type}`);
console.log(`Summary: ${debug.rootCause.summary}`);

// Show step-level errors
debug.stepErrors.forEach((e) => {
console.log(`  Step "${e.stepName}" failed: ${e.error}`);
});

// Show the quickest fix
const quickFix = debug.suggestions.find((s) => s.tier === "quick");
if (quickFix) {
console.log(`Quick fix: ${quickFix.title} — ${quickFix.description}`);
}

import requests

# Fetch debug analysis for a failed workflow
response = requests.get("http://localhost:3000/api/workflows/wf-4a7c-e2d1/debug")
debug: dict = response.json()

print(f"Root cause: {debug['rootCause']['type']}")
print(f"Summary: {debug['rootCause']['summary']}")

# Show step-level errors
for e in debug["stepErrors"]:
  print(f"  Step "{e['stepName']}" failed: {e['error']}")

# Show the quickest fix
quick_fix = next((s for s in debug["suggestions"] if s["tier"] == "quick"), None)
if quick_fix:
  print(f"Quick fix: {quick_fix['title']} — {quick_fix['description']}")

Example response for a budget-exceeded failure:

{
  "rootCause": {
    "type": "budget_exceeded",
    "summary": "The workflow exceeded its allocated budget before completing all steps."
  },
  "timeline": [
    {
      "timestamp": "2026-04-03T10:01:00.000Z",
      "event": "step_started",
      "severity": "warning",
      "details": "research",
      "stepId": "research"
    },
    {
      "timestamp": "2026-04-03T10:04:11.000Z",
      "event": "step_failed",
      "severity": "error",
      "details": "Budget exceeded: $5.00 limit reached",
      "stepId": "research"
    }
  ],
  "suggestions": [
    {
      "tier": "quick",
      "title": "Raise budget to $10",
      "description": "Increase the per-step or workflow budget cap to allow the agent more room to finish.",
      "action": "raise_budget"
    },
    {
      "tier": "better",
      "title": "Reduce document context per step",
      "description": "Attach fewer or smaller documents to each step so the agent consumes less budget on context.",
      "action": "reduce_docs"
    },
    {
      "tier": "best",
      "title": "Split into smaller workflows",
      "description": "Break the workflow into focused sub-workflows that each stay within budget.",
      "action": "restructure"
    }
  ],
  "stepErrors": [
    {
      "stepId": "research",
      "stepName": "Research",
      "error": "Budget exceeded: $5.00 limit reached"
    }
  ]
}

Resume Workflow

POST /api/workflows/{id}/resume

Manually resume a paused workflow, skipping any remaining delay. Returns 202 immediately and runs the resume asynchronously. The scheduler also resumes workflows automatically when their delay window expires.

Response 202 Accepted — { "status": "resuming", "workflowId": "..." }

Resume is safe to call concurrently with the scheduler tick — the internal status transition is atomic, so a racing scheduler resume will complete silently without duplicating execution.

Errors:

404 — Workflow not found
409 — Workflow is not in paused state (already resumed or was never paused)

Resume a paused workflow immediately without waiting for the delay window to expire:

// Resume a paused workflow — skips remaining delay
const res: Response = await fetch("http://localhost:3000/api/workflows/wf-4a7c-e2d1/resume", {
method: "POST",
});

if (res.status === 202) {
const { workflowId }: { workflowId: string } = await res.json();
console.log(`Workflow ${workflowId} is resuming`);
} else if (res.status === 409) {
const { status }: { error: string; status: string } = await res.json();
console.log(`Cannot resume — current status: ${status}`);
}

import requests

# Resume a paused workflow — skips remaining delay
res = requests.post("http://localhost:3000/api/workflows/wf-4a7c-e2d1/resume")

if res.status_code == 202:
  data: dict = res.json()
  print(f"Workflow {data['workflowId']} is resuming")
elif res.status_code == 409:
  data = res.json()
  print(f"Cannot resume — current status: {data['status']}")

Example response:

{
  "status": "resuming",
  "workflowId": "wf-4a7c-e2d1"
}

Optimize Workflow

POST /api/workflows/optimize

Analyze a workflow definition and return data-driven optimization suggestions covering document binding efficiency, budget estimates, runtime recommendations, and pattern insights. Each sub-analysis is independent — partial results are returned even if one analysis fails.

Request Body

Field	Type	Req	Description
definition	WorkflowDefinition	*	The workflow definition to analyze (same shape as the definition field in Create Workflow)
workflowId	string (UUID)	—	Existing workflow ID — enables budget estimates using historical execution data

Response 200 — { "suggestions": OptimizationSuggestion[] }

OptimizationSuggestion

Field	Type	Req	Description
type	enum	*	Suggestion category: document_binding, budget_estimate, runtime_recommendation, or pattern_insight
title	string	*	Short suggestion headline
description	string	*	Detailed explanation with concrete numbers where available
data	object	*	Analysis data backing the suggestion (shape varies by type)
action.label	string	—	Human-readable label for the suggested action
action.type	string	—	Machine-readable action identifier (e.g. set_runtime, change_pattern)
action.payload	object	—	Payload to apply the action (e.g. { runtimeId: "claude-code" })

Errors: 400 — definition missing

Analyze a workflow definition before executing it to catch efficiency issues early:

// Analyze a workflow definition for optimization opportunities
interface OptimizationSuggestion {
type: string;
title: string;
description: string;
data: Record<string, unknown>;
action?: { label: string; type: string; payload: Record<string, unknown> };
}

const response: Response = await fetch("http://localhost:3000/api/workflows/optimize", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
  workflowId: "wf-4a7c-e2d1",
  definition: {
    pattern: "sequence",
    steps: [
      { id: "research", name: "Research", prompt: "Research competitor pricing pages" },
      { id: "draft", name: "Draft", prompt: "Write pricing page draft", documentIds: ["doc-1", "doc-2", "doc-3", "doc-4"] },
      { id: "review", name: "Review", prompt: "Review the draft for clarity", documentIds: ["doc-1", "doc-2", "doc-3", "doc-4"] },
    ],
  },
}),
});
const { suggestions }: { suggestions: OptimizationSuggestion[] } = await response.json();

suggestions.forEach((s) => {
console.log(`[${s.type}] ${s.title}`);
console.log(`  ${s.description}`);
if (s.action) {
  console.log(`  → ${s.action.label} (action: ${s.action.type})`);
}
});

import requests

# Analyze a workflow definition for optimization opportunities
response = requests.post("http://localhost:3000/api/workflows/optimize", json={
  "workflowId": "wf-4a7c-e2d1",
  "definition": {
      "pattern": "sequence",
      "steps": [
          {"id": "research", "name": "Research", "prompt": "Research competitor pricing pages"},
          {"id": "draft", "name": "Draft", "prompt": "Write pricing page draft", "documentIds": ["doc-1", "doc-2", "doc-3", "doc-4"]},
          {"id": "review", "name": "Review", "prompt": "Review the draft for clarity", "documentIds": ["doc-1", "doc-2", "doc-3", "doc-4"]},
      ],
  },
})
data: dict = response.json()

for s in data["suggestions"]:
  print(f"[{s['type']}] {s['title']}")
  print(f"  {s['description']}")
  if s.get("action"):
      print(f"  → {s['action']['label']} (action: {s['action']['type']})")

Example response:

{
  "suggestions": [
    {
      "type": "document_binding",
      "title": "Reduce document injections with per-step binding",
      "description": "4 docs × 3 steps = 12 injections → only 5 needed with per-step binding",
      "data": {
        "globalDocCount": 4,
        "stepCount": 3,
        "totalInjections": 12,
        "perStepInjections": 5
      },
      "action": {
        "label": "Enable per-step docs",
        "type": "set_per_step_docs",
        "payload": { "strategy": "per-step" }
      }
    },
    {
      "type": "budget_estimate",
      "title": "Estimated cost: $0.3240",
      "description": "Within budget cap of $15.00",
      "data": {
        "totalEstimatedCostUsd": 0.324,
        "totalBudgetCapUsd": 15.0,
        "overBudget": false,
        "perStepBreakdown": [
          { "name": "Research", "estimatedCostUsd": 0.085, "budgetCapUsd": 5.0 },
          { "name": "Draft", "estimatedCostUsd": 0.142, "budgetCapUsd": 5.0 },
          { "name": "Review", "estimatedCostUsd": 0.097, "budgetCapUsd": 5.0 }
        ],
        "warnings": []
      }
    },
    {
      "type": "runtime_recommendation",
      "title": "Recommended runtime: claude-code",
      "description": "Based on 24 similar runs with 92% success rate",
      "data": {
        "runtimeId": "claude-code",
        "similarWorkflowStats": { "sampleCount": 24, "successRate": 0.92 }
      },
      "action": {
        "label": "Use this runtime",
        "type": "set_runtime",
        "payload": { "runtimeId": "claude-code" }
      }
    }
  ]
}

Workflow Statuses

Status	Description
draft	Created but not yet executed
active	Currently executing steps
paused	Execution suspended (can resume via execute)
completed	All steps finished successfully
failed	One or more steps failed

Execution State

Workflow execution state is persisted inside the definition field:

_state — Used by sequence, parallel, planner-executor, checkpoint, and swarm patterns. Tracks currentStepIndex, per-step states, and overall workflow status.
_loopState — Used by loop pattern. Tracks iteration count, per-iteration results, stop reason, and timing.

Both are automatically cleared when a completed or failed workflow is re-executed or edited back to draft.