UniteLabs

Your First Workflow

Clone the workflow template, run the hello-world example locally, and set up your own first workflow in under 15 minutes.

In this guide you'll clone the workflow template, run its hello-world example locally to confirm your environment works, and copy it into a new workflow of your own.

Prerequisites

  • Python 3.12 or newer
  • uv installed (latest)
  • A VS Code based IDE
  • A UniteLabs account with API credentials (only needed when you deploy — not for local runs)
  • Registry credentials in ~/.netrc, so uv can install the unitelabs-* packages (see SDK installation)

Clone the template

git clone https://gitlab.com/unitelabs/workflows/workflow-template.git
cd workflow-template

The template ships four reference workflows side-by-side:

workflow-template/
├── shared/                     # shared library (config, custom labware, reusable @step steps)
├── w01-hello-world/            # the example we'll use here
├── w02-liquid-handling/        # real liquid handling demo (runs in simulation)
├── w03-plateloc-sealer/        # human-in-the-loop demo
├── w04-tecan-fluent-control/   # Tecan FluentControl demo
└── scripts/deploy.py           # the deploy CLI

Each workflow directory is a standalone Python package with its own pyproject.toml, its own lockfile (uv.lock, where uv records the exact version of every dependency), and its own .venv. The shared/ library is consumed via a path dependency (a dependency that points at the local folder instead of a published package), so edits there are picked up by every workflow without a re-sync.

Sync the dependencies

Sync shared first, then the workflow you want to run:

uv sync --directory shared
uv sync --directory w01-hello-world

This creates w01-hello-world/.venv/ with the UniteLabs workflow engine, the UniteLabs SDK pins from the workflow's pyproject.toml, and shared installed editable.

Tour w01-hello-world

Every workflow has the same three pieces.

1. The package metadata — pyproject.toml

The interesting bit is [tool.unitelabs.workflow] — that's the contract with the deploy script:

w01-hello-world/pyproject.toml
[project]
name = "w01-hello-world"           # the workflow's slug
version = "1.0.1"                  # bump this when you release
description = "Simplest possible workflow. Logs SDK version info as a sanity check."
dependencies = [
    "shared",
    "unitelabs-sdk[automate]~=0.12.0",
    "unitelabs-labware",
    "unitelabs-liquid-handling",
]

[project.scripts]
workflow = "w01_hello_world.__main__:main"

[tool.unitelabs.workflow]
display_name = "[deploy-test] Hello World"     # what shows on the platform
entrypoint = "workflow.py:hello_world"         # <file>:<func> inside the package
tags = ["deploy-test"]

[tool.uv.sources]
shared = { path = "../shared", editable = true }

The deploy script reads [project].name, [project].version, and [tool.unitelabs.workflow] directly from this file. There's no central manifest to keep in sync.

2. The flow — src/w01_hello_world/workflow.py

w01-hello-world/src/w01_hello_world/workflow.py
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."""
    logger = get_logger()
    logger.info(f"Hello, {recipient_name}!")
    logger.info(f"UniteLabs SDK: {sdk_version}")

A workflow is just an async function decorated with @workflow. Larger workflows split into phases (also @phase-decorated) and steps (@step-decorated, often coming from shared/steps/) — see the Workflow → Phase → Step taxonomy.

3. The entrypoint — src/w01_hello_world/__main__.py

This is what uv run --directory w01-hello-world workflow calls — a thin wrapper that runs hello_world via asyncio.run.

Run it locally

uv run --directory w01-hello-world workflow

You should see the log output with the greeting and the SDK version. No credentials, no hardware.

The workflow console script is declared in each workflow's pyproject.toml under [project.scripts] — every workflow exposes the same entry, so the run command is identical regardless of which workflow you point at.

Create your own first workflow

The mechanical recipe: copy w01-hello-world to a new directory named w<NN>-<name> (the next free two-digit number plus a lowercase name with hyphens, here w05-sample-quality). Then rename every occurrence of the old name:

  1. The package directory: src/w01_hello_world/ becomes src/w05_sample_quality/.
  2. In pyproject.toml: [project].name, the [project.scripts] module path, and the [tool.unitelabs.workflow] entrypoint (all shown below).
  3. In src/w05_sample_quality/__main__.py: the import and the function it runs.
  4. Add the new folder to the folders array in workflow-template.code-workspace.

Then update the new workflow's pyproject.toml:

w05-sample-quality/pyproject.toml
[project]
name = "w05-sample-quality"
version = "0.1.0"
description = "Fetch samples and evaluate them against a concentration threshold."

[project.scripts]
workflow = "w05_sample_quality.__main__:main"   # match the new package name

[tool.unitelabs.workflow]
display_name = "Sample Quality"
entrypoint = "workflow.py:sample_quality"
tags = ["qc"]

And rewrite workflow.py with the flow of your own — for example, a quality check that fetches samples and applies a concentration threshold:

w05-sample-quality/src/w05_sample_quality/workflow.py
from unitelabs.sdk import get_logger, phase, step, workflow


@step(name="Fetch Samples")
async def fetch_samples(source: str) -> list[dict]:
    logger = get_logger()
    logger.info(f"Fetching samples from {source}")
    # Replace with your actual data retrieval.
    return [
        {"id": "sample_001", "concentration": 450.0},
        {"id": "sample_002", "concentration": 210.0},
        {"id": "sample_003", "concentration": 890.0},
    ]


@step(name="Evaluate Samples")
async def evaluate_samples(samples: list[dict], min_concentration: float) -> dict:
    passed = [s for s in samples if s["concentration"] >= min_concentration]
    failed = [s for s in samples if s["concentration"] < min_concentration]
    return {
        "passed": passed,
        "failed": failed,
        "pass_rate": len(passed) / len(samples) if samples else 0.0,
    }


@phase(name="Analyze Samples")
async def analyze_samples(source: str, min_concentration: float) -> dict:
    samples = await fetch_samples(source=source)
    return await evaluate_samples(samples=samples, min_concentration=min_concentration)


@workflow(name="Sample Quality")
async def sample_quality(
    source: str = "data_warehouse",
    min_concentration: float = 300.0,
) -> dict:
    """Fetch samples and evaluate them against a concentration threshold."""
    return await analyze_samples(source=source, min_concentration=min_concentration)

Sync and run:

uv sync --directory w05-sample-quality
uv run --directory w05-sample-quality workflow
Keep each @step focused on one responsibility — fetch, transform, persist, notify. Orchestrate them in the @phase, not inside the steps themselves. The UniteLabs workflow engine runs independent steps concurrently.

Next steps