Workflows API

Run a node graph (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.

View as Markdown

A workflow is a small graph of typed blocks that turns a source (an image, a video, or a dataset) into an answer: run a model over every frame, keep the classes you care about, optionally track and count them, and emit results to a sink. You build the graph visually in the app, then drive runs from here.

A workflow’s model block loads your trained model directly, so no deployment is required. Runs are billed ONCE, on success, from the measured GPU time they use - there is no charge up front, and a failed or cancelled run is free (even if it spent real GPU time first). Management calls (create, update, delete, run, cancel) require member+ role.

The graph

A graph is {"version": 1, "nodes": [...], "edges": [...]}. Each node is {"id", "type", "config"}; each edge is {"source", "sourceHandle", "target", "targetHandle"}.

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)

An image or video source addresses its file by gcs_uri - the value client.video.upload() hands back, so you pass it through rather than construct it. A dataset source uses dataset_project_id instead.

Structural rules are validated before a run starts: one source, no loops, every block connected, and a counting or 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

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 - the default
`union`	everything any model found	maximum recall
`intersection`	only what every model found	maximum precision

Two models found “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.

{
  "nodes": [
    {"id": "s",  "type": "source", "config": {"kind": "video", "gcs_uri": "<from client.video.upload()>", "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). An ensemble costs about as many times as it has models: every model runs over every frame.

create

Create a workflow from a graph. Returns a Workflow in draft status.

Source: Workflows.create

Arg	Type	Default	Notes
`name`	`str`	required	Workflow name.
`graph`	`dict[str, Any]`	required	Node graph defining the workflow - source, model, filter, sink.
`readme`	`str \| None`	`None`	Markdown workflow card shown on the workflow page.
`description`	`str \| None`	`None`	One-line summary shown in the workflows list.
`template_key`	`str \| None`	`None`	Start from a built-in template instead of an explicit `graph`.

clip = client.video.upload(
    local_path="clip.mp4",
)

graph = {
    "version": 1,
    "nodes": [
        {"id": "s", "type": "source", "config": {"kind": "video", "gcs_uri": clip.gcs_uri, "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(
    name="Vehicle detector",
    graph=graph,
)
print(wf.id, wf.status)   # 'draft'

pictograph workflows create "Vehicle detector" --graph graph.json

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": '"$(cat graph.json)"'}'

Returns Workflow

Workflow · 10 fields

class Workflow(BaseModel):
    """A saved node-graph workflow."""
    id: str
    organization_id: str
    name: str
    description: str | None = None
    graph: dict[str, Any] = {}
    template_key: str | None = None
    status: Literal['draft', 'ready', 'archived'] = 'draft'
    last_run_id: str | None = None
    created_at: datetime | None = None
    updated_at: datetime | None = None

list

Every workflow in your organization.

Source: Workflows.list

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

pictograph workflows list

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

Returns Sequence[Workflow]

Workflow · 10 fields

class Workflow(BaseModel):
    """A saved node-graph workflow."""
    id: str
    organization_id: str
    name: str
    description: str | None = None
    graph: dict[str, Any] = {}
    template_key: str | None = None
    status: Literal['draft', 'ready', 'archived'] = 'draft'
    last_run_id: str | None = None
    created_at: datetime | None = None
    updated_at: datetime | None = None

get

Fetch one workflow. The REST response also carries a freshly computed validation array listing anything that would block a run.

Source: Workflows.get

Arg	Type	Default	Notes
`workflow`	`str`	required	Workflow name or id.

wf = client.workflows.get(
    workflow="Vehicle detector",
)
print(wf.status, wf.last_run_id, wf.graph)

pictograph workflows get "Vehicle detector"

curl -s "https://api.pictograph.io/api/v1/developer/workflows/Vehicle%20detector" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns Workflow

Workflow · 10 fields

class Workflow(BaseModel):
    """A saved node-graph workflow."""
    id: str
    organization_id: str
    name: str
    description: str | None = None
    graph: dict[str, Any] = {}
    template_key: str | None = None
    status: Literal['draft', 'ready', 'archived'] = 'draft'
    last_run_id: str | None = None
    created_at: datetime | None = None
    updated_at: datetime | None = None

update

Change a workflow’s name, description, graph, or status (draft, ready, archived). There is no CLI command for this - edit the graph in the app, or use the SDK.

Source: Workflows.update

Arg	Type	Default	Notes
`workflow`	`str`	required	Workflow name or id.
`name`	`str \| None`	`None`	New workflow name. Left unchanged when `None`.
`readme`	`str \| None`	`None`	New workflow card. Left unchanged when `None`.
`description`	`str \| None`	`None`	New summary. Left unchanged when `None`.
`graph`	`dict[str, Any] \| None`	`None`	Replaces the node graph. Left unchanged when `None`.
`status`	`WorkflowStatus \| None`	`None`	New state: `draft`, `ready` or `archived`. Left unchanged when `None`.

wf = client.workflows.update(
    workflow="Vehicle detector",
    status="ready",
)

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

# No update command; re-create or edit in the app.
pictograph workflows get nightly-count

Returns Workflow

Workflow · 10 fields

class Workflow(BaseModel):
    """A saved node-graph workflow."""
    id: str
    organization_id: str
    name: str
    description: str | None = None
    graph: dict[str, Any] = {}
    template_key: str | None = None
    status: Literal['draft', 'ready', 'archived'] = 'draft'
    last_run_id: str | None = None
    created_at: datetime | None = None
    updated_at: datetime | None = None

run

Validate the graph and start a run. Raises ValidationError (400) with the specific issues if the graph is not runnable, and PaymentRequiredError (402) if there is not enough compute credit.

Source: Workflows.run

Arg	Type	Default	Notes
`workflow`	`str`	required	Workflow name or id.

created = client.workflows.run(
    workflow="Vehicle detector",
)
print(created.run_id, created.deposit_micro_usd)

pictograph workflows run "Vehicle detector" --no-wait

curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/Vehicle%20detector/run" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns WorkflowRunCreated

WorkflowRunCreated · 2 fields

class WorkflowRunCreated(BaseModel):
    """Run response - the new run id + `deposit_micro_usd`, which is the un-charged pre-run ESTIMATE. Workflows bill ONCE, on success, from measured GPU time; a failed or cancelled run is free. The field name is kept for wire-compat."""
    run_id: str
    deposit_micro_usd: int = 0

wait_for_run

Block until a run reaches a terminal state - use this instead of hand-rolling a poll loop. Returns the run on completed, raises ApiError if it ends error or cancelled, and PollTimeoutError if timeout elapses (the run keeps going server-side).

Source: Workflows.wait_for_run

Arg	Type	Default	Notes
`run_id`	`str`	required	The run UUID from `run`.
`poll_interval`	`float`	`5.0`	Seconds between checks (default 5s).
`timeout`	`float`	`3600.0`	Max seconds to wait (default 3600 = 1h, the runner’s cap).

run = client.workflows.wait_for_run(
    run_id=created.run_id,
    poll_interval=5.0,
    timeout=3600.0,
)
for art in run.artifacts:
    print(art["kind"], art.get("download_url"))

# `run` waits by default; drop --no-wait to block until the run finishes.
pictograph workflows run "Vehicle detector" --timeout 3600

# Poll the run until status is terminal.
curl -s "https://api.pictograph.io/api/v1/developer/workflows/runs/<run-id>" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns WorkflowRun

WorkflowRun · 16 fields

class WorkflowRun(BaseModel):
    """One execution of a workflow over a source."""
    id: str
    organization_id: str
    workflow_id: str
    status: Literal['queued', 'processing', 'completed', 'error', 'cancelled']
    progress: float = 0.0
    frames_total: int | None = None
    frames_done: int = 0
    sample_fps: float | None = None
    step_results: dict[str, Any] = {}
    artifacts: list[dict[str, Any]] = []
    warnings: list[str] = []
    deposit_micro_usd: int = 0
    final_micro_usd: int | None = None
    error: str | None = None
    created_at: datetime | None = None
    completed_at: datetime | None = None

get_run

Read a run’s current state and, once complete, its artifacts (signed download URLs).

Source: Workflows.get_run

Arg	Type	Default	Notes
`run_id`	`str`	required	Run id.

run = client.workflows.get_run(
    run_id=created.run_id,
)
print(run.status, f"{run.frames_done}/{run.frames_total}", f"{run.progress:.0f}%")

pictograph workflows run-status 7f3a1b2c-4d5e-4f60-8a91-2b3c4d5e6f70

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

Returns WorkflowRun

WorkflowRun · 16 fields

class WorkflowRun(BaseModel):
    """One execution of a workflow over a source."""
    id: str
    organization_id: str
    workflow_id: str
    status: Literal['queued', 'processing', 'completed', 'error', 'cancelled']
    progress: float = 0.0
    frames_total: int | None = None
    frames_done: int = 0
    sample_fps: float | None = None
    step_results: dict[str, Any] = {}
    artifacts: list[dict[str, Any]] = []
    warnings: list[str] = []
    deposit_micro_usd: int = 0
    final_micro_usd: int | None = None
    error: str | None = None
    created_at: datetime | None = None
    completed_at: datetime | None = None

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 and refund the deposit.

Source: Workflows.cancel_run

Arg	Type	Default	Notes
`run_id`	`str`	required	Run id.

client.workflows.cancel_run(
    run_id=created.run_id,
)

pictograph workflows cancel 7f3a1b2c-4d5e-4f60-8a91-2b3c4d5e6f70 --yes

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

Returns None

bulk_cancel_runs

Cancel many runs in one call, each stopped in flight with its deposit refunded idempotently. Foreign, terminal, or missing ids land in not_found rather than failing the call.

Source: Workflows.bulk_cancel_runs

Arg	Type	Default	Notes
`run_ids`	`Sequence[str]`	required	Run ids to act on. Ids that do not resolve are reported, not raised.

result = client.workflows.bulk_cancel_runs(
    run_ids=[
        "7f3a1b2c-4d5e-4f60-8a91-2b3c4d5e6f70",
        "8a4b2c3d-5e6f-4071-9b02-3c4d5e6f7081",
    ],
)
print(result.count, "cancelled;", result.not_found, "skipped")

pictograph workflows cancel-batch \
  7f3a1b2c-4d5e-4f60-8a91-2b3c4d5e6f70 8a4b2c3d-5e6f-4071-9b02-3c4d5e6f7081 --yes

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": ["7f3a1b2c-4d5e-4f60-8a91-2b3c4d5e6f70", "8a4b2c3d-5e6f-4071-9b02-3c4d5e6f7081"]}'

Returns BulkActionResult

BulkActionResult · 3 fields

class BulkActionResult(BaseModel):
    """Result of a server-side bulk state-change (e.g. pause/resume)."""
    succeeded: list[str] = []
    not_found: list[str] = []
    count: int = 0

delete

Delete a workflow and its run history.

Source: Workflows.delete

Arg	Type	Default	Notes
`workflow`	`str`	required	Workflow name or id.

client.workflows.delete(
    workflow="Vehicle detector",
)

pictograph workflows delete "Vehicle detector" --yes

curl -s -X DELETE "https://api.pictograph.io/api/v1/developer/workflows/Vehicle%20detector" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns None

bulk_delete

Delete many workflows in one call. Names that do not resolve land in not_found rather than raising, so a re-run still succeeds.

Source: Workflows.bulk_delete

Arg	Type	Default	Notes
`workflows`	`Sequence[str]`	required	Names to delete (UUIDs are accepted too). Duplicates are ignored; ids that don’t resolve in your organization are reported in `BulkDeleteResult.not_found` rather than raising, so a re-run still succeeds.

res = client.workflows.bulk_delete(
    workflows=["Vehicle detector", "Dwell monitor"],
)
print(res.succeeded, res.not_found, res.count)

pictograph workflows delete "Vehicle detector" "Dwell monitor" --yes

curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/bulk-delete" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"workflow_ids": ["<uuid-a>", "<uuid-b>"]}'

Returns BulkDeleteResult

BulkDeleteResult · 3 fields

class BulkDeleteResult(BaseModel):
    """Result of a server-side bulk delete (one chunked, org-scoped call)."""
    succeeded: list[str] = []
    not_found: list[str] = []
    count: int = 0