Workflows API

Base path: /api/workflows

All endpoints require authentication. See API Overview for auth details.

Workflows

List workflows

GET /api/workflows

Query parameters: status, trigger_type, tag, limit (max 200), offset

Response:

{
  "workflows": [
    {
      "id": "wf_abc123",
      "name": "Data pipeline",
      "version": 3,
      "status": "active",
      "trigger_type": "manual",
      "tags": ["data", "etl"],
      "created_at": "2026-01-01T00:00:00Z",
      "updated_at": "2026-03-01T00:00:00Z"
    }
  ],
  "total": 5
}

Create a workflow

POST /api/workflows

{
  "definition": {
    "name": "Fetch and notify",
    "steps": [
      {
        "id": "fetch",
        "type": "http_request",
        "config": {
          "method": "GET",
          "url": "https://api.example.com/data"
        }
      },
      {
        "id": "notify",
        "type": "agent_task",
        "config": {
          "agent_id": "agent_123",
          "goal": "Summarize the fetched data and post to Slack"
        },
        "depends_on": ["fetch"]
      }
    ]
  },
  "trigger_type": "manual",
  "tags": ["notifications"]
}

Returns 201 with {"id": "wf_abc123", "workflow": {...}}.

Get a workflow

GET /api/workflows/{id}

Update a workflow

PUT /api/workflows/{id}

{
  "definition": {...},
  "change_summary": "Added error handling step"
}

Delete a workflow

DELETE /api/workflows/{id}

Update workflow status

PATCH /api/workflows/{id}/status

{"status": "paused"}

Valid statuses: draft, active, paused, archived

Validation

POST /api/workflows/validate

Validate a workflow definition without creating it:

{"definition": {...}}

Response:

{
  "valid": true,
  "errors": []
}

Errors example:

{
  "valid": false,
  "errors": [
    "Step 'notify' depends on 'fetch' which doesn't exist",
    "Cycle detected: step_a → step_b → step_a"
  ]
}

Version control

List versions

GET /api/workflows/{id}/versions

Response:

{
  "versions": [
    {"version": 3, "change_summary": "Added error step", "created_at": "..."},
    {"version": 2, "change_summary": "Fixed URL", "created_at": "..."},
    {"version": 1, "change_summary": "Initial version", "created_at": "..."}
  ]
}

Get a specific version

GET /api/workflows/{id}/versions/{version}

Returns the full workflow definition at that version.

Diff two versions

GET /api/workflows/{id}/diff?v1=1&v2=3

Response:

{
  "diff": [
    {"op": "add", "path": "/steps/2", "value": {...}},
    {"op": "replace", "path": "/steps/0/config/url", "value": "https://new-url.com"}
  ],
  "summary": "Added 1 step, modified 1 step",
  "v1": 1,
  "v2": 3
}

Rollback to a version

POST /api/workflows/{id}/rollback

{"version": 2}

Returns the workflow after rollback. Creates a new version entry.

Execution

Run a workflow

POST /api/workflows/{id}/run

{
  "input_data": {"dataset_id": "ds_123"},
  "trigger": "manual"
}

Returns 202 immediately:

{"run_id": "run_abc123"}

The workflow executes asynchronously. Use the run ID to track progress.

Concurrent run limit: configurable via WORKFLOW_MAX_CONCURRENT_RUNS (default: 10).

Runs

List runs for a workflow

GET /api/workflows/{id}/runs

Query: status, limit, offset

Get a run

GET /api/workflows/runs/{run_id}

Response includes step-by-step results:

{
  "run": {
    "id": "run_abc123",
    "workflow_id": "wf_abc123",
    "status": "completed",
    "trigger": "manual",
    "started_at": "2026-03-02T14:00:00Z",
    "completed_at": "2026-03-02T14:01:23Z",
    "duration_seconds": 83,
    "step_runs": [
      {
        "step_id": "fetch",
        "status": "completed",
        "output": {"status_code": 200, "body": {...}},
        "started_at": "...",
        "duration_seconds": 1
      },
      {
        "step_id": "notify",
        "status": "completed",
        "output": "Posted summary to #general",
        "started_at": "...",
        "duration_seconds": 82
      }
    ]
  }
}

Cancel a run

POST /api/workflows/runs/{run_id}/cancel

Resume a paused run

POST /api/workflows/runs/{run_id}/resume

{"from_step": "notify"}

Omit from_step to resume from the last checkpoint.

Delete a run

DELETE /api/workflows/runs/{run_id}

Stream run events (SSE)

GET /api/workflows/runs/{run_id}/stream

Server-Sent Events stream:

event: step_started
data: {"run_id":"run_abc123","step_id":"fetch","timestamp":"..."}

event: step_completed
data: {"run_id":"run_abc123","step_id":"fetch","output":{...}}

event: run_completed
data: {"run_id":"run_abc123","status":"completed","duration_seconds":83}

Templates

List templates

GET /api/workflows/templates

Query: category, is_builtin, limit, offset

Get a template

GET /api/workflows/templates/{id}

Install a template (create workflow from template)

POST /api/workflows/templates/{id}/install

{"name": "My custom pipeline"}

Returns 201 with {"workflow_id": "wf_abc123"}.

Schedules

List schedules for a workflow

GET /api/workflows/{id}/schedules

Create a schedule

POST /api/workflows/{id}/schedules

{
  "cron_expression": "0 9 * * 1-5",
  "timezone": "America/New_York",
  "input_data": {"dataset": "daily"}
}

Returns 201 with {"schedule_id": "sched_abc123"}.

Delete a schedule

DELETE /api/workflows/schedules/{id}

Note

Additional undocumented endpoints exist for workflow import, export, dry-run, approval, and schedule management. Refer to the source code for the complete list.

Step types

Type	Description
http_request	Make an HTTP API call
script	Run a shell or Python script
agent_task	Delegate to an AI agent
condition	Branch based on an expression
loop	Iterate over a list
wait	Pause for duration or condition
webhook	Emit an outbound HTTP request