Workflow Orchestration
A workflow written in a scheduler GUI, a spreadsheet, or an instrument script lives outside your normal software practices. It can't be version-controlled, reviewed in a PR, tested automatically, or deployed reliably across locations.
UniteLabs lets you define workflows as Python code. They run on the UniteLabs automation engine, are tracked end-to-end, and behave exactly the same whether you trigger them once or a thousand times.
What this unlocks
- Workflows in Git: version-control your experiment logic alongside your analysis code
- CI/CD compatible: test workflow logic against simulated devices in CI before running on real hardware
- Reproducible runs: every run is tracked with inputs, logs, and output artifacts
- Workcell-aware scheduling (WIP): the engine tracks which instruments are in use; idle devices can be picked up by other workflows running in the same workcell, increasing throughput without manual scheduling
- Unattended operation: reliable recovery and retry logic suitable for overnight or weekend runs
How it works
A workflow is a Python class that defines phases — ordered groups of steps that coordinate instruments, data, and logic. The orchestrator handles scheduling, device allocation, and run tracking.
import asyncio
from prefect import flow
from unitelabs.sdk import AsyncApiClient
@flow() # phase
async def seed_plate():
client = AsyncApiClient()
hamilton = await client.get_service_by_name("Hamilton STAR")
await hamilton.initialize()
... # dispense cells into 96-well plate
@flow() # phase
async def add_reagent():
client = AsyncApiClient()
hamilton = await client.get_service_by_name("Hamilton STAR")
... # dispense CCK-8 reagent
@flow() # phase
async def incubate(hours: float = 2.0):
client = AsyncApiClient()
incubator = await client.get_service_by_name("CO2 Incubator")
... # hold at 37 °C for the given duration
@flow() # phase
async def measure() -> dict:
client = AsyncApiClient()
reader = await client.get_service_by_name("Plate Reader")
... # read absorbance at 450 nm
return results
@flow(name="Cell Viability Assay") # workflow
async def cell_viability_assay():
await seed_plate()
await incubate(hours=24.0) # cells settle overnight
await add_reagent()
await incubate(hours=2.0) # CCK-8 reaction window
return await measure()
asyncio.run(cell_viability_assay())
Workflows are registered with the platform and can be triggered via the REST API, the Platform UI, or on a schedule.
Workcell scheduling
The orchestrator knows which instruments each running workflow has claimed. When a device finishes its current phase and becomes idle, the engine can make it available to another workflow in the same workcell — no manual coordination needed.
When to use this
Workflow orchestration is the right approach when:
- You need reproducible, auditable runs (GxP, SOPs, tech transfer)
- You want to run a protocol repeatedly with different inputs
- Your workcell is shared across multiple teams or experiments
- You need reliable unattended or long-running operations