Running workflows from the SDK

Flux gives you two ways to run a workflow from Python: an in-process call using workflow.run(), and a remote call using FluxClient. The in-process path suits scripts, tests, and local iteration. The remote path is how production systems trigger workflows on a deployed server with its own worker pool.

In-process execution with workflow.run()

Call .run() directly on any decorated workflow. Flux executes the workflow in the current process using asyncio.run, persists state to a local .flux/ directory, and returns the completed ExecutionContext:

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


@task
async def add(a: int, b: int) -> int:
    return a + b


@workflow
async def compute_sum(ctx: ExecutionContext[dict]):
    data = ctx.input
    return await add(data["x"], data["y"])


ctx = compute_sum.run({"x": 10, "y": 32})
print(ctx.output)          # 42
print(ctx.has_succeeded)   # True
print(ctx.execution_id)    # e.g. "005d4b2f..."

.run() is synchronous from the caller’s perspective — it blocks until the workflow finishes and returns the final ExecutionContext. The working directory must be writable; Flux creates .flux/ automatically on the first call.

Reading the result

After .run() returns, the context object holds everything about that execution:

Structured inputs

Pass any dill-serializable value as the argument to .run(). Use a dataclass when the input has multiple fields; it keeps the call site readable and lets static analysis infer ctx.input:

from dataclasses import dataclass
from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow


@dataclass
class IngestRequest:
    source: str
    batch_size: int = 100


@task
async def fetch_batch(source: str, size: int) -> list[str]:
    return [f"{source}-record-{i}" for i in range(size)]


@task
async def summarize(records: list[str]) -> dict:
    return {"count": len(records), "source": records[0].split("-")[0]}


@workflow
async def ingest(ctx: ExecutionContext[IngestRequest]):
    req = ctx.input
    records = await fetch_batch(req.source, req.batch_size)
    return await summarize(records)


ctx = ingest.run(IngestRequest(source="warehouse", batch_size=5))
print(ctx.output)   # {'count': 5, 'source': 'warehouse'}

Remote execution with FluxClient

When your workflow runs on a Flux server, use FluxClient to trigger executions over HTTP. FluxClient is an async HTTP client; use it inside an async function or an async context manager:

import asyncio
from flux.client import FluxClient

async def main():
    async with FluxClient("http://localhost:8000") as client:
        # Fire-and-forget: returns immediately with an execution record
        result = await client.run_workflow("ingest", {"source": "s3://bucket", "batch_size": 500})
        print(result["execution_id"])   # use this to track progress
        print(result["state"])          # "SCHEDULED" or "RUNNING"

asyncio.run(main())

The workflow_ref argument accepts either a bare name ("ingest") or a namespaced form ("billing/invoice"). If no namespace is given, Flux uses "default".

Async vs sync trigger

Two methods start a workflow execution:

Use run_workflow for long-running workflows and anything you want to track asynchronously. Use run_workflow_sync for short, latency-sensitive calls or when you need the result inline.

Polling for completion

When you trigger with run_workflow, use get_execution to check status:

import asyncio
from flux.client import FluxClient

async def run_and_poll():
    async with FluxClient("http://localhost:8000") as client:
        result = await client.run_workflow("ingest", {"source": "s3://bucket", "batch_size": 200})
        eid = result["execution_id"]

        while True:
            execution = await client.get_execution(eid)
            state = execution["state"]
            print(f"state: {state}")

            if state in ("COMPLETED", "FAILED", "CANCELLED"):
                break

            await asyncio.sleep(2)

        print("final output:", execution.get("output"))

asyncio.run(run_and_poll())

Pass detailed=True to get_execution to receive the full event log alongside the summary. The event log includes per-task start and end times, inputs, outputs, and any captured exceptions.

Cancelling an execution

cancel_execution stops a running workflow and requires both the workflow reference and the execution ID:

async with FluxClient("http://localhost:8000") as client:
    await client.cancel_execution("ingest", execution_id)

The server transitions the execution to CANCELLING, signals the worker, and the execution reaches CANCELLED once the worker unwinds. Calling cancel_execution on an already-finished execution raises an HTTP error.

In-process vs server: when to use each

Both paths share the same workflow code, event log, and ExecutionContext model. A workflow written and tested with .run() deploys to a server without modification.

What’s next

• Running workflows from the CLI — trigger and inspect executions using the flux workflow run command.
• Running workflows from the REST API — call the HTTP endpoints directly from any language.
• Replays and re-execution — re-run a completed execution by ID, replaying its event log without re-executing finished tasks.
• Manual cancellation — stop an in-flight execution from the CLI, REST API, or SDK.