Orchestration engine for durable workflows
Orchestrate and observe computational workflows defined in plain Python.
Suitable for data pipelines, background tasks, agentic systems.

Docs · Studio · PyPI · Website

Why Coflux?

• Plain Python: workflows are regular Python functions with decorators - no DSLs, no YAML, no static DAGs.
• Low latency: millisecond task startup using warm executor processes.
• Real-time observability: watch workflows execute live using the CLI, or in Coflux Studio, with graph visualisation, logs, metrics, and results.
• Self-hosted: you run the server; your data stays in your infrastructure.
• Workspace inheritance: branch production into development workspaces and re-run individual steps with real data.

Quick example

A workflow is the entry point for a run. It can call tasks, which are the individual operations that Coflux orchestrates — scheduling them across workers and tracking results, dependencies, retries, etc. When you call a task from a workflow, Coflux intercepts the call and executes it remotely:

importcofluxascf@cf.task(retries=cf.Retries(3, when=ConnectionError))deffetch_data(url: str) ->dict:
returnrequests.get(url).json()
@cf.task(cache=True)deftransform(data: dict) ->list:
returnsorted(data["items"], key=lambdax: x["score"], reverse=True)
@cf.workflow()defmy_pipeline(url: str):
returntransform(fetch_data(url))

Tasks and workflows are just functions — call them directly in tests, or let Coflux orchestrate them across workers.

Features

Caching

Reuse results across runs, with TTLs, parameter filtering, and cross-workspace cache sharing:

# cache indefinitely@cf.task(cache=True)defget_user(user_id): ...
# cache for 10 minutes@cf.task(cache=600)deffetch_prices(): ...
# cache by specific params only@cf.task(cache=cf.Cache(params=["product_id"]))defget_product(product_id, include_reviews=False): ...

Retries with conditions

Retry on specific exceptions with configurable backoff:

@cf.task(retries=cf.Retries(5, backoff=(1, 60), when=TransientError))defcall_api():
...

Parallel execution

Submit tasks concurrently and collect results:

@cf.workflow()defprocess_order(user_id, product_id):
user=load_user.submit(user_id) # starts immediatelyproduct=load_product.submit(product_id) # starts immediatelycreate_order(user.result(), product.result()) # waits for results before calling

Assets

Share files and directories between tasks, with glob filtering and composition:

@cf.task()defgenerate_report() ->cf.Asset:
Path("report.csv").write_text(build_csv())
returncf.asset(match="*.csv")
@cf.workflow()defmy_workflow():
report=generate_report()
paths=report.restore() # download files locally

Memoisation

Memoise task calls within a single run (unlike caching, which works across runs). Useful for debugging workflows without repeating side effects — re-run steps within the workflow and memoised steps return their previous result:

@cf.task(memo=True)defsend_email(recipient):
mailer.send(recipient.email, ...)
@cf.workflow()defnotify(campaign_id):
forringet_recipients(campaign_id):
send_email.submit(r)

Metrics

Record numeric values from tasks and visualise them in Studio:

loss=cf.Metric("loss", group="training")
@cf.task()deftrain(epochs):
forepochinrange(epochs):
loss.record(run_epoch(epoch), at=epoch)

Track iteration progress with a built-in progress bar:

foritemincf.progress(items):
process(item)

Groups

Organise parallel tasks into groups:

@cf.workflow()defmap_reduce(n: int):
withcf.group("Process chapters"):
results= [process.submit(i) foriinrange(n)]
returnmerge([r.result() forrinresults])

Structured logging

cf.log_info("Processing {count} items for {user}", count=42, user="alice")

More features

• Debouncing: defer execution until a task stops being called (defer=True)
• Recurrence: automatically re-execute workflows for polling (recurrent=True)
• Suspense: pause a task and free resources while waiting (cf.suspend())
• Timeouts: kill executions that exceed a time limit (timeout=30)
• Worker pools: auto-launch and manage workers with Docker, process, or Kubernetes launchers

Getting started

1. Install the CLI

curl -fsSL https://coflux.com/install.sh | sh

Or download a binary from the releases page.

2. Start the server

Use the CLI to start the server:

coflux server --no-auth

Or run it with Docker.

3. Create a workflow

# myapp/workflows.pyimportcofluxascf@cf.task()defgreet(name: str) ->str:
returnf"Hello, {name}!"@cf.workflow()defhello(name: str):
print(greet(name))

4. Start a worker

coflux worker --dev myapp.workflows

The worker attempts to automatically detect your Python environment. If you have uv installed, the (dependency-less) coflux package is installed automatically. Otherwise, install it first with pip install coflux.

The --dev flag (equivalent to --watch --register) watches for code changes and automatically restarts the worker.

5. Submit a run

Arguments are JSON values. The module/target path uses / as a separator:

coflux submit myapp/hello '"world"'

6. Open Studio

Visit studio.coflux.com and create a project with your server address (localhost:7777) - a Studio account isn't required. Submit workflows runs and watch them execute in real time.