coflux

KO | EN

gitlite — search

#javascript #python #hacktoberfest #react #ai #typescript #llm #go #golang #android #machine-learning #rust #deep-learning #linux

★ 15 Open GitHub ↗

Orchestration engine for durable workflows - orchestrate and observe computational workflows defined in plain Python.

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

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

<a href="https://docs.coflux.com">Docs</a> &middot;
 <a href="https://studio.coflux.com">Studio</a> &middot;
 <a href="https://pypi.org/project/coflux/">PyPI</a> &middot;
 <a href="https://coflux.com">Website</a>

## 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](https://studio.coflux.com), 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:

```python
import coflux as cf

@cf.task(retries=cf.Retries(3, when=ConnectionError))
def fetch_data(url: str) -> dict:
    return requests.get(url).json()

@cf.task(cache=True)
def transform(data: dict) -> list:
    return sorted(data["items"], key=lambda x: x["score"], reverse=True)

@cf.workflow()
def my_pipeline(url: str):
    return transform(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:

```python
# cache indefinitely
@cf.task(cache=True)
def get_user(user_id): ...

# cache for 10 minutes
@cf.task(cache=600)
def fetch_prices(): ...

# cache by specific params only
@cf.task(cache=cf.Cache(params=["product_id"]))
def get_product(product_id, include_reviews=False): ...
```

### Retries with conditions

Retry on specific exceptions with configurable backoff:

```python
@cf.task(retries=cf.Retries(5, backoff=(1, 60), when=TransientError))
def call_api():
    ...
```

### Parallel execution

Submit tasks concurrently and collect results:

```python
@cf.workflow()
def process_order(user_id, product_id):
    user = load_user.submit(user_id)              # starts immediately
    product = load_product.submit(product_id)     # starts immediately
    create_order(user.result(), product.result()) # waits for results before calling
```

### Assets

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

```python
@cf.task()
def generate_report() -> cf.Asset:
    Path("report.csv").write_text(build_csv())
    return cf.asset(match="*.csv")

@cf.workflow()
def my_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:

```python
@cf.task(memo=True)
def send_email(recipient):
    mailer.send(recipient.email, ...)

@cf.workflow()
def notify(campaign_id):
    for r in get_recipients(campaign_id):
        send_email.submit(r)
```

### Metrics

Record numeric values from tasks and visualise them in Studio:

```python
loss = cf.Metric("loss", group="training")

@cf.task()
def train(epochs):
    for epoch in range(epochs):
        loss.record(run_epoch(epoch), at=epoch)
```

Track iteration progress with a built-in progress bar:

```python
for item in cf.progress(items):
    process(item)
```

### Groups

Organise parallel tasks into groups:

```python
@cf.workflow()
def map_reduce(n: int):
    with cf.group("Process chapters"):
        results = [process.submit(i) for i in range(n)]
    return merge([r.result() for r in results])
```

### Structured logging

```python
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

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

Or download a binary from the [releases page](https://github.com/bitroot/coflux/releases).

### 2. Start the server

Use the CLI to start the server:

```bash
coflux server --no-auth
```

Or [run it with Docker](https://docs.coflux.com/getting_started/server).

### 3. Create a workflow

```python
# myapp/workflows.py

import coflux as cf

@cf.task()
def greet(name: str) -> str:
    return f"Hello, {name}!"

@cf.workflow()
def hello(name: str):
    print(greet(name))
```

### 4. Start a worker

```bash
coflux worker --dev myapp.workflows
```

The worker attempts to automatically detect your Python environment. If you have [`uv`](https://docs.astral.sh/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:

```bash
coflux submit myapp/hello '"world"'
```

### 6. Open Studio

Visit [studio.coflux.com](https://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.

coflux

Explore Similar Repositories

Was this content helpful?

Recent Feedback

Download README