Workflow inputs and outputs

Every workflow receives a single input value through ctx.input and produces a single output through its return statement. Flux serializes both with dill (a pickle-compatible serializer) and stores them in the execution record. Knowing where the defaults hold and where they break down lets you write workflows with data contracts that stay explicit and appropriately sized.

Typing workflow inputs

ExecutionContext[T] is generic. The type parameter T is the type of the value you pass to .run(), and it flows through to ctx.input so static analysis and your editor can infer the correct shape.

Using a dataclass

dataclass has the smallest footprint. Fields are documented, defaults work naturally, and dill serializes them without extra configuration:

from dataclasses import dataclass

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


@dataclass
class AnalysisRequest:
    text: str
    language: str = "en"


@task
async def tokenize(text: str, lang: str) -> list[str]:
    return text.lower().split()


@workflow
async def analyze(ctx: ExecutionContext[AnalysisRequest]):
    req = ctx.input
    tokens = await tokenize(req.text, req.language)
    return {"tokens": tokens, "count": len(tokens)}


if __name__ == "__main__":
    ctx = analyze.run(AnalysisRequest(text="Hello durable world", language="en"))
    print(ctx.output)         # {'tokens': ['hello', 'durable', 'world'], 'count': 3}
    print(ctx.has_succeeded)  # True

The @dataclass decorator is all Flux needs. No registration, no schema declaration — the workflow passes req.text and req.language to its task exactly as you wrote them.

Using a Pydantic model

Pydantic models work as inputs too — field validation runs at construction time, before the workflow ever starts:

from pydantic import BaseModel

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


class SearchRequest(BaseModel):
    query: str
    max_results: int = 10
    include_archived: bool = False


@task
async def run_search(query: str, limit: int) -> list[str]:
    return [f"result-{i}: {query}" for i in range(limit)]


@workflow
async def search_workflow(ctx: ExecutionContext[SearchRequest]):
    req = ctx.input
    results = await run_search(req.query, req.max_results)
    return {"query": req.query, "hits": results}


if __name__ == "__main__":
    req = SearchRequest(query="durable execution", max_results=3)
    ctx = search_workflow.run(req)
    print(ctx.output["query"])      # durable execution
    print(len(ctx.output["hits"]))  # 3
    print(ctx.has_succeeded)        # True

The practical difference between dataclass and BaseModel for workflow inputs is validation. A dataclass accepts whatever Python gives it; a Pydantic model validates field types and raises ValidationError before execution begins. Use Pydantic when the input crosses a boundary (REST API, CLI, external queue) and you want a clean error before any durable state is written.

Typing workflow outputs

A workflow can return any dill-serializable value. Return a dataclass when the output has multiple fields — it keeps the call site readable and gives static analysis something to work with:

from dataclasses import dataclass

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


@dataclass
class PipelineResult:
    raw_count: int
    processed_count: int
    summary: str


@task
async def count_records(data: list[str]) -> int:
    return len(data)


@task
async def process(data: list[str]) -> list[str]:
    return [item.strip().lower() for item in data]


@workflow
async def etl_pipeline(ctx: ExecutionContext[list[str]]) -> PipelineResult:
    data = ctx.input
    raw_count = await count_records(data)
    processed = await process(data)
    processed_count = await count_records(processed)
    return PipelineResult(
        raw_count=raw_count,
        processed_count=processed_count,
        summary=f"Processed {processed_count} of {raw_count} records.",
    )


if __name__ == "__main__":
    ctx = etl_pipeline.run(["  Apple  ", "banana", "  Cherry"])
    result = ctx.output
    print(result.summary)          # Processed 3 of 3 records.
    print(result.processed_count)  # 3

Pydantic models work as return types with the same caveat: they serialize correctly from imported modules, and produce PicklingWarning messages when the class is defined in __main__.

How Flux serializes inputs and outputs

Flux uses dill to serialize workflow inputs, task outputs, and the workflow’s final result. The default serializer identifier is pkl — visible in flux.toml under [flux] serializer = "pkl".

Two serializer modes are available:

The pkl default means you can return almost any Python value. The tradeoff is that the result is a binary blob: not human-readable, and not portable outside the Python version that wrote it. If your workflow’s output needs to be inspected by other tools or stored long-term, switch to json and constrain your return type accordingly.

To override the serializer globally, set the environment variable:

FLUX_SERIALIZER=json flux start server

Or add it to flux.toml:

[flux]
serializer = "json"

Storing large outputs externally

By default, Flux keeps every output inline in the execution record. That works for small results: a count, a status string, a short list. For large outputs (DataFrames, embeddings, binary artifacts), storing them in the database row degrades query performance and inflates backup sizes.

LocalFileStorage writes the serialized output to disk and stores only a lightweight OutputStorageReference in the database. The workflow’s ctx.output then holds that reference rather than the data itself:

from flux import ExecutionContext
from flux.task import task
from flux.workflow import workflow
from flux.output_storage import LocalFileStorage


file_storage = LocalFileStorage()


@task.with_options(output_storage=file_storage)
async def generate_large_output(n: int) -> list[int]:
    return list(range(n))


@workflow.with_options(output_storage=file_storage)
async def large_workflow(ctx: ExecutionContext[int]):
    data = await generate_large_output(ctx.input)
    return data


if __name__ == "__main__":
    ctx = large_workflow.run(100)
    print(type(ctx.output).__name__)  # OutputStorageReference
    print(ctx.has_succeeded)         # True

LocalFileStorage reads its base path from settings.local_storage_path (.flux/.data by default) and its serializer from settings.serializer. Files are named {workflow_name}_{execution_id}.pkl (or .json). The workflow’s reference_id is f"{ctx.workflow_name}_{ctx.execution_id}". To retrieve the data, call file_storage.retrieve(ctx.output).

Switching LocalFileStorage to JSON

Pass FLUX_SERIALIZER=json before starting the server (or set it in flux.toml) and LocalFileStorage will write .json files instead of .pkl. The reference’s metadata field records which serializer was used at write time, so retrieval always uses the right format regardless of the current setting:

# With FLUX_SERIALIZER=json in the environment:
ctx = json_workflow.run({"key": "value"})
print(ctx.output.storage_type)  # local_file
print(ctx.output.metadata)      # {'serializer': 'json'}

Custom output storage backends

OutputStorage is an abstract base class with three methods: store, retrieve, and delete. Implement all three to create a custom backend — an S3 bucket, a GCS blob, a Redis key. The built-in LocalFileStorage shows the full pattern. See Operate: Storage backends for production backend configurations.

What’s next

• Defining workflows — the @workflow decorator, ExecutionContext, and how to run a workflow in-process.
• Composing workflows — call one workflow from another and fan out in parallel.
• Operate: Storage backends — S3, GCS, and custom backends for production deployments.