Workflow

A workflow is the top-level executable process you write to achieve a scientific outcome. It coordinates timing, logic, data flow, and physical state across one or more instruments, and it owns the state of all samples and resources for the duration of the run.

Put another way, a workflow is a lab protocol written in Python. The protocol is the procedure your lab already follows. The workflow is that procedure as code, which the platform runs, records, and lets your team start.

The automation hierarchy

UniteLabs workflows are built from four nested concepts:

Once started, a workflow creates a Run: a single execution of that workflow. Keep workflows, phases, and steps under version control in Git. To reuse phases and steps across workflows, collect them in a shared library inside your workflow repository. The workflow template does this in its shared/ package.

A minimal example

from unitelabs.sdk import __version__ as sdk_version
from unitelabs.sdk import get_logger, workflow


@workflow(name="Hello World")
async def hello_world(recipient_name: str = "world") -> None:
    """
    Log a greeting and SDK version info to confirm the environment works.

    Args:
      recipient_name: Name to greet (default: "world").
    """

    logger = get_logger()
    logger.info(f"Hello, {recipient_name}!")
    logger.info(f"UniteLabs SDK: {sdk_version}")

The @workflow decorator registers the function with the workflow engine. Phases are called like regular Python functions; the workflow engine handles scheduling, parallelism, and recovery.

Composing phases

from unitelabs.sdk 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).

Run locally or on the platform

The same workflow file runs on your machine and on the platform, without code changes. During development, run it directly from your IDE or terminal. The pyproject.toml defines a script that calls the workflow, so you can run it like this:

uv run workflow

When you're ready for scheduled, tracked, or team-accessible runs, deploy the same file to the platform, with no code changes required. See Deploy a workflow for the deployment steps.

Next steps

• Workflow template: clone the reference workflows and run one in simulation
• Automate your workflows: write and run a workflow end to end
• Phase: the logical stages a workflow is composed of
• Runs: what happens when a workflow executes
• Input: parameterize a workflow at run time