---
title: Workflows API
description: Run a node-graph pipeline (source to model to filter to track to step to sink) over an image, a video, or a dataset. Create, run, poll, and read artifacts headlessly.
section: API Reference
order: 18
---
A workflow is a small graph of typed blocks that turns a source (an image, a
video, or a Pictograph dataset) into an answer: run a model over every frame,
keep the classes you care about, optionally track and count them, and emit the
results to a sink (JSON, CSV, a webhook, or an annotated video). You build the
graph visually in the app, then drive runs from here.

A workflow's model block runs your trained model directly, so no deployment is
required. Runs are billed per frame processed (denser frame sampling costs more),
charged as a deposit at start and trued up when the run finishes.

Every example below shows the Python SDK call and the equivalent raw REST
request. The REST examples authenticate with an `X-API-Key` header; set
`PICTOGRAPH_API_KEY` in your shell to copy-and-run them.

```python
from pictograph import Client
client = Client()  # reads PICTOGRAPH_API_KEY
```

> The `client.workflows` resource is the graph-workflows feature. It is separate
> from the `pictograph.pipelines` module (`full_pipeline` and friends), which
> chains upload, annotate, and train steps for convenience. (That orchestrator
> module was named `pictograph.workflows` before SDK 1.7.0; the old import path
> still works as a deprecated alias.)

## The graph

A graph is `{"version": 1, "nodes": [...], "edges": [...]}`. Each node is
`{"id", "type", "config"}`; each edge is `{"source", "sourceHandle", "target",
"targetHandle"}`. Block types and what they consume / produce:

| Block | In | Out | Config highlights |
|---|---|---|---|
| `source` | — | frames | `kind` (image/video/dataset), `gcs_uri` or `dataset_project_id`, `sample_fps` |
| `model` | frames | detections | `model_id` (a ready model), `confidence_threshold`, `class_filter` |
| `merge` | detections | detections | `strategy` (consensus/union/intersection), `min_votes`, `iou_threshold` |
| `filter` | detections | detections | `keep_classes`, `min_confidence` |
| `track` | detections | detections | `tracker` (bytetrack/botsort) |
| `step` | detections | events | `step_type` (line_cross/dwell/occupancy), `geometry` |
| `visualize` | det + events | frames | `draw` toggles |
| `sink` | det/events/frames | — | `kind` (json/csv/webhook/annotated_video) |

The structural rules are validated server-side before a run starts: one source,
no loops, every block connected, and a counting/dwell `step` must sit downstream
of a `track`, which must sit downstream of a `model`. A single-image source
cannot use `track` or `step` (no temporal axis).

## Ensembles — running several models together

Two detectors rarely fail the same way. A workflow can run up to **three** models over the
same frames and combine them with a `merge` block:

| `strategy` | Keeps | Use it for |
|---|---|---|
| `consensus` | detections at least `min_votes` models found (default 2) | fewer false positives, balanced — the default |
| `union` | everything any model found | maximum recall |
| `intersection` | only what **every** model found | maximum precision |

Two models are "the same detection" when they share a class name and their boxes overlap by
at least `iou_threshold` (default `0.5`). A model votes at most once per object, so a model
that fires twice on one thing cannot manufacture its own consensus. The surviving box is the
highest-confidence one, unchanged — coordinates are never averaged into a box no model
predicted. Its `confidence` becomes the mean across the models that voted, and each detection
carries `votes` and `model_count`.

```json
{
  "nodes": [
    {"id": "s",  "type": "source", "config": {"kind": "video", "gcs_uri": "gs://...", "sample_fps": 5}},
    {"id": "m1", "type": "model",  "config": {"model_id": "<uuid-a>"}},
    {"id": "m2", "type": "model",  "config": {"model_id": "<uuid-b>"}},
    {"id": "mg", "type": "merge",  "config": {"strategy": "consensus", "min_votes": 2, "iou_threshold": 0.5}},
    {"id": "out","type": "sink",   "config": {"kind": "json"}}
  ],
  "edges": [
    {"source": "s",  "target": "m1"}, {"source": "s",  "target": "m2"},
    {"source": "m1", "target": "mg"}, {"source": "m2", "target": "mg"},
    {"source": "mg", "target": "out"}
  ]
}
```

More than one `model` block is only valid when **all** of them feed the `merge` block — a
model wired around it would be run and billed and then ignored, so the run is rejected
(`models_need_merge` / `model_not_merged`) rather than quietly producing the wrong answer.

**An ensemble costs about as many times as it has models**: every model runs over every
frame, and you are billed for the GPU time that takes.

## create

Create a workflow from a graph. Returns a `Workflow` in `draft` status.
Management calls (create, update, delete, run, cancel) require **member+** role.

```python
graph = {
    "version": 1,
    "nodes": [
        {"id": "s", "type": "source", "config": {"kind": "video", "gcs_uri": "gs://.../clip.mp4", "sample_fps": 5}},
        {"id": "m", "type": "model", "config": {"model_id": "your-model-uuid", "confidence_threshold": 0.4}},
        {"id": "f", "type": "filter", "config": {"keep_classes": ["car", "truck"]}},
        {"id": "k", "type": "sink", "config": {"kind": "json"}},
    ],
    "edges": [
        {"source": "s", "target": "m"},
        {"source": "m", "target": "f"},
        {"source": "f", "target": "k"},
    ],
}

wf = client.workflows.create("Vehicle detector", graph)
print(wf.id, wf.status)   # 'draft'
```

```bash
curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Vehicle detector",
    "graph": {
      "version": 1,
      "nodes": [
        {"id": "s", "type": "source", "config": {"kind": "video", "gcs_uri": "gs://.../clip.mp4", "sample_fps": 5}},
        {"id": "m", "type": "model", "config": {"model_id": "your-model-uuid", "confidence_threshold": 0.4}},
        {"id": "f", "type": "filter", "config": {"keep_classes": ["car", "truck"]}},
        {"id": "k", "type": "sink", "config": {"kind": "json"}}
      ],
      "edges": [
        {"source": "s", "target": "m"},
        {"source": "m", "target": "f"},
        {"source": "f", "target": "k"}
      ]
    }
  }'
```

Returns a `Workflow`.

## list

List every workflow in your organization.

```python
for w in client.workflows.list():
    print(w.id, w.name, w.status)
```

```bash
curl -s "https://api.pictograph.io/api/v1/developer/workflows/" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"
```

Returns `Workflow` objects.

## get

Fetch a single workflow by UUID. The response includes a `validation` array
(empty when the graph is runnable).

```python
wf = client.workflows.get(wf.id)
print(wf.status, wf.validation)
```

```bash
curl -s "https://api.pictograph.io/api/v1/developer/workflows/{workflow_id}" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"
```

Returns a `Workflow`.

## update

Update a workflow's name, description, graph, or status. Valid statuses are
`draft`, `ready`, and `archived`. The response includes the refreshed
`validation` array.

```python
wf = client.workflows.update(wf.id, status="ready")
```

```bash
curl -s -X PATCH "https://api.pictograph.io/api/v1/developer/workflows/{workflow_id}" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"status": "ready"}'
```

Returns a `Workflow`.

## delete

Delete a workflow and its run history.

```python
client.workflows.delete(wf.id)
```

```bash
curl -s -X DELETE "https://api.pictograph.io/api/v1/developer/workflows/{workflow_id}" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"
```

## run

Validate the graph and start a run. The SDK raises `ValidationError` / 400 with
the specific issues if the graph is not runnable, and `PaymentRequiredError` /
402 if there is not enough compute credit. It returns a `WorkflowRunCreated`
with the `run_id`.

```python
created = client.workflows.run(wf.id)
print(created.run_id, created.deposit_micro_usd)
```

```bash
curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/{workflow_id}/run" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"
```

## get_run

Poll the run until it finishes, then read its artifacts (signed download URLs):

```python
import time

while True:
    run = client.workflows.get_run(created.run_id)
    print(run.status, f"{run.frames_done}/{run.frames_total}", f"{run.progress:.0f}%")
    if run.status in ("completed", "error", "cancelled"):
        break
    time.sleep(3)

if run.status == "completed":
    for art in run.artifacts:
        print(art["kind"], art.get("download_url"))
    # run.step_results carries the per-step aggregates + time series
```

```bash
# Poll this endpoint until status is completed / error / cancelled.
curl -s "https://api.pictograph.io/api/v1/developer/workflows/runs/{run_id}" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"
```

Returns a `WorkflowRun`.

| Field | Meaning |
|---|---|
| `status` | `queued`, `processing`, `completed`, `error`, or `cancelled`. |
| `progress` | 0–100. |
| `frames_done` / `frames_total` | Frames processed / estimated. |
| `step_results` | Per step block: aggregate (headline number), time series, per-track events. |
| `artifacts` | `[{kind, gcs_path, bytes, download_url}]` for JSON / CSV / annotated-video outputs. |
| `error` | Set when `status == "error"`. |

## cancel_run

Stop an in-flight run. This stops the GPU job and refunds the deposit.

```python
client.workflows.cancel_run(created.run_id)
```

```bash
curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/runs/{run_id}/cancel" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"
```

## bulk_cancel_runs

Cancel many runs in one call. Each run gets a Modal cancel plus an idempotent
deposit refund. Foreign, terminal, or missing ids land in `not_found` rather
than failing the call.

```python
result = client.workflows.bulk_cancel_runs(["run-1", "run-2"])
print(result.count, "cancelled;", result.not_found, "skipped")
```

```bash
curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/runs/bulk-cancel" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"run_ids": ["run-1", "run-2"]}'
```

Returns `BulkActionResult` with `succeeded`, `not_found`, and `count`.