Workflows

Multi-step agent pipelines written as Python functions — the graph is derived from your type annotations, not drawn.

Describe the pipeline you want in the Dreadnode TUI, or write it directly as Python:

Add a workflow named review to this capability. It should ask the existing
reviewer agent to review a target, then return the verdict.

The Dreadnode agent writes the workflow module and manifest entry for you. Review the generated Python before running it: event types define the graph, while function bodies define the work.

The bundled creating-capabilities skill includes small Python examples for a sequence, fan-out and join, fixed fork, exclusive branch, and typed agent output. Ask dn to adapt one into your capability and test it. The plain Python examples run through WorkflowHost without a platform connection or model credentials; dn workflow run --local creates a platform run and still needs platform access.

from pydantic import BaseModel

from dreadnode.workflows import Ctx, StartEvent, StopEvent, Workflow, WorkflowEvent


class ReviewInput(BaseModel):
    """What this run reviews."""

    target: str


class Reviewed(WorkflowEvent):
    """The reviewer produced a verdict."""

    verdict: str


workflow = Workflow(name="review", input=ReviewInput, title="Review {target}")


@workflow.step
async def review(ctx: Ctx, ev: StartEvent[ReviewInput]) -> Reviewed:
    """Ask the capability's reviewer agent to inspect the target."""
    result = await ctx.agent("reviewer", f"Review {ev.input.target}")
    return Reviewed(verdict=result.output)


@workflow.step
async def finish(ctx: Ctx, ev: Reviewed) -> StopEvent[str]:
    """Return the review as the workflow result."""
    return StopEvent(result=ev.verdict)

Declare it in capability.yaml:

agents:
  - agents/reviewer.md
workflows:
  - workflows/review.py

Each workflow file must declare exactly one Workflow object. Workflow modules are listed explicitly; the SDK does not auto-discover workflows/.

Before your first run

Workflows is available in public beta for all organizations, including self-hosted deployments. Open Workflows in the navigation to browse definitions and runs.

You need the Dreadnode CLI configured, a project UUID, and the agents named by your workflow. A remote run also needs the UUID of a runtime in that project. Use dn runtime list --json to list runtimes.

Compile without uploading while you iterate:

dn capability push . --skip-upload

dn capability validate checks the capability structure, but workflow compilation happens during dn capability push, including --skip-upload. The compiler checks the graph and reports every problem in one pass.

Definitions belong to your organization, while runs belong to one workspace and project. A run and every session it opens share that scope. Workflow routes require workflow read or write permission plus access to the workspace. Run inputs, results, sessions, and structured output are private to the user who started the run; the owner of an assigned runtime can also read the run so that runtime can execute it.

Constructs

Everything structural comes from a return annotation.

You want	Write	You get
The entry point	`ev: StartEvent[Input]`	The one step that starts a run
A sequence	`-> NextEvent`	One edge
Parallel work	`-> list[Task]`	One instance of the consuming step per item
Two branches at once	`-> tuple[A, B]`	A fork — both paths run
A choice	`-> A \| B`	A branch; the untaken arm is recorded as skipped
Wait for many	`c: Collect[A, B]`	A join across one or more event types
The result	`-> StopEvent[str]`	The run’s output

Fan-out and joins

A step returning a list produces one instance of the consuming step per item. Each gets its own row on the run page and, when it calls ctx.agent(), its own agent session.

from pydantic import BaseModel

from dreadnode.workflows import Collect, Ctx, StartEvent, StopEvent, Workflow, WorkflowEvent


class ScanInput(BaseModel):
    targets: list[str]


class ScanTask(WorkflowEvent):
    target: str


class Finding(WorkflowEvent):
    target: str
    summary: str


workflow = Workflow(name="scan", input=ScanInput)


@workflow.step
async def plan(ctx: Ctx, ev: StartEvent[ScanInput]) -> list[ScanTask]:
    """Create one independent unit of work per target."""
    return [ScanTask(target=target) for target in ev.input.targets]


@workflow.step(max_concurrency=3, timeout_sec=300)
async def scan(ctx: Ctx, ev: ScanTask) -> Finding:
    """Run independently, with at most three targets active."""
    return Finding(target=ev.target, summary=f"Reviewed {ev.target}")


@workflow.step(min_success="all")
async def summarize(ctx: Ctx, findings: Collect[Finding]) -> StopEvent[str]:
    """Run after every scan has settled."""
    return StopEvent(result="\n".join(f.summary for f in findings.ok))

A join becomes eligible when its inputs have settled, not only when they succeed. Collect.ok contains successful events and Collect.failed contains node failures. Use min_success=N or min_success="all" when the workflow must fail instead of producing a partial result. Compilation rejects a numeric threshold above the maximum input count implied by upstream fan-out caps. A threshold that exceeds the actual successful results fails the run before the join body runs.

Returning an empty list is valid. A single-type join defaults to min_success=1, so it fails before the join body runs on an empty fan-out. Set min_success="all" when zero inputs should produce an empty, successful collection.

A fan-out can feed several processing steps before a join. Each step keeps one instance per input item, and Collect waits for every instance of its producers, including instances skipped because an upstream step failed or took another branch.

Use Collect.get(EventType) when a join waits for one event of each type, and Collect.all_of(EventType) when it receives several events of the same type.

Step options

@workflow.step(
    max_concurrency=3,        # cap simultaneous instances within one run
    min_success="all",        # join only: require every input to succeed
    timeout_sec=1800,         # per-step ceiling; exceeding it is a node failure
    materialization_cap=25,   # ceiling on fan-out width
)

materialization_cap defaults to 25. Increase it deliberately when a fan-out can create more instances.

The context object

Call	Purpose
`ctx.input`	The validated run input
`ctx.workspace`	A directory shared by steps in this runtime process
`ctx.agent(name, prompt, ...)`	Run one agent turn in its own session
`ctx.result(step)`	The output event of an earlier step
`ctx.log(message, **fields)`	A structured entry on the run’s timeline

ctx.agent() returns an AgentResult. Read .output, .tool_calls, or .session_id, or call .parse(Model) to validate structured text. Each call opens a session linked to the step on the run page.

ctx.workspace is local to the V1 runtime host. Files remain available across steps in one run, but they do not survive a runtime restart and are not portable to a future durable host. Pass durable state through workflow events or stored artifacts.

Naming a run

title is a template over the input model, rendered by the platform when the run is created:

workflow = Workflow(name="review", input=ReviewInput, title="Review {target}")

A run then reads as Review packages/api rather than a UUID. Placeholders are checked against the input model at compile time, so a typo is a build error rather than a run named {targte}.

Running one

Push the capability, then copy the definition and runtime UUIDs from the JSON output:

dn capability push .
dn workflow definitions --name review --json
dn runtime list --json

dn workflow run <definition-uuid> \
  --project-id <project-uuid> \
  --runtime-id <runtime-uuid> \
  --input target=packages/api

--project-id and --runtime-id take UUIDs, not project or runtime keys. The runtime must belong to the project. The platform binds the pushed capability version, starts the runtime when needed, and assigns the run to it. The runtime executes the Python while the platform records progress.

To run the workflow host in your current process, install the same capability source locally and start an agent runtime:

# Terminal 1
dn capability install .
dn serve

# Terminal 2
dn workflow run <definition-uuid> \
  --project-id <project-uuid> \
  --input target=packages/api \
  --local

--local still creates the run on the platform, but the CLI process imports the installed workflow source and executes its steps. The CLI verifies the installed capability version and workflow source digest against the pushed definition before it starts. The run exits if you stop the CLI process. Use --runtime-url when the agent runtime is not listening on its default address.

Monitoring a run

Open Workflows in the Dreadnode app to inspect the compiled graph, node status, structured output, and agent sessions created by each step. The definition page aggregates runs; a run page shows one execution.

Choose a project in the page toolbar. The catalog shows only workflow definitions with runs you created in that project, and switching projects reloads the list. A published workflow appears after you start its first run in the selected project.

Node status and agent session links update while steps execute. You can open an agent’s transcript before its turn finishes; steps that call several agents retain links to every session. Your zoom, pan, and selected step stay in place during updates and when switching between Graph and Data. Use Fit View to bring newly expanded steps into view, or the layout buttons to change direction.

The CLI exposes the same run state:

dn workflow status <run-uuid>
dn workflow list --project-id <project-uuid>

Structured output

Anything your agents record with report_item — findings, assets, or any item type your capability declares — is attributed to the run and to the step that produced it, and appears on both the run page and the workflow’s own Data tab. Enable agent output in the manifest:

outputs: true

The agent calls report_item; the platform attributes the item to the run and step. Only the run creator or its runtime owner can attach sessions to the run. Workflow attribution labels are fixed when the session is created, and items retain their attribution if the session is deleted.

Compilation errors

dn capability push compiles before it uploads and reports every problem it finds, not just the first:

workflow 'analysis' failed to compile (2 problems):
  - step 'validate': step 'validate' is unreachable from the entry point [WF-VALID-005]
    hint: nothing emits the event it consumes
  - step 'review': calls unknown agent 'reviewr' [WF-VALID-008]
    hint: declared agents: final-reviewer, finding-validator

Common causes include an event no step consumes, two steps consuming StartEvent, a cycle, an ambiguous ctx.result() call after fan-out, a missing return annotation, or an agent name that does not match a file in agents/. Avoid from __future__ import annotations in workflow modules because compilation needs the runtime annotation objects. Event class names must be unique within a workflow, including types imported from other modules (WF-VALID-019).

The platform also validates topology documents when capabilities are pushed. Documents must use a supported topology version, reference existing nodes, and fit within the 256 KiB size limit.

Tracing

Every span an agent emits during a workflow carries two extra attributes:

Attribute	Value
`dreadnode.workflow.run.id`	The run
`dreadnode.workflow.unit`	The step and its fan-out instance, e.g. `specialist:3:0`

So “every span this run produced” is one filter, and per-instance latency across a fan-out is a group-by. Workflow steps do not emit spans of their own — step timing lives in the run’s fact log, and duplicating it into trace storage would give two answers to the same question.

What runs where

A workflow executes on the capability runtime, beside your workers. Progress is recorded as an append-only fact log on the platform, and the run page is a projection folded from it — so it can be behind while facts are in flight, but it is never wrong.

If a platform append fails temporarily, the runtime retries the same ordered batch and waits for an acknowledgement before advancing. It stops after 60 seconds instead of retrying a rejected batch forever. This retry queue is in memory; it does not make the run durable.

V1 limits

V1 runs in one runtime process. If that runtime or a local CLI host stops, the run does not resume. The platform marks the run lost after its heartbeat expires and the run is read again. The in-memory fact queue also disappears with the process.

The platform rejects a V1 run whose graph contains an ApprovalRequest. The user who started a run can cancel it while it is still pending, but once a runtime claims it, V1 cannot interrupt the active process. Cycles, durable timers, and durable agent-step retries are also unavailable.