Your First Workflow
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
uvinstalled (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 theunitelabs-*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:
[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
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 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 and 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.
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:
- The package directory:
src/w01_hello_world/becomessrc/w05_sample_quality/. - In
pyproject.toml:[project].name, the[project.scripts]module path, and the[tool.unitelabs.workflow]entrypoint (all shown below). - In
src/w05_sample_quality/__main__.py: the import and the function it runs. - Add the new folder to the
foldersarray inworkflow-template.code-workspace.
Then update the new workflow's 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:
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
@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.Read workflow parameters from the context object
Every argument sample_quality_flow is called with — source and min_concentration — is automatically published to the run's context by @workflow. Instead of passing min_concentration down through evaluate_samples_task's own signature, the step can read it directly:
from unitelabs.sdk import get_context, get_logger, step, workflow
@step(name="Evaluate Samples")
async def evaluate_samples_task(samples: list[dict]) -> dict:
min_concentration = get_context().workflow_parameters["min_concentration"]
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,
}
@workflow(name="Sample Quality")
async def sample_quality_flow(
source: str = "data_warehouse",
min_concentration: float = 300.0,
) -> dict:
"""Fetch samples and evaluate them against a concentration threshold."""
samples = await fetch_samples_task(source=source)
return await evaluate_samples_task(samples=samples)
This matters once a parameter needs to reach a step nested a few calls deep: without context, adding it means editing every intermediate phase's and step's signature along the way; with context, only the workflow's own signature changes. See Run context and Input for the full picture.
Next steps
- Deploy your workflow — bundle and ship with
scripts/deploy.py. - Set up CI/CD — automate dev/stg/prd deploys on GitHub Actions or GitLab CI.
- Add error handling — make failures surface cleanly.
- Add human-in-the-loop steps — use W03 as the reference pattern.