How Do You Write Workflows in Flyte 2?

Flyte 2 workflows are written as plain async functions in a supported language SDK — no DSL and no YAML. You declare a TaskEnvironment (image, resources, secrets, retry policy) once, then write tasks as ordinary functions and compose them with normal control flow. The runtime records inputs and outputs for every step, so a crash resumes from the last completed step instead of restarting the whole workflow. This applies if you are on Flyte 2 with one of the supported SDKs; the examples below use Python.

The two building blocks

The environment is defined once and reused, so image and resource settings are not repeated per task. Tasks stay ordinary functions — you can call them, test them, and run them locally without a cluster.

Writing a task

A task is an async function. Its signature is the contract: typed parameters in, typed return value out. Because the runtime records I/O per step, a failure mid-workflow resumes rather than re-running completed work.

env = TaskEnvironment(
    image="my-image:latest",
    resources=...,
    secrets=...,
    retry_policy=...,
)

@env.task
async def fetch_shard(shard_id: int) -> DataFrame:
    ...

@env.task
async def train(data: DataFrame) -> Model:
    ...

Composing with normal control flow

Branching, looping, and fan-out use the language's own constructs — if, for, asyncio.gather, and similar. There is no separate graph syntax to learn, and the runtime still records every step's I/O.

@env.task
async def pipeline(shards: list[int]) -> Model:
    results = await asyncio.gather(*(fetch_shard(s) for s in shards))
    merged = merge(results)
    if needs_more_data(merged):
        merged = await fetch_shard(extra_shard)
    return await train(merged)

Fan-out across shards, a conditional re-fetch, and a final training step are all expressed as ordinary Python.

Typed I/O between tasks

Data passed between tasks is type-checked using language-native types. Supported values include native types, files, DataFrames, and dataclasses. This means a mismatch between what one task returns and what the next expects surfaces as a type error rather than a runtime surprise.

Supported SDKs

Flyte 2 ships SDKs for several languages, so the same model — environment plus plain functions — applies beyond Python:

• Python — flyte-sdk
• TypeScript — flyte-sdk-ts
• Rust — flyte-sdk-rs
• Go — flyte-sdk-go
• Mojo — flyte-sdk-mojo

The Python SDK installs with pip install flyte.

Agents and serving use the same API

Native agents wrap tools as tasks, and every LLM call is traced. The same Python API also serves long-running apps next to your workflows, with native integrations for Streamlit, FastAPI, vLLM, SGLang, and Ollama — you declare the app, image, resources, and scaling, and Flyte handles the serving plumbing. That includes OpenAI-compatible LLM endpoints with model streaming and tensor parallelism, interactive dashboards served as long-running apps with auth, and scale-to-zero when idle via one line of config.

Verifying it works

The same code runs locally, on the devbox, or on your cluster, so you can validate a workflow before deploying. Run it locally first to confirm types and control flow, then move it to a cluster without rewriting it. If a task fails — an API rate limit, a network error, an OOM kill, a node interruption, a missing GPU, or a code exception — the retry policy in the TaskEnvironment governs recovery, and completed steps are not re-run.

Common sticking points

• Putting config on tasks instead of the environment. Image, resources, secrets, and retry policy belong in the TaskEnvironment so they are declared once.
• Expecting a graph definition. There is no separate workflow DSL to write; composition is just control flow.
• Assuming a crash restarts everything. Because I/O is recorded per step, recovery resumes from the failed step — but only if the work is expressed as tasks within the environment.