Builder Guide

Building a Workflow

The registration contract, loop nodes, and the correction loop

Building a workflow

Audience: Customer developers. This page shows the shape of a workflow, the registration contract the platform discovers, and the correction-loop pattern at the heart of the SDK.

The registration contract

Every customer repo has a workflow/registration.py that exports four names. This is the contract the platform worker discovers — it's validated in CI, so keep the shapes exact:

# workflow/registration.py
from workflow_sdk import WorkflowRequirements
from .workflow import ReferenceReviewWorkflow, ingest_activity, agent_review_activity, ReviewInput

WORKFLOW_REGISTRY = {                       # name -> your @workflow.defn class
    "ReferenceReviewWorkflow": ReferenceReviewWorkflow,
}

WORKFLOW_REQUIREMENTS = {                    # governance: what each workflow may call
    "ReferenceReviewWorkflow": WorkflowRequirements(
        workflow_id="ReferenceReviewWorkflow",
        invokes=["agent-finance-compliance"],   # agents this workflow is allowed to call
        input_model=ReviewInput,                 # typed input (Pydantic)
    ),
}

ACTIVITIES = [ingest_activity, agent_review_activity]   # @activity.defn functions
INTERCEPTORS = []                                       # optional Temporal interceptors

WORKFLOW_REGISTRY — maps a stable name to your workflow class.
WORKFLOW_REQUIREMENTS — declares each workflow as a governance principal: which agents (invokes) and tools (tools) it may call, and its typed input_model. The input_model also becomes the Studio start form (see Running a workflow); the grants are covered in Calling agents & tools.
ACTIVITIES — the functions that do the actual work (I/O, calls). Activities are where side effects live; the workflow orchestrates them.
INTERCEPTORS — advanced; usually empty.

The shape of a workflow

A workflow is a normal Temporal workflow class — @workflow.defn with a @workflow.run method, both re-exported from workflow_sdk. The SDK's reviewable_workflow drives it as an async generator you iterate with async for:

from datetime import timedelta
from workflow_sdk import workflow, reviewable_workflow
from .models import ReviewRecord, WorkflowInput

@workflow.defn
class ReferenceReviewWorkflow:
    @workflow.run
    async def run(self, payload: dict) -> dict:
        wf_input = WorkflowInput.model_validate(payload)
        state = {"case_id": wf_input.case_id}

        async for loop in reviewable_workflow(
            state, loop_id="main", max_iterations=5, entity_type=ReviewRecord,
        ):
            await loop.start("ingest", ingest_activity, wf_input.payload,
                             schedule_to_close_timeout=timedelta(seconds=10))
            await loop.target(agent_review_activity, state_key="review",
                              schedule_to_close_timeout=timedelta(seconds=120))
            await loop.hitl(
                hitl_id=f"review_{loop.run_count}",
                raw_keys=["ingest"], reviewable_keys=["review"],
                display_fields=["value", "label"],
                hitl_metadata={"risk_score": 40, "impact": "Reference Review"},
            )

        return {"case_id": wf_input.case_id,
                "review": loop.entity.model_dump() if loop.entity else None}

The loop nodes

Inside the async for, you compose your process from a few node types:

Node	Runs when	Use for
`loop.start(name, activity, arg, …)`	Once, at the top of the first iteration	Ingest / setup that shouldn't repeat
`loop.step(activity, …)`	Every iteration	Intermediate processing
`loop.target(activity, state_key=…, …)`	The repeat entry point — re-runs on repeat	The step a human asks to redo, with their feedback
`loop.hitl(…)`	Every iteration	A governed human review (see Human-in-the-loop)

Each node runs a Temporal activity and stores its typed result into state under a key. Steps before target() are cached on a repeat; target() and everything after it re-run.

The correction loop

reviewable_workflow is the correction loop. If a reviewer asks to repeat a step, the loop runs another iteration — resuming at loop.target() and carrying the reviewer's feedback forward as a prompt your activity can read. max_iterations bounds it: exceeding it raises MaxIterationsError (non-retryable) instead of looping forever.


Scroll to zoom · Drag to pan · 100%

What Temporal gives you (for free)

Because the SDK runs on Temporal, you inherit durable-execution guarantees without writing them:

Reliability — activities retry on failure with backoff; timeouts are explicit (schedule_to_close_timeout); a crashed worker resumes exactly where it left off.
Durable state — your workflow's state and position survive restarts and can run for weeks (waiting on a human or an email costs nothing while parked).
Branching — ordinary Python if/match inside @workflow.run, plus child workflows for larger sub-processes.
Parallelism — start several activities and await them together (e.g. Python's asyncio.gather) to run steps concurrently.

The raw-Temporal escape hatch. workflow and activity are the genuine Temporal objects, so you can call workflow.execute_activity, workflow.execute_child_workflow, workflow.sleep, or define your own signals/queries directly in the same run method. That's fully supported — but the SDK does not wrap those, so the agentic-hub patterns (correction loop, HITL, email watcher, managed state) won't apply to code you write against raw Temporal. Prefer the SDK nodes; drop down only for what they don't cover.

Next: Human-in-the-loop.