Workflow

A workflow is the top-level process, defined by the scientific result it produces. It coordinates phases, manages the data flow between them, and owns the state of all samples and resources for the duration of the run.

You define a workflow as a Python function decorated with @workflow. Inside it, you call phases like regular functions. The workflow engine handles scheduling, parallelism, error recovery, and checkpoints.

Example

from unitelabs.sdk.automate import workflow
from .phases import sample_preparation, agitation, washing_cycle, detection

@workflow(name="ELISA")
async def main_workflow():
    plate = await sample_preparation()
    await agitation(plate)

    for _ in range(3):
        await washing_cycle()

    await detection()

The workflow drives the scientific narrative. Phases represent the meaningful stages — sample_preparation, agitation, detection — while the workflow defines the order and control flow between them.

Key properties

• Scientific goal: a workflow is defined by what result it produces (e.g., "ELISA"), not the hardware it uses
• Control flow: supports loops, conditionals, and branching; not limited to a linear sequence
• State ownership: the workflow is the one place that keeps track of sample identity, lineage, and all consumed resources across phases
• Versioned: workflows are deployed and versioned; each run is linked to a specific version

Non-linear control flow

Because a workflow is just a Python function, you can use standard control flow:

@workflow(name="Repeat Until Clean")
async def repeat_workflow():
    result = await initial_wash()

    while not result.is_clean:
        result = await additional_wash()

    await final_rinse()

Parallel phases

Phases with no dependency between them can run in parallel. The workflow engine detects independence automatically — you only need to express the data dependency:

@workflow(name="Parallel Preparation")
async def parallel_workflow():
    # These three phases have no shared inputs — the workflow engine runs them concurrently
    reagent_a = await prepare_reagent_a()
    reagent_b = await prepare_reagent_b()
    await wash_plate()

    # This phase depends on both reagents, so it waits for both to complete
    await combine(reagent_a, reagent_b)

Run context

Every run carries a context object: the single place workflow-level state lives for the duration of the run. Rather than passing a run mode, a simulation flag, or a feature flag as an argument through every intermediate phase and step, you declare it once as a workflow input and read it from context wherever it's needed — no matter how deeply nested the phase or step is.

from unitelabs.sdk import get_context, phase, workflow

@workflow(name="ELISA")
async def main_workflow(simulate: bool = False):
    await sample_preparation()
    await detection()

@phase()
async def detection():
    simulate = get_context().workflow_parameters.get("simulate", False)
    ...

detection never declares simulate in its own signature, and neither does any step it calls in turn — it reads the value straight from context. Adding a new workflow-level parameter later needs no signature changes anywhere downstream.

• get_context() returns the RuntimeContext active for the currently running workflow, phase, or step.
• context.workflow_parameters is a read-only mapping, populated automatically from the arguments the top-level @workflow function is called with. Nothing needs to be called to fill it in. This is only available with SDK >= 0.15.0 or liquid handling SDK >= 0.34.0.
• Only workflow-level arguments are published this way. A phase's or step's own arguments remain ordinary function arguments — they are not added to workflow_parameters. See Input for the full workflow-vs-phase breakdown.
• The context also carries the run across phase boundaries — it's what a resumed run reloads to pick up where it left off (see Runs).

Related concepts

• Phase: the logical stages a workflow is composed of
• Runs: what happens when a workflow executes
• Input: parameterize a workflow at run time