Workflows API

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"}.

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 measure block must sit downstream of a track, which must sit downstream of a model. A single-image source cannot use track or measure (no temporal axis).

Multiple models

A workflow can run up to three models over the same frames - just wire each model block from the source. Their detections are combined automatically with a union (every model’s detections are kept, deduplicated by IoU overlap of the same class); there is no separate block to configure. The surviving box for an overlap is the highest-confidence one, unchanged - coordinates are never averaged into a box no model predicted.

{
  "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": "out","type": "sink",   "config": {"kind": "json"}}
  ],
  "edges": [
    {"source": "s",  "target": "m1"}, {"source": "s",  "target": "m2"},
    {"source": "m1", "target": "out"}, {"source": "m2", "target": "out"}
  ]
}

Running N models costs about N times as much: every model runs over every frame. The cap is three because they must all fit on the GPU at once.

create

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

Source: Workflows.create

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

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]

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

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

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, readme, description, graph, or status (draft, ready, archived) - edit the graph headlessly (SDK, CLI, or REST) or in the app. Export the current graph with pictograph workflows get <id>, edit the JSON, then update. Pass graph to replace the whole node graph (e.g. to tweak a Measure node’s geometries - labels, per-geometry badge_scale / badge_inline, count indicators - across a run).

Source: Workflows.update

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

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

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

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
    status_url: str | None = None

upload_source

Upload a local video or image as a workflow source and get back its gcs_uri. The file streams straight to object storage via a signed PUT URL (no API hop, so a multi-GB video never passes through the API pod), under your organization’s prefix. invoke calls this for you when you hand it a local path; call it directly to upload once and invoke several times. The kind (video/image) and content type are inferred from the file extension.

Source: Workflows.upload_source

gcs_uri = client.workflows.upload_source(
    local_path="clip.mp4",
)
# Reuse it across several runs without re-uploading:
run = client.workflows.invoke(
    workflow="Vehicle detector",
    source={"kind": "video", "gcs_uri": gcs_uri},
    wait=True,
)

# The CLI uploads + invokes in one step:
pictograph workflows invoke "Vehicle detector" --file clip.mp4

# 1. Get a signed PUT URL + the gcs_uri the file will live at:
curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/upload-url" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" -H "Content-Type: application/json" \
  -d '{"filename": "clip.mp4", "content_type": "video/mp4"}'
# 2. PUT the bytes to the returned upload_url (same Content-Type), then pass the gcs_uri to invoke.

Returns str

invoke

Run a saved workflow on demand over a source given at call time, without editing the workflow’s stored source node - the workflow-DAG equivalent of a model deployment’s /predict endpoint (“run this pipeline on my media”). Billed on success like run (a failed or cancelled run is free). Async by default (returns the run id + a poll URL); pass wait=True to block client-side for the terminal run.

source is one of:

• a local file path (str / Path) - a video or image on disk; it is uploaded (see upload_source) and the workflow runs over it. This is the quickest way to run a pipeline on your own media.
• a WorkflowInvokeSource / mapping with a kind:
  • dataset - a dataset (dataset = its name or id) in your organization, optionally narrowed to one directory_path (absent = the whole dataset).
  • video / image - a gcs_uri you already own (e.g. from upload_source), under your organization’s prefix. sample_fps and duration_seconds apply to video only.

A source that doesn’t fit the graph is rejected like any invalid graph (an image under a tracking pipeline raises ValidationError), and a gcs_uri outside your organization is a 400. A mistyped mapping field raises locally (WorkflowInvokeSource is strict).

Source: Workflows.invoke

# Run a saved workflow over a LOCAL video (uploaded for you), blocking for the result.
run = client.workflows.invoke(
    workflow="Vehicle detector",
    source="clip.mp4",
    wait=True,
)

# Or over a different dataset than the workflow's configured source (fire-and-forget).
created = client.workflows.invoke(
    workflow="Vehicle detector",
    source={"kind": "dataset", "dataset": "Rush-hour clips", "directory_path": "/north-cam"},
)
print(created.run_id, created.status_url)

# Upload + run a local file:
pictograph workflows invoke "Vehicle detector" --file clip.mp4

# Over a dataset/folder in your org (fire-and-forget):
pictograph workflows invoke "Vehicle detector" --dataset "Rush-hour clips" --directory /north-cam --no-wait

curl -s -X POST "https://api.pictograph.io/api/v1/developer/workflows/Vehicle%20detector/invoke" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY" -H "Content-Type: application/json" \
  -d '{"source": {"kind": "dataset", "dataset": "Rush-hour clips", "directory_path": "/north-cam"}}'

# Add ?wait=true (bounded, max 300s) to block for the terminal run record.

Returns WorkflowRunCreated | WorkflowRun

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
    status_url: str | None = None

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] = {}
    condition_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

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

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

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] = {}
    condition_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

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

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] = {}
    condition_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

list_runs

List a workflow’s runs, newest first, without opening the app. Each row is a slim summary - status, progress, frame counts, the settled or estimated cost, and the artifact count; call get_run for a single run’s full record plus signed artifact download URLs. The page is offset-paginated (a request past the last run returns an empty page, never an error); iter_runs pages automatically via client.workflows.iter_runs.

Source: Workflows.list_runs

runs = client.workflows.list_runs(
    workflow="Line counter",
    limit=20,
)
for run in runs:
    print(run.id, run.status, f"{run.cost_micro_usd} uUSD", f"{run.artifact_count} artifacts")

pictograph workflows runs "Line counter" --limit 20

curl -s "https://api.pictograph.io/api/v1/developer/workflows/$WORKFLOW_ID/runs?limit=20" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns Sequence[WorkflowRunSummary]

class WorkflowRunSummary(BaseModel):
    """A slim row from `client.workflows.list_runs` - the run-LIST projection."""
    id: str
    status: Literal['queued', 'processing', 'completed', 'error', 'cancelled']
    created_at: datetime | None = None
    completed_at: datetime | None = None
    runtime_seconds: float | None = None
    cost_micro_usd: int = 0
    cost_settled: bool = False
    frames_done: int = 0
    frames_total: int | None = None
    progress: float = 0.0
    artifact_count: int = 0
    error: str | None = None

iter_runs

Auto-paging iterator over every run of a workflow, newest first. Stops on the server-computed pagination.has_more flag; each item is a WorkflowRunSummary.

Source: Workflows.iter_runs

for run in client.workflows.iter_runs(workflow="Line counter"):
    print(run.id, run.status)

# Or materialize every run:
all_runs = client.workflows.iter_runs(workflow="Line counter").all()
print(len(all_runs), "runs")

# The CLI does not auto-page; this lists a single page of up to 100.
pictograph workflows runs "Line counter" --limit 100

# Page manually with limit + offset until pagination.has_more is false.
curl -s "https://api.pictograph.io/api/v1/developer/workflows/$WORKFLOW_ID/runs?limit=100&offset=100" \
  -H "X-API-Key: $PICTOGRAPH_API_KEY"

Returns OffsetPager[WorkflowRunSummary]

class WorkflowRunSummary(BaseModel):
    """A slim row from `client.workflows.list_runs` - the run-LIST projection."""
    id: str
    status: Literal['queued', 'processing', 'completed', 'error', 'cancelled']
    created_at: datetime | None = None
    completed_at: datetime | None = None
    runtime_seconds: float | None = None
    cost_micro_usd: int = 0
    cost_settled: bool = False
    frames_done: int = 0
    frames_total: int | None = None
    progress: float = 0.0
    artifact_count: int = 0
    error: str | None = None

cancel_run

Stop an in-flight run. Cancelled runs are free - workflows bill once, on success, so there is nothing to refund (a run that spent GPU time before you cancelled it is still not charged).

Source: Workflows.cancel_run

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 (cancelled runs are free - nothing to refund). Foreign, terminal, or missing ids land in not_found rather than failing the call.

Source: Workflows.bulk_cancel_runs

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

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

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

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

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