flyte-sdk-author

Flyte 2 SDK Author Skill

Create Flyte 2 workflows, tasks, and apps from scratch using pure Python — no DSL.

Grounding References

Resource	URL
Official docs	https://www.union.ai/docs/v2/flyte
Docs index (LLMs)	https://www.union.ai/docs/v2/flyte/llms.txt
SDK API reference	https://www.union.ai/docs/v2/union/api-reference/flyte-sdk/
CLI API reference	https://www.union.ai/docs/v2/union/api-reference/flyte-cli/
flyte-sdk source	https://github.com/flyteorg/flyte-sdk
Example code	https://github.com/unionai/unionai-examples
Flyte MCP tools	Available via the flyte-cluster and flyte-docs MCP servers

Ground unfamiliar APIs in real examples. When unsure of a current Flyte 2 API, or for a pattern not shown below, and the flyte-docs search tools are available, search them first — by exact symbol (TaskEnvironment, flyte.io.File, map_task), since matching is literal substring, not semantic — then adapt a real example rather than inventing one, and cite the file or section you pulled it from. (Flyte 2 is not flytekit; priors are often wrong.)

Tool Priority

1. Flyte MCP — if the harness has Flyte MCP tools, prefer them for inspecting and discovering registered tasks, listing and interacting with runs, executing a task remotely, managing apps and triggers, and searching Flyte SDK examples and docs.
2. flyte CLI — for local commands: flyte run, flyte deploy, flyte serve, flyte create config, flyte start devbox
3. Python SDK — for anything the CLI/MCP can't do (custom task environments, type transformers, dynamic workflows, programmatic run control)

Core Patterns

TaskEnvironment + @env.task (recommended)

import flyte
import flyte.io

env = flyte.TaskEnvironment(
    name="training",
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "pandas", "torch", "transformers",
    ),
)

@env.task(retries=2, cache="auto")
async def preprocess(data: list[str]) -> flyte.io.File:
    # ETL: clean + write to remote storage
    ...
    return flyte.io.File(path="/tmp/output.parquet")

@env.task
async def train(data_path: flyte.io.File) -> flyte.io.File:
    # ML: model training
    ...
    return flyte.io.File(path="/tmp/model.pt")

@env.task
async def evaluate(model_path: flyte.io.File) -> dict:
    # ML: evaluation
    return {"accuracy": 0.95, "f1": 0.92}

@env.task
async def main(data: list[str]) -> dict:
    path = await preprocess(data)
    model = await train(path)
    return await evaluate(model)

if __name__ == "__main__":
    import asyncio
    result = asyncio.run(main(["a", "b", "c"]))
    print(result)