Flyte

View saved version →

---
name: deploy-flyte-kind
description: Deploy a complete Flyte stack (flyte-binary + a hosted PostgreSQL + an object store) onto a kind cluster, running on the user's own machine or a DigitalOcean VM (droplet). PostgreSQL is hosted (Supabase or external); the object store is AWS S3 or Cloudflare R2. Use when the user wants to run Flyte on kind — either reusing an existing kind cluster or creating a new one. For evaluation only (no TLS/auth on the base deployment).
---

# Deploy Flyte to a kind cluster

Stand up Flyte on a [kind](https://kind.sigs.k8s.io/) cluster: the flyte-binary
plus a hosted PostgreSQL and an S3-compatible object store. For **evaluation
only** — no TLS, no auth, static credentials.

kind runs anywhere Docker runs, so the cluster can live on the user's **own
machine** (default) or a **DigitalOcean VM** (droplet) — the host is a choice
made in Step 0.

The PostgreSQL and object store are independent choices the user makes in Step 2:

- **PostgreSQL** — **Supabase** or another external/self-hosted PostgreSQL.
- **Object store** — **AWS S3** or **Cloudflare R2**.

Both are hosted; kind runs only the flyte-binary. The user supplies connection
details for each.

## Step 0: Choose the host, check prerequisites, and check for an existing cluster

**First, ask the user where kind should run** (use `AskUserQuestion`):

- **the user's own machine** (default), or
- a **DigitalOcean VM** (droplet) — the only cloud-VM host this skill supports.

Do **not** offer or hand-roll AWS EC2 or GCP VM setups; for a real cloud
deployment, point the user at the AWS deployment skill instead.

If the user picks the droplet, **every `kind`, `kubectl`, and `helm` command
below runs on the droplet** (over SSH) — only the SDK/CLI and browser run on the
user's own machine. Provision it and install the tools there first (needs a few
GB of headroom for kind, so ≥ 4 vCPU / 8 GB):

```bash
# create the droplet (dashboard or doctl)
doctl compute droplet create flyte-kind \
  --image ubuntu-24-04-x64 --size s-4vcpu-8gb --region nyc1 \
  --ssh-keys <your-ssh-key-id>

# SSH in and install Docker, kind, kubectl, helm ON the droplet
ssh root@<droplet-ip>
curl -fsSL https://get.docker.com | sh
curl -Lo /usr/local/bin/kind \
  https://github.com/kubernetes-sigs/kind/releases/latest/download/kind-linux-amd64 \
  && chmod +x /usr/local/bin/kind
curl -Lo /usr/local/bin/kubectl \
  "https://dl.k8s.io/release/$(curl -Ls https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl" \
  && chmod +x /usr/local/bin/kubectl
curl -fsSL https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash
```

> [!WARNING] A cloud VM is exposed to the internet
> On a droplet the stack is reachable from the public internet. Restrict ports
> `80`, `443`, and `22` to the user's own IP with a
> [cloud firewall](https://docs.digitalocean.com/products/networking/firewalls/)
> while they evaluate.

Then verify the required tools **on whichever host runs kind** (run the check on
the droplet over SSH if that's the host), and decide whether to create a cluster
or reuse one:

```bash
for t in docker kind kubectl helm; do command -v $t >/dev/null || echo "MISSING: $t"; done
kind get clusters
```

- If any tool is `MISSING`, stop and tell the user to install it
  ([docker](https://docs.docker.com/get-docker/),
  [kind](https://kind.sigs.k8s.io/docs/user/quick-start/#installation),
  [kubectl](https://kubernetes.io/docs/tasks/tools/),
  [helm](https://helm.sh/docs/intro/install/)).
Read the `kind get clusters` output literally — it is the exact list of existing
cluster names, one per line. Do **not** assume a `flyte` cluster exists because
this guide uses that name; only treat it as present if `flyte` appears verbatim
in the output. A line like `kind` is a cluster named `kind`, not `flyte`.

- **`flyte` is in the list** → reuse it: skip Step 1, use `--context kind-flyte`
  below.
- **`flyte` is not in the list** → create it in Step 1. (Don't silently reuse a
  differen

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: deploy-flyte-kind-vm
description: Deploy Flyte on a kind cluster onto any host — your local machine, or a fresh cloud VM (DigitalOcean, AWS EC2, or GCP Compute Engine). Provisions the VM (firewall scoped to your IP), installs Docker/kind/kubectl/helm, then runs the kind Flyte deploy on that host and tunnels access back to your machine. Use when the user wants Flyte-on-kind but hasn't decided where to run it, or wants it on a cloud VM rather than locally. For evaluation only (single-node kind, static credentials). Delegates the Flyte install itself to the deploy-flyte-kind skill.
---

# Deploy Flyte on kind — local machine or a cloud VM

kind runs anywhere Docker runs, so the same Flyte-on-kind deploy works on your
own machine **or** on a cloud VM. This skill picks the **host**, provisions it if
it's a cloud VM, installs the prerequisites, and then runs the actual Flyte
deploy — the cluster, the hosted PostgreSQL + object store, the helm install, and
optional auth — by handing off to the **`deploy-flyte-kind`** skill. The only
things that change per host are *where the commands run* and *how you reach the
API afterward*.

> **For evaluation only.** Single-node kind, static credentials, no workload
> identity. On a cloud VM the stack is reachable from the public internet, so the
> provisioning steps below **restrict inbound 80/443 (and 22) to your own IP**.
> For production, use the `flyte-deploy-aws` skill instead.

This skill is guided: several steps need **human-in-the-loop** input — which host,
confirming billable VM creation, SSH keys, and (in the deploy hand-off) the
Supabase / R2 / S3 credentials. Ask; never invent account IDs, IPs, keys, or
passwords.

## Step 0: Choose the host

Ask the user where to run the cluster (use `AskUserQuestion`):

- **Local machine** — kind runs in your local Docker. Simplest; nothing to
  provision.
- **DigitalOcean** — a Droplet, provisioned with `doctl`.
- **AWS EC2** — an instance, provisioned with the `aws` CLI.
- **GCP Compute Engine** — an instance, provisioned with `gcloud`.

Then:

- **Local machine** → skip to **Step 3** (nothing to provision). The whole deploy
  runs locally.
- **A cloud VM** → do **Step 1** (provision) and **Step 2** (define how remote
  commands run), then Step 3.

kind needs a few GB of headroom, so any VM should be **at least 4 vCPU / 8 GB**.

## Step 1: Provision the cloud VM (skip for local)

First confirm the provider CLI is installed and authenticated locally, since these
commands run on **your** machine:

```bash
# DigitalOcean
command -v doctl >/dev/null && doctl account get >/dev/null 2>&1 || echo "doctl: install and run 'doctl auth init'"
# AWS
command -v aws   >/dev/null && aws sts get-caller-identity >/dev/null 2>&1 || echo "aws: install and configure credentials"
# GCP
command -v gcloud >/dev/null && gcloud auth list >/dev/null 2>&1 || echo "gcloud: install and run 'gcloud auth login'"
```

If the CLI is missing or unauthenticated, stop and have the user set it up (an
interactive login like `gcloud auth login` is easiest run by the user — suggest
they type `! gcloud auth login` so it runs in this session). **Creating a VM
incurs cost — confirm with the user before running any `create`/`run-instances`
command,** and ask for the values it needs (SSH key ID / key-pair name / zone).

Pick the tab for the chosen provider. Each: create a firewall/security rule
scoped to the user's IP, create the VM, SSH in, install Docker + kind + kubectl +
helm.

### DigitalOcean

Setting up `doctl` if it's missing or unauthenticated (`brew install doctl` on
macOS): auth is interactive by default, but if the user pastes an API token it
can be done non-interactively with `doctl auth init -t <token>`. The SSH key must
be **imported to DigitalOcean** before it can be passed to `droplet create`:

```bash
doctl compute ssh-key list                 # already imported? use its ID below
doctl compute ssh-key import <name> --public-key-file ~/.ssh/id_ed25519.pub   # prin

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-deploy-aws
description: 'Use when deploying a Flyte v2 (flyte-binary / flyte2) cluster on AWS from scratch — provisions EKS + S3 + RDS PostgreSQL + AWS Load Balancer Controller, then helm-installs the flyte-binary chart behind an ALB, with optional TLS and Okta SSO. Trigger words: "deploy flyte", "flyte v2 on AWS", "flyte EKS".'
---

# Deploying Flyte v2 on AWS (EKS + RDS + S3 + ALB)

Flyte v2 ships as a single unified binary (`flyte-binary-v2`) plus a separate console
image. One HTTP ingress serves the console (`/v2`), the `flyteidl2.*` Connect API, and
auth-discovery — there is no separate gRPC port. You scale it vertically.

The chart does NOT provision infrastructure. Stand up four things first:
**EKS cluster, S3 bucket, PostgreSQL (RDS), and (for external access) an ingress
controller.** This skill does all four with `eksctl` + `aws` + `helm`, then installs the
`flyte-binary` chart (the v2 chart; defaults to `flyte-binary-v2` + `flyteconsole-v2`).

**Get the chart first.** Install from the published Helm repo — this is what the official
docs do, and the released chart pins the Flyte image tag to the chart version (see Image
selection in Step 5). Also `helm pull --untar` a local copy so the TaskAction CRD file is on
disk for the Step 5 idempotency check:
```bash
helm repo add flyteorg https://flyteorg.github.io/flyte && helm repo update
helm pull flyteorg/flyte-binary --untar   # ./flyte-binary/templates/crds/flyte.org_taskactions.yaml now resolves
```
(Alternatively clone the repo — `git clone https://github.com/flyteorg/flyte` — and install
from the local `./charts/flyte-binary` path for the bleeding-edge chart; its default image tag
is a floating `:latest`, so pair it with `pullPolicy: Always` — see Image selection.) Official
docs: https://www.union.ai/docs/v2/flyte/oss-deployment/aws-deployment/. Validated end-to-end
on EKS.

> Replace every placeholder in angle brackets and the example hostnames/IDs with your own.

## Prerequisites & decisions

- CLIs: `aws` v2, **`eksctl` ≥ 0.227** (older caps out at k8s 1.29 — see gotcha), `kubectl`, `helm`, `jq`.
- Admin (or EKS+RDS+IAM+S3+EC2) creds. STS/SSO works — export the 3 env vars + region.
- eksctl writes the kubeconfig context (e.g. `<user>@flyte-v2.<region>.eksctl.io`). Pass
  `kubectl --context <ctx>` (and `helm --kube-context <ctx>`) per command rather than
  `kubectl config use-context` — that way you don't mutate the operator's current context.
- Decide: region, name prefix, and **exposure**: ALB+TLS needs a Route53 zone + ACM cert;
  **ALB HTTP-only needs neither** (reached at the auto `*.elb.amazonaws.com` name) — the
  simplest default when you own no domain. (This skill provisions **RDS PostgreSQL** for the DB;
  an in-cluster Postgres is out of scope here — RDS is assumed by Steps 3–5.)

**Persist your variables.** This deploy spans many commands and derives values you can't
recover later — most critically the **random `DBPW`** (Step 3), plus `ACCT`, `BUCKET`,
`RDS_HOST`, `IRSA_ARN`, etc. If your shell state resets between steps (or your AWS session
token expires and you re-auth in a fresh shell), these are gone. Keep them in a file you
re-`source` at the start of every step, and append each derived value as you compute it:
```bash
export AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... AWS_SESSION_TOKEN=...
export AWS_DEFAULT_REGION=us-west-2
ENVF=~/flyte-deploy.env                                   # source this at every step
{ echo "export PREFIX=flyte-v2 REGION=us-west-2 CLUSTER=flyte-v2"
  echo "export ACCT=$(aws sts get-caller-identity --query Account --output text)"  # confirm the RIGHT account
} >> $ENVF && source $ENVF
# As you create infra, append its outputs, e.g.:  echo "export DBPW='$DBPW' RDS_HOST=$RDS_HOST" >> $ENVF
# Check for an existing domain/cert (empty => go ALB HTTP-only):
aws route53 list-hosted-zones --query 'HostedZones[].Name' --output text
aws acm list-certificates --region $REGION --query 'CertificateSummaryList[].Domai

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-migrate
description: Entry-point orchestrator for porting Flyte 1 (flytekit) code to Flyte 2 (flyte). Explains the v1 to v2 shift, the terminology mapping, a recommended migration strategy, hybrid v1/v2 pipelines, and routes to sibling migration skills. Use when the user wants to migrate, port, or upgrade Flyte 1 (flytekit) code to Flyte 2. Trigger words are migrate, flytekit, v1 to v2, port, upgrade, convert workflow.
---

# Flyte 1 to 2 Migration Skill

This is the entry point for migrating a Flyte 1 (`flytekit`) codebase to Flyte 2 (`flyte`). Flyte 2 is a fundamental shift: there is no `@workflow` decorator, everything is a `@env.task`, orchestration runs as real Python at runtime, and parallelism is expressed with `asyncio`. This skill explains the overall shift, gives a recommended migration strategy, covers hybrid v1/v2 pipelines during the transition, and routes to the sibling skills that handle each theme in depth.

## Grounding References

| Resource | URL |
|---|---|
| Migration guide | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/ |
| 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/ |
| Example code | https://github.com/unionai/unionai-examples |
| Flyte MCP tools | Available via `flyte-mcp` server |

## The overall v1 to v2 shift

Two conceptual shifts motivate almost every change — **pure Python execution** and the **asynchronous model** — after which most migrations come down to a couple of mechanical moves.

- **`flytekit` (package) becomes `flyte`.** Imports change from `import flytekit` to `import flyte`.
- **`pyflyte` (CLI) becomes `flyte`.** The command-line tool was renamed.
- **Everything is a task.** In Flyte 1, `@workflow` functions were constrained to a DSL subset of Python that compiled to a static DAG. In Flyte 2 there is **no `@workflow` decorator**: everything is a `@env.task`, and a "workflow" is simply a task that calls other tasks. Loops, conditionals, and `try`/`except` work anywhere.
- **Async is the parallelism model.** Flyte 2 is built on `asyncio`, with the Flyte orchestrator acting as the event loop, scheduling awaited tasks across distributed infrastructure. `await` signals where a task can be scheduled in parallel, and `asyncio.gather` tells the orchestrator that a set of tasks are independent.

### Simplified API mapping

| Use case | Flyte 1 | Flyte 2 |
| --- | --- | --- |
| Environment management | `N/A` | `TaskEnvironment` |
| Perform basic computation | `@task` | `@env.task` |
| Combine tasks into a workflow | `@workflow` | `@env.task` |
| Create dynamic workflows | `@dynamic` | `@env.task` |
| Fanout parallelism | `flytekit.map` | Python `for` loop with `asyncio.gather` |
| Conditional execution | `flytekit.conditional` | Python `if-elif-else` |
| Catching workflow failures | `@workflow(on_failure=...)` | Python `try-except` |

## Terminology and concept mapping

Several Flyte 1 concepts were renamed or reshaped in Flyte 2. The table below maps the ones you'll meet most often.

| Flyte 1 | Flyte 2 | Notes |
|---|---|---|
| `flytekit` (package) | `flyte` (package) | The Python SDK was renamed; imports change from `import flytekit` to `import flyte`. |
| `pyflyte` (CLI) | `flyte` (CLI) | The command-line tool was renamed. |
| `@task` / `@workflow` / `@dynamic` | `@env.task` | A single task decorator off a `flyte.TaskEnvironment`. Workflows and dynamic tasks are no longer distinct constructs: everything is a task, and orchestration is plain Python. |
| `map_task()` | `flyte.map()` | Plus `asyncio.gather()` for async fan-out. |
| `conditional()` | native `if` / `elif` / `else` | Branching is now ordinary Python control flow, not a DSL. |
| `ImageSpec` | `flyte.Image` | Container image definition. |
| `current_context()` | `flyte.ctx()` | Runtime context access. |
| `FlyteFile` / `FlyteDirectory` | `fl

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-migrate-config
description: Migrates Flyte 1 task configuration, container images, resources, caching, secrets, scheduling, and CLI/config-file usage to Flyte 2 equivalents. Use when migrating Flyte 1 task configuration, images, resources, secrets, scheduling, or CLI/config-file usage to Flyte 2. Trigger words include resources, ImageSpec, cache_version, secrets, LaunchPlan, CronSchedule, pyflyte, config, register, and deploy.
---

# Flyte 1 to Flyte 2 Migration: Task Configuration and CLI/Config

In Flyte 1, image, resources, caching, secrets, and scheduling were configured per-task on the `@task` decorator or per-workflow on a `LaunchPlan`. In Flyte 2 most of this moves to the `flyte.TaskEnvironment`, so it is declared once and shared. The CLI is renamed from `pyflyte` to `flyte` and the config file is trimmed down. This skill covers migrating those settings and commands.

## Grounding References

| Resource | URL |
|---|---|
| Migration guide (Task configuration) | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/configuration/ |
| Migration guide (CLI) | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/cli-and-configuration/ |
| 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/ |
| Example code | https://github.com/unionai/unionai-examples |

## Image, resources, and caching move to the TaskEnvironment

Image, resources, and caching move from the `@task` decorator to the `TaskEnvironment`. Per-task settings like `retries` and `timeout` stay on `@env.task`. Note that `mem` is renamed to `memory`, and there are no separate `requests`/`limits` — a single `Resources` value serves as both.

### Flyte 1

```python
from datetime import timedelta

import flytekit
from flytekit import Resources

image = flytekit.ImageSpec(
    name="training-image",
    packages=["scikit-learn", "pandas"],
)

@flytekit.task(
    container_image=image,
    requests=Resources(cpu="2", mem="4Gi"),
    limits=Resources(cpu="4", mem="8Gi"),
    cache=True,
    cache_version="1.0",
    retries=3,
    timeout=timedelta(minutes=30),
)
def train_epoch(step: int) -> float:
    # A stand-in for a training step that returns the current loss.
    return 1.0 / (step + 1)

@flytekit.workflow
def main(step: int) -> float:
    return train_epoch(step=step)
```

### Flyte 2

```python
from datetime import timedelta

import flyte

# Image, resources, and caching move to the TaskEnvironment, so they are declared
# once and shared by every task in the environment.
env = flyte.TaskEnvironment(
    name="training",
    image=flyte.Image.from_debian_base().with_pip_packages("scikit-learn", "pandas"),
    resources=flyte.Resources(cpu="2", memory="4Gi"),  # "memory", not "mem"
    cache="auto",
)

# retries and timeout stay on the task decorator.
@env.task(retries=3, timeout=timedelta(minutes=30))
def train_epoch(step: int) -> float:
    # A stand-in for a training step that returns the current loss.
    return 1.0 / (step + 1)

@env.task
def main(step: int) -> float:
    return train_epoch(step)
```

## Container images: ImageSpec to flyte.Image

Flyte 1's `ImageSpec` is replaced by Flyte 2's `flyte.Image` with a fluent builder API. Instead of one constructor with many arguments, you start from a base and chain builder methods.

```python
from flyte import Image

image = (
    Image.from_debian_base(name="my-image", registry="ghcr.io/myorg", python_version=(3, 11))
    .with_pip_packages("pandas", "numpy")
    .with_apt_packages("curl", "git")
    .with_env_vars({"MY_VAR": "value"})
)
```

| Constructor | Use case |
|---|---|
| `Image.from_debian_base()` | Most common; includes the Flyte SDK |
| `Image.from_base(image_uri)` | Start from any existing image |
| `Image.from_dockerfile(path)` | Com

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-migrate-control-flow
description: "Migrates Flyte 1 branching, dynamic workflows, failure handling, and fan-out to native Flyte 2 Python. Use when migrating Flyte 1 branching, dynamic workflows, failure handling, or map_task/fan-out to Flyte 2. Trigger words: conditional, @dynamic, map_task, on_failure, branching, parallelism, fan-out, flyte.map, asyncio.gather."
---

# Flyte 1 to 2 Migration: Control Flow and Parallelism

Flyte 1 expressed branching, dynamic fan-out, and failure handling through DSL constructs (`conditional()`, `@dynamic`, `@workflow(on_failure=...)`) and `map_task`. In Flyte 2 these are all ordinary Python, because orchestration runs as real Python at runtime. Native `if`/`elif`/`else` replaces the conditional DSL, plain task loops replace `@dynamic`, `try`/`except` replaces `on_failure`, and `flyte.map` / `asyncio.gather` replace `map_task`.

## Grounding References

| Resource | URL |
|---|---|
| Migration guide (Control flow) | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/control-flow/ |
| Migration guide (Parallelism) | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/parallelism/ |
| 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/ |
| Example code | https://github.com/unionai/unionai-examples |
| Flyte MCP tools | Available via `flyte-mcp` server |

## Conditional Execution

The `conditional()` DSL becomes ordinary Python `if` / `elif` / `else` — for example, choosing a model based on dataset size.

### Flyte 1

```python
from flytekit import task, workflow, conditional

@task
def train_gradient_boosting(n_rows: int) -> str:
    return f"trained gradient boosting on {n_rows} rows"

@task
def train_logistic_regression(n_rows: int) -> str:
    return f"trained logistic regression on {n_rows} rows"

@workflow
def main(n_rows: int) -> str:
    # Pick the model based on dataset size.
    return (
        conditional("model_choice")
        .if_(n_rows > 10_000)
        .then(train_gradient_boosting(n_rows=n_rows))
        .else_()
        .then(train_logistic_regression(n_rows=n_rows))
    )
```

### Flyte 2

```python
import flyte

env = flyte.TaskEnvironment(name="conditional")

@env.task
def train_gradient_boosting(n_rows: int) -> str:
    return f"trained gradient boosting on {n_rows} rows"

@env.task
def train_logistic_regression(n_rows: int) -> str:
    return f"trained logistic regression on {n_rows} rows"

# Branching is now ordinary Python control flow -- no conditional() DSL.
@env.task
def main(n_rows: int) -> str:
    if n_rows > 10_000:
        return train_gradient_boosting(n_rows)
    return train_logistic_regression(n_rows)
```

## Dynamic Workflows

`@dynamic` existed so a task could generate a variable number of subtask calls at runtime (e.g. one per data partition discovered at runtime). In Flyte 2 every task can do this natively, so `@dynamic` simply disappears — loop over runtime data in an ordinary `@env.task`.

### Flyte 1

```python
from flytekit import task, workflow, dynamic

@task
def list_partitions(n: int) -> list[int]:
    return list(range(n))

@task
def process_partition(partition_id: int) -> int:
    # Aggregate one data partition.
    return partition_id * 2

@dynamic
def process_all(partitions: list[int]) -> list[int]:
    results = []
    for partition_id in partitions:
        results.append(process_partition(partition_id=partition_id))
    return results

@workflow
def main(n: int) -> list[int]:
    partitions = list_partitions(n=n)
    return process_all(partitions=partitions)
```

### Flyte 2

```python
import flyte

env = flyte.TaskEnvironment(name="dynamic")

@env.task
def process_partition(partition_id: int) -> int:
    # Aggregate one data partition.
    return partition_id * 2

# No @dynamic decorator needed: a plain task can loop over runtime data (e.g. a
# varia

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-migrate-data-io
description: Migrates Flyte 1 data types and offloaded I/O to Flyte 2. Use when migrating Flyte 1 data types and I/O (files, directories, dataframes, dataclasses) to Flyte 2, converting FlyteFile, FlyteDirectory, or StructuredDataset to flyte.io.File, flyte.io.Dir, and flyte.io.DataFrame. Trigger words are FlyteFile, FlyteDirectory, StructuredDataset, DataFrame, dataclass, Pydantic, type, I/O, and serialization.
---

# Flyte 1 to 2 Migration: Data Types and I/O

Flyte 2 renames the offloaded-data types and makes their I/O `async`, but the mental model is the same: pass lightweight references to large data between tasks, not the materialized bytes. `FlyteFile`, `FlyteDirectory`, and `StructuredDataset` become `flyte.io.File`, `flyte.io.Dir`, and `flyte.io.DataFrame`. Plain dataclasses and Pydantic `BaseModel`s work directly as task I/O with no JSON mixin.

## Grounding References

| Resource | URL |
|---|---|
| Migration guide | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/data-io/ |
| 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/ |
| Example code | https://github.com/unionai/unionai-examples |
| Flyte MCP tools | Available via `flyte-mcp` server |

## Type Mapping

| Flyte 1 | Flyte 2 | Notes |
|---|---|---|
| `flytekit.types.file.FlyteFile` | `flyte.io.File` | I/O is `async` |
| `flytekit.types.directory.FlyteDirectory` | `flyte.io.Dir` | I/O is `async` |
| `flytekit.types.structured.StructuredDataset` | `flyte.io.DataFrame` | build with `from_df`, read with `open(...).all()` |
| `@dataclass_json` + `@dataclass` | plain `@dataclass` | no mixin needed |
| Pydantic `BaseModel` (+ config) | plain Pydantic `BaseModel` | works directly as task I/O |

## Offloaded Data: The Mental Model

`File`, `Dir`, and `DataFrame` are lightweight references (pointers) to data offloaded in blob storage — not the materialized bytes. In Flyte 2 the read/write operations are `async`: upload with `await File.from_local(local_path)`, read with `async with f.open("rb") as fh: await fh.read()`, build a frame with `flyte.io.DataFrame.from_df(df)` (sync constructor), and read it with `await fdf.open(pandas.DataFrame).all()`.

## Files and Directories

`FlyteFile` and `FlyteDirectory` become `flyte.io.File` and `flyte.io.Dir` — the way you pass model artifacts and datasets between tasks. Use `await File.from_local(...)` to upload and `async with file.open(...)` to read.

### Flyte 1

```python
import os

from flytekit import task, workflow, current_context
from flytekit.types.file import FlyteFile

@task
def write_file(content: str) -> FlyteFile:
    path = os.path.join(current_context().working_directory, "out.txt")
    with open(path, "w") as f:
        f.write(content)
    return FlyteFile(path=path)

@task
def read_file(f: FlyteFile) -> str:
    with open(f.download()) as fh:
        return fh.read()

@workflow
def main(content: str) -> str:
    f = write_file(content=content)
    return read_file(f=f)
```

### Flyte 2

```python
import flyte
from flyte.io import File

env = flyte.TaskEnvironment(name="files")

@env.task
async def write_file(content: str) -> File:
    with open("out.txt", "w") as f:
        f.write(content)
    # File.from_local uploads the file to blob storage and returns a reference
    # (a lightweight pointer, not the materialized bytes).
    return await File.from_local("out.txt")

@env.task
async def read_file(f: File) -> str:
    async with f.open("rb") as fh:
        return (await fh.read()).decode("utf-8")

@env.task
async def main(content: str) -> str:
    f = await write_file(content)
    return await read_file(f)
```

Directories follow the same pattern: import `Dir` from `flyte.io` and use its `async` upload/read methods in place of `FlyteDirectory`. See [Files and directories](https://www.union.ai/docs/v2/flyte/user-

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-migrate-ml
description: Migrates Flyte 1 machine learning code to Flyte 2 and unlocks net-new v2 patterns. Use when migrating Flyte 1 ML workloads (training, HPO, GPU/deep learning, batch inference) to Flyte 2, specifying GPU resources, or building the end-to-end pipeline pattern. Trigger words - migrate training, HPO, GPU, deep learning, batch inference, model serving, pytorch.
---

# Flyte 1 to Flyte 2 ML Migration Skill

Migrate existing Flyte 1 ML workloads — small-model training, hyperparameter optimization, deep learning on GPUs, and batch inference — to Flyte 2, then take advantage of patterns that were not possible in Flyte 1 (real-time serving, apps, sandboxed execution).

This skill is specifically about **migrating existing v1 ML code**. For greenfield authoring in Flyte 2, use the companion skills:

- `flyte-sdk-ml` — writing new ML training / inference tasks in Flyte 2.
- `flyte-sdk-app` — writing new apps and serving endpoints.
- `flyte-sdk-agent` — writing new agents and sandboxed / code-mode workloads.

## Grounding References

| Resource | URL |
|---|---|
| Migration guide (ML workloads) | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/ml-workloads/ |
| Migration guide (New in Flyte 2) | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/new-in-flyte-2/ |
| 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/ |
| Example code | https://github.com/unionai/unionai-examples |
| Flyte MCP tools | Available via `flyte-mcp` server |

## Migration Cheat Sheet

| Flyte 1 | Flyte 2 |
|---|---|
| `ImageSpec(name=..., packages=[...])` | `flyte.Image.from_debian_base().with_pip_packages(...)` |
| `@task(container_image=..., requests=..., cache=...)` | Set `image`, `resources`, `cache` once on `flyte.TaskEnvironment`, then `@env.task` |
| `Resources(cpu=..., mem=...)` | `flyte.Resources(cpu=..., memory=...)` (note `mem` becomes `memory`) |
| `Resources(gpu="1")` + `accelerator=T4` | `flyte.Resources(gpu="T4:1")` |
| `FlyteFile` / `FlyteFile(path=...)` | `flyte.io.File` / `await File.from_local(...)` |
| `model_file.download()` | `await model_file.download()` |
| `current_context().working_directory` | `os.getcwd()` |
| `@workflow` | An orchestrating `@env.task` (plain `async` Python) |
| `map_task(fn)(x=xs)` | `await asyncio.gather(*[fn(x) for x in xs])` |
| A "pick the best" task | Plain Python after `gather` |

## Small model training (scikit-learn / XGBoost)

Train a model, persist it as a `File`, and evaluate it. Image, resources, and caching move to the `TaskEnvironment`; `FlyteFile` becomes `flyte.io.File`.

### Flyte 1

```python
import os

import joblib
from flytekit import task, workflow, ImageSpec, Resources, current_context
from flytekit.types.file import FlyteFile
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
from xgboost import XGBClassifier

image = ImageSpec(
    name="xgb-image",
    packages=["xgboost", "scikit-learn", "joblib"],
)

@task(container_image=image, requests=Resources(cpu="2", mem="4Gi"))
def train_model(n_estimators: int, max_depth: int) -> FlyteFile:
    data = load_breast_cancer()
    X_train, _, y_train, _ = train_test_split(data.data, data.target, random_state=42)
    model = XGBClassifier(n_estimators=n_estimators, max_depth=max_depth)
    model.fit(X_train, y_train)

    model_path = os.path.join(current_context().working_directory, "model.json")
    joblib.dump(model, model_path)
    return FlyteFile(path=model_path)

@task(container_image=image)
def evaluate(model_file: FlyteFile) -> float:
    model = joblib.load(model_file.download())
    data = load_breast_cancer()
    _, X_test, _, y_test = train_test_split(data.data, data.target, random_state=42)
    return float(model.score(X_test, y_test))

@workflow
def main(n_estimators: int, max_

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-migrate-slurm
description: Migrates Slurm (sbatch/srun) HPC workloads to Flyte 2 (the flyte Python SDK) — job scripts become typed tasks, `#SBATCH` pragmas become TaskEnvironment config, job arrays become flyte.map, and multi-node training becomes a clustered task environment. Use when porting an HPC or supercomputer cluster workload off Slurm, translating `#SBATCH` pragmas, or replacing sbatch chains, job arrays, and module load with Flyte. Trigger words are sbatch, srun, SLURM, `#SBATCH`, HPC, job array, partition, squeue, sinfo, module load, scancel, supercomputer, cluster migration.
---

# Slurm to Flyte 2 Migration Skill

Slurm schedules jobs. Somewhere along the way ML work stopped being jobs and became pipelines — a data prep step, a training step, an eval step, a sweep over configs, each with different hardware and failure characteristics. A Slurm job is a bash script with `#SBATCH` pragmas at the top; in Flyte 2 that script becomes a typed Python function decorated with `@env.task`, a pipeline is just a task that calls other tasks, and control flow is plain Python. This skill maps the Slurm surface area onto the Flyte 2 SDK and is honest about the places where Slurm still wins.

## Grounding References

| Resource          | URL                                                              |
| ----------------- | ---------------------------------------------------------------- |
| Migration guide   | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/ |
| 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/      |
| Example code      | https://github.com/unionai/unionai-examples                      |
| Flyte MCP tools   | Available via `flyte-mcp` server                                 |

## Three mental shifts

Almost every point of friction for a Slurm migrant traces back to one of these three.

### 1. Modules become images

On Slurm you `module load cuda/12.1`, activate a venv on NFS, and hope the login node and the compute node agree. In Flyte 2 the environment is declared once, in Python, as a `flyte.Image`. There is no Dockerfile to write: images are built remotely and content-hashed, so an unchanged spec is a cache hit and a changed one rebuilds automatically.

```python
image = (
    flyte.Image.from_debian_base(python_version=(3, 12))
    .with_pip_packages("torch", "transformers", "datasets")
    .with_env_vars({"HF_HUB_ENABLE_HF_TRANSFER": "1"})
)
```

### 2. The shared filesystem becomes explicit data

There is no shared `/home` or `/scratch` that every node sees. Instead you pass `flyte.io.File` and `flyte.io.Dir` — typed references to object storage that stream rather than copy. This is more typing than `/scratch/$USER/run17/ckpt.pt`, and in exchange you get lineage for free: every input and output of every task is recorded, so "which dataset produced this checkpoint" is a question the system answers instead of a question you grep for. If you genuinely need a parallel filesystem (Lustre, GPFS, FSx), mount it through a CSI driver and a `flyte.PodTemplate` — that path stays open.

### 3. Job scripts become functions

`sbatch` chains held together by `--dependency=afterok`, sentinel files on NFS, and a cron job that polls for them all collapse into ordinary Python: call a function, await it, branch on the result. Binaries you can't or won't rewrite in Python still run — as container tasks with typed inputs and outputs.

## Migration Cheat Sheet

| Slurm                                             | Flyte 2                                                     |
| ------------------------------------------------- | ----------------------------------------------------------- |
| `sbatch train.sh`                                 | `flyte run train.py main`                           

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-migrate-tasks-workflows
description: >-
  Migrates Flyte 1 tasks and workflows to Flyte 2, where the @task, @workflow,
  and @dynamic decorators collapse into a single @env.task on a
  flyte.TaskEnvironment and a workflow becomes a task that calls other tasks.
  Use when the user is migrating Flyte 1 tasks/workflows to Flyte 2. Trigger
  words: migrate task, migrate workflow, @task, @workflow, @dynamic,
  TaskEnvironment, env.task.
---

# Flyte 1 to 2 Migration: Tasks and Workflows

The biggest structural change in Flyte 2 is that everything is a task. The Flyte 1 `@task`, `@workflow`, and `@dynamic` decorators all collapse into a single `@env.task` on a `flyte.TaskEnvironment`, and a "workflow" is now just a task that calls other tasks. This skill covers the structural shift, sequential ordering without the `>>` operator, nested subworkflows, TaskEnvironment configuration basics, and the full `@task` parameter mapping.

## Grounding References

| Resource | URL |
|---|---|
| Migration guide | https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/tasks-and-workflows/ |
| 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/ |
| Example code | https://github.com/unionai/unionai-examples |
| Flyte MCP tools | Available via `flyte-mcp` server |

## The Structural Shift

In Flyte 1 you decorated units of work with `@task`, composed them with `@workflow`, and used `@dynamic` for runtime-generated graphs. In Flyte 2 you create one `flyte.TaskEnvironment` that carries the configuration, then decorate every function — leaf tasks and orchestrating "workflows" alike — with `@env.task`. There is no separate `@workflow` decorator: the entrypoint is just a task that calls other tasks.

## Hello World: Tasks and Workflows

A `@task` plus `@workflow` becomes two `@env.task`s, where the entrypoint task calls the others. Sequential calls are naturally ordered — no `>>` operator required.

### Flyte 1

```python
import flytekit

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

@flytekit.task
def to_upper(greeting: str) -> str:
    return greeting.upper()

@flytekit.workflow
def main(name: str) -> str:
    greeting = say_hello(name=name)
    return to_upper(greeting=greeting)
```

### Flyte 2

```python
import flyte

env = flyte.TaskEnvironment(name="hello_world")

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

@env.task
def to_upper(greeting: str) -> str:
    return greeting.upper()

# The "workflow" is now just a task that calls other tasks.
@env.task
def main(name: str) -> str:
    greeting = say_hello(name)
    return to_upper(greeting)
```

Note that Flyte 2 task calls pass arguments positionally (`say_hello(name)`) rather than requiring keyword arguments as in Flyte 1 (`say_hello(name=name)`).

## Chaining and Ordering

In Flyte 1 you sometimes used `>>` to force ordering between tasks with no data dependency. In Flyte 2, sequential (synchronous) calls run in the order they are written, and `await`ing async tasks in sequence does the same. The `>>` operator is gone.

### Flyte 1

```python
from flytekit import task, workflow

@task
def clear_staging_table() -> None:
    # Side effect only: truncate the staging table.
    print("cleared staging table")

@task
def load_into_staging() -> None:
    # Side effect only: load fresh rows into staging.
    print("loaded staging table")

@task
def publish_to_prod() -> None:
    # Side effect only: swap staging into the production table.
    print("published to prod")

@workflow
def main() -> None:
    clear = clear_staging_table()
    load = load_into_staging()
    publish = publish_to_prod()

    # These tasks pass no data between them, so use the >> operator to force
    # ordering: clear must finish before load, which must finish before publish.
    clear >> load >> publ

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-sdk-agent
description: 'Builds durable agents with Flyte 2 — ReAct patterns, Plan-and-Execute, LangGraph integration, PydanticAI integration, OpenAI Agents SDK integration, agent memory, MCP tool integration, skills, and agent chat UI. Use when the user wants to build AI agents, implement ReAct loops, integrate agent frameworks, add agent memory, or build agent-powered workflows. Trigger words: "agent", "ReAct", "LangGraph", "PydanticAI", "OpenAI agents", "MCP", "tool calling", "memory", "skills", "agentic".'
---

# Flyte 2 SDK Agent Skill

Build durable, observable AI agents with Flyte 2.

## 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.)

## Pure Python Agents (No Framework)

### ReAct Pattern — Reason, Act, Observe

```python
import flyte

env = flyte.TaskEnvironment(
    name="react-agent",
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "openai", "boto3",
    ),
)

@env.task
async def think(observation: str) -> str:
    """LLM generates next action."""
    from openai import OpenAI
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": f"Observation: {observation}\nWhat do you do next?"},
        ],
    )
    return response.choices[0].message.content

@env.task
async def act(thought: str) -> dict:
    """Parse thought and execute tool call."""
    # Parse the thought to extract tool name and arguments
    # Then call the appropriate tool
    return {"tool": "search", "result": "search results..."}

@env.task
async def observe(result: dict) -> str:
    """Format tool result for the next reasoning step."""
    return f"Tool {result['tool']} returned: {result['result']}"

@env.task
async def react_loop(initial_query: str, max_steps: int = 5) -> str:
    """ReAct loop: think → act → observe → think → ..."""
    observation = initial_query
    for i in range(max_steps):
        thought = await think(observation)
        if "FINAL_ANSWER" in thought:
            return thought.split("FINAL_ANSWER:")[-1].strip()
        result = await act(thought)
        observation = await observe(result)
    return "Max steps reached"

if __name__ == "__main__":
    import asyncio
    result = asyncio.run(react_loop("What is the weather in Tokyo?"))
    print(result)
```

### Plan-and-Execute with Parallel Fan-out

```python
@env.task
async def plan(query: str) -> list[str]:
    """Generate a plan of sub-tasks."""
    from openai import OpenAI
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "Break this query into sub-tasks."},
            {"role": "user", "content": query},
        ],
    )
    # Parse response into list of sub-tasks
    return response.choices[0].message.content.split("\n")

@env.task
async def execute_subtask(task: str) -> str:
    """Execute a singl

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-sdk-app
description: 'Builds and serves Flyte 2 apps — FastAPI, Streamlit, vLLM, SGLang, WebSocket, and browser apps. Use when the user wants to serve a model, create a REST API, build a dashboard, deploy an LLM backend, or create a web app with Flyte. Trigger words: "app", "serve", "deploy app", "FastAPI", "Streamlit", "vLLM", "SGLang", "REST API", "dashboard", "serving", "endpoint", "webhook", "WebSocket".'
---

# Flyte 2 SDK App Skill

Build and serve applications with Flyte 2.

## 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 |

## App Types

| App Type | Use Case | Import |
|---|---|---|
| `FastAPIAppEnvironment` | REST APIs, model serving | `from flyte.app.extras import FastAPIAppEnvironment` |
| `StreamlitAppEnvironment` | Dashboards, data apps | `from flyte.app.extras import StreamlitAppEnvironment` |
| `vLLMAppEnvironment` | LLM serving | `from flyte.app.extras import vLLMAppEnvironment` |
| `SGLangAppEnvironment` | Structured generation | `from flyte.app.extras import SGLangAppEnvironment` |
| Custom (`AppEnvironment`) | Any HTTP server | `import flyte` |

## FastAPI App — Model Serving

### Basic FastAPI app

```python
from fastapi import FastAPI
import flyte
from flyte.app.extras import FastAPIAppEnvironment

app = FastAPI()
env = FastAPIAppEnvironment(
    name="my-model",
    app=app,
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "fastapi", "uvicorn", "torch",
    ),
)

@app.get("/predict")
async def predict(x: float) -> dict:
    return {"result": x * 2 + 5}

if __name__ == "__main__":
    flyte.init_from_config()
    flyte.serve(env)
```

### Model serving with loading

```python
from fastapi import FastAPI
import flyte
from flyte.app.extras import FastAPIAppEnvironment

app = FastAPI()
env = FastAPIAppEnvironment(
    name="text-classifier",
    app=app,
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "fastapi", "uvicorn", "torch", "transformers",
    ),
)

model = None  # Loaded once at startup

@app.on_event("startup")
async def load_model():
    global model
    model = transformers.AutoModelForSequenceClassification.from_pretrained("bert-base")

@app.get("/predict")
async def predict(text: str) -> dict:
    assert model is not None
    outputs = model(transformers.encode(text))
    return {"prediction": outputs.argmax().item(), "confidence": outputs.softmax().max().item()}

if __name__ == "__main__":
    flyte.init_from_config()
    flyte.serve(env)
```

### Multi-file FastAPI app

```
app/
  __init__.py
  main.py        # FastAPI app entry
  routes/
    __init__.py
    predict.py
    health.py
  models/
    __init__.py
    classifier.py
```

```python
# app/main.py
from fastapi import FastAPI
from .routes import predict, health

app = FastAPI()
app.include_router(predict.router, prefix="/api")
app.include_router(health.router, prefix="/health")
```

## Streamlit App — Data Dashboards

### Basic Streamlit app

```python
import streamlit as st
import flyte
from flyte.app.extras import StreamlitAppEnvironment

st.title("Data Dashboard")

df = st.dataframe(load_data())

if st.button("Refresh"):
    st.rerun()

env = StreamlitAppEnvironment(
    name="dashboard",
    script="app.py",
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "streamlit", "pandas", "matplotlib",
    ),
)

if __name__ == "__main__":
    flyte.init_from_config()
    flyte.serve(env)
```

### Streamlit with

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-sdk-data
description: 'Handles data engineering patterns: ETL pipelines, data processing, data quality checks, fanout/map tasks, conditions, dynamic workflows, and batch data transformations. Use when the user wants to build ETL pipelines, process large datasets, run data quality checks, fan out data processing tasks, or handle batch data transformations. Trigger words: "ETL", "data pipeline", "data processing", "fanout", "map", "transform", "data quality", "parquet", "CSV", "batch", "extract", "load", "validate", "schema".'
---

# Flyte 2 SDK Data Engineering Skill

Build ETL pipelines, data processing workflows, and data quality checks with Flyte 2.

## 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.)

## ETL Pipeline Patterns

### Basic Extract-Transform-Load

```python
import flyte
import flyte.io

env = flyte.TaskEnvironment(
    name="etl-pipeline",
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "pandas", "polars", "pyarrow", "boto3", "sqlalchemy",
    ),
)

@env.task(retries=3, cache="auto")
async def extract(source_uri: str) -> flyte.io.DataFrame:
    """Extract data from various sources."""
    import polars as pl
    if source_uri.endswith(".csv"):
        df = pl.read_csv(source_uri)
    elif source_uri.endswith(".parquet"):
        df = pl.read_parquet(source_uri)
    else:
        raise ValueError(f"Unsupported format: {source_uri}")
    return flyte.io.DataFrame(df)

@env.task(retries=2, cache="auto")
async def transform(df: flyte.io.DataFrame) -> flyte.io.DataFrame:
    """Clean and transform data."""
    inner = df.to_polars()
    cleaned = (
        inner
        .drop_nulls()
        .unique()
        .with_columns([
            pl.col("date").str.strptime(pl.Date, "%Y-%m-%d").alias("date_parsed"),
        ])
    )
    return flyte.io.DataFrame(cleaned)

@env.task(retries=1, cache="auto")
async def load(df: flyte.io.DataFrame, destination: str) -> str:
    """Load transformed data to destination."""
    inner = df.to_polars()
    if destination.endswith(".parquet"):
        inner.write_parquet(destination)
    elif destination.endswith(".csv"):
        inner.write_csv(destination)
    return destination

@env.task
async def etl_pipeline(source_uri: str, destination: str) -> dict:
    """Orchestrate the ETL pipeline."""
    raw = await extract(source_uri)
    cleaned = await transform(raw)
    loaded_path = await load(cleaned, destination)
    return {"source": source_uri, "destination": loaded_path}
```

### Multi-step ETL with intermediate storage

```python
@env.task(cache="auto")
async def extract_raw(source_uri: str) -> flyte.io.File:
    """Extract and save raw data to remote storage."""
    import polars as pl
    df = pl.read_csv(source_uri)
    path = "/tmp/raw.parquet"
    df.write_parquet(path)
    return flyte.io.File(path=path)

@env.task(cache="auto")
async def validate_raw(raw: flyte.io.File) -> dict:
    """Validate raw data quality."""
    df = pl.read_parquet(raw.path)
    return {
        "

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-sdk-eval
description: 'Builds minimal evaluation harnesses (unit tests + small-run workflows) and suggests ways to validate correctness and performance early. Use when the user wants to test Flyte tasks, validate pipeline outputs, set up evaluation pipelines, or write unit tests for ML/data workflows. Trigger words: "test", "evaluate", "validation", "unit test", "verify", "assert", "data quality", "metrics", "benchmark".'
---

# Flyte 2 SDK Eval Skill

Build evaluation harnesses, unit tests, and validation pipelines for Flyte 2 workflows.

## 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 |

## Testing Patterns

### Direct Task Invocation (unit testing)

Test task logic directly without remote execution:

```python
import pytest
from pipeline import preprocess, train, evaluate

def test_preprocess():
    """Test preprocessing logic in isolation."""
    result = preprocess(["a", "b", "c"])
    assert result is not None
    assert len(result) == 3

def test_train():
    """Test training with a small dataset."""
    import flyte
    import flyte.io
    data = flyte.io.DataFrame(polars.DataFrame({"x": [1, 2, 3], "y": [4, 5, 6]}))
    model = train(data)
    assert model is not None

def test_evaluate():
    """Test evaluation metrics."""
    import flyte
    model = flyte.io.File(path="/tmp/mock_model.pt")
    metrics = evaluate(model)
    assert "accuracy" in metrics
    assert 0 <= metrics["accuracy"] <= 1
```

### Using flyte.run() for Integration Testing

Test the full workflow execution locally:

```python
import pytest
import flyte
from pipeline import main

def test_full_pipeline():
    """Run the full pipeline locally with test data."""
    result = flyte.run(main, inputs={"data": ["test1", "test2"]})
    assert result is not None
    assert "accuracy" in result.outputs

def test_full_pipeline_with_inputs():
    """Test with specific inputs via flyte.run()."""
    result = flyte.run(
        main,
        inputs={"data": ["a", "b", "c"]},
    )
    assert result.status == "SUCCEEDED"
```

### Testing Async Tasks

```python
import asyncio
import pytest

def test_async_task():
    """Test async task by running in event loop."""
    result = asyncio.run(preprocess(["a", "b"]))
    assert result is not None

@pytest.mark.asyncio
async def test_async_task_mark():
    """Test async task with pytest-asyncio."""
    result = await preprocess(["a", "b"])
    assert len(result) == 2
```

## Evaluation Pipeline Patterns

### ML Model Evaluation

```python
import flyte
import flyte.io

env = flyte.TaskEnvironment(
    name="eval",
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "scikit-learn", "scipy", "pandas", "matplotlib",
    ),
)

@env.task
async def load_test_data() -> flyte.io.DataFrame:
    """Load ground truth test data."""
    ...
    return flyte.io.DataFrame(df)

@env.task
async def load_model(model_uri: str) -> object:
    """Load a trained model."""
    ...

@env.task
async def predict(model: object, data: flyte.io.DataFrame) -> flyte.io.DataFrame:
    """Run model predictions on test data."""
    ...

@env.task
async def compute_metrics(
    predictions: flyte.io.DataFrame,
    ground_truth: flyte.io.DataFrame,
) -> dict:
    """Compute evaluation metrics."""
    from sklearn.metrics import (
        accuracy_score, f1_score, precision_score, recall_score,
        roc_auc_score, mean_squared_error,
    )
    y_true = ground_truth.to_polars()["label"].to_list()
    y_pred

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-sdk-ml
description: 'Handles ML workload patterns: model training, hyperparameter optimization, experiment tracking, model evaluation and selection, batch inference, real-time serving, and model monitoring. Use when the user wants to train models, run hyperparameter search, track experiments, evaluate models, do batch or real-time inference, or set up model monitoring. Trigger words: "train", "training", "hyperparameter", "HPO", "experiment", "tracking", "evaluation", "inference", "batch inference", "model serving", "monitoring", "GPU", "PyTorch", "TensorFlow", "scikit-learn", "HuggingFace", "model".'
---

# Flyte 2 SDK ML Skill

Build ML training, HPO, evaluation, and inference pipelines with Flyte 2.

## 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.)

## Model Training

### PyTorch Training

```python
import flyte
import flyte.io

env = flyte.TaskEnvironment(
    name="training",
    image=flyte.Image.from_base("pytorch/pytorch:2.1-cuda12.1-cudnn8-devel").with_pip_packages(
        "transformers", "datasets", "accelerate",
    ),
)

@env.task(
    requests=flyte.Resources(
        cpu="4", memory="16Gi", gpu="1", gpu_model="nvidia-a10g",
    ),
)
async def train(
    train_data: flyte.io.File,
    val_data: flyte.io.File,
    hyperparams: dict,
) -> flyte.io.File:
    """Train a model and save checkpoint."""
    import torch
    from transformers import AutoModelForSequenceClassification, AutoTokenizer

    # Load data
    tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
    model = AutoModelForSequenceClassification.from_pretrained(
        "bert-base-uncased", num_labels=2
    )

    # Train
    for epoch in range(hyperparams["epochs"]):
        # ... training loop ...
        pass

    # Save checkpoint
    output_path = "/tmp/model_checkpoint"
    model.save_pretrained(output_path)
    tokenizer.save_pretrained(output_path)
    return flyte.io.File(path=output_path)

@env.task
async def main(
    train_uri: str,
    val_uri: str,
    lr: float = 0.001,
    batch_size: int = 32,
    epochs: int = 3,
) -> dict:
    hyperparams = {"lr": lr, "batch_size": batch_size, "epochs": epochs}
    checkpoint = await train(
        train_data=flyte.io.File(path=train_uri),
        val_data=flyte.io.File(path=val_uri),
        hyperparams=hyperparams,
    )
    return {"checkpoint": checkpoint, "hyperparams": hyperparams}
```

### scikit-learn Training

```python
import flyte
import flyte.io

env = flyte.TaskEnvironment(
    name="sklearn-training",
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "scikit-learn", "pandas", "polars", "joblib",
    ),
)

@env.task
async def train_sklearn(
    train_data: flyte.io.DataFrame,
    val_data: flyte.io.DataFrame,
    model_type: str = "random_forest",
) -> flyte.io.File:
    """Train a scikit-learn model."""
    from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
    from sklearn.linear_model import LogisticRegression
    import joblib

    X_train = 

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-sdk-optimize
description: 'Suggests performance improvements (task granularity, caching, resource requests, data format changes) using observed run metadata when available. Use when the user wants to optimize workflow performance, debug slow tasks, configure caching, tune resources, or improve throughput. Trigger words: "optimize", "performance", "slow", "cache", "caching", "resource", "throughput", "latency", "speed up", "bottleneck", "profiling", "metadata".'
---

# Flyte 2 SDK Optimize Skill

Optimize Flyte 2 workflows for performance, cost, and reliability.

## 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 |

## Optimization Strategy Overview

Performance optimization in Flyte follows a hierarchy:

1. **Reduce container overhead** — use traces for lightweight ops
2. **Parallelize work** — use `flyte.map` for fan-out
3. **Cache results** — use `cache="auto"` for idempotent tasks
4. **Tune resources** — set appropriate CPU/memory/GPU
5. **Optimize data transfer** — choose efficient formats, reduce inline I/O
6. **Use reusable containers** — shared environments reduce image pull time

## Caching

### Enable automatic caching

```python
@env.task(cache="auto")  # versioned by function body + inputs
async def preprocess(data: list[str]) -> flyte.io.File:
    ...
```

### Cache key strategies

```python
@env.task(cache="auto")  # default: function body + inputs
async def task_a(data: str) -> flyte.io.File:
    ...

@env.task(cache="override", salt="v2")  # add salt for cache key variation
async def task_b(data: str) -> flyte.io.File:
    ...

@env.task(cache="disable")  # always re-run
async def task_c(data: str) -> flyte.io.File:
    ...
```

### Content-based caching for DataFrames

```python
@env.task(cache="auto")
async def transform(df: flyte.io.DataFrame) -> flyte.io.DataFrame:
    """Cache key includes DataFrame content hash."""
    ...
```

### Ignoring specific inputs in cache key

```python
@env.task(cache="auto", cache_ignore_inputs=["api_key"])
async def fetch_data(api_key: str, url: str) -> flyte.io.File:
    """Don't include api_key in cache key."""
    ...
```

### Cache policies

```python
@env.task(cache="auto", cache_policy=flyte.CachePolicy(min_cached_age="1h"))
async def cached_task(data: str) -> flyte.io.File:
    """Only use cache if result is at least 1 hour old."""
    ...
```

## Resource Tuning

### Setting task resources

```python
@env.task(
    requests=flyte.Resources(cpu="500m", memory="1Gi"),
    limits=flyte.Resources(cpu="2", memory="4Gi"),
)
async def light_task(data: str) -> str:
    """Lightweight task — small resources."""
    ...

@env.task(
    requests=flyte.Resources(cpu="4", memory="16Gi"),
    limits=flyte.Resources(cpu="8", memory="32Gi"),
)
async def heavy_task(data: flyte.io.DataFrame) -> flyte.io.DataFrame:
    """Heavy data processing — large resources."""
    ...

@env.task(
    requests=flyte.Resources(cpu="1", memory="4Gi", gpu="1", gpu_model="nvidia-a10g"),
    limits=flyte.Resources(cpu="2", memory="8Gi", gpu="1", gpu_model="nvidia-a10g"),
)
async def train_model(data: flyte.io.File) -> flyte.io.File:
    """GPU training task."""
    ...
```

### GPU resource configuration

```python
@env.task(
    requests=flyte.Resources(
        cpu="2",
        memory="8Gi",
        gpu="1",
        gpu_model="nvidia-a10g",  # or "nvidia-a100", "nvidia-h100"
    ),
)
async def inference(batch: flyte.io.DataFrame) -> flyte.io.DataFrame:
    ...
```

### Resource recommendations by

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-sdk-run
description: 'Runs Flyte 2 workflows, interacts with runs and actions, retrieves logs and data, and manages run lifecycle. Use when the user wants to run a workflow, check run status, view logs, get run outputs, re-run a workflow, or manage runs programmatically. Trigger words: "run", "execute", "logs", "status", "output", "input", "watch", "rerun", "cancel", "abort", "run metadata", "action".'
---

# Flyte 2 SDK Run Skill

Run workflows, interact with runs, and manage the execution lifecycle.

## 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 |

## Tool Priority

1. **Flyte MCP** — if the harness has Flyte MCP tools, prefer them over shelling out to
   the CLI. They cover listing runs, fetching run details and inputs/outputs, polling to
   completion, executing a task, and aborting a run, and they return structured data
   instead of text you have to parse.
2. **`flyte` CLI** — for local run commands, and anything MCP does not expose
3. **Python SDK** — for programmatic run control

## Running Workflows

### Via Python SDK

```python
import flyte

if __name__ == "__main__":
    # Run with defaults from config
    result = flyte.run(main, inputs={"data": ["a", "b", "c"]})
    print(f"Run name: {result.name}")
    print(f"Status: {result.status}")
```

### Via CLI

```bash
# Run with local config
flyte run pipeline.py main --data '[1,2,3]'

# Run with specific project/domain
flyte run pipeline.py main --data '[1,2,3]' --project flytesnacks --domain development

# Run with custom run name
flyte run pipeline.py main --data '[1,2,3]' --name my-custom-run

# Run with specific image
flyte run pipeline.py main --data '[1,2,3]' --image ghcr.io/myorg/task:v1.0

# Run with local mode (in-process, no remote)
flyte run --local pipeline.py main --data '[1,2,3]'

# Run with TUI
flyte run --tui --local pipeline.py main --data '[1,2,3]'

# Pass arguments by type
flyte run pipeline.py main \
  --data '[1,2,3]' \
  --learning-rate 0.001 \
  --batch-size 32 \
  --train-data s3://bucket/train.parquet \
  --flag true
```

### Run command options

| Flag | Description |
|---|---|
| `--project` / `--domain` | Target project and domain |
| `--run-project` / `--run-domain` | Override run project/domain |
| `--local` | Run locally (in-process) |
| `--tui` | Terminal UI for local runs |
| `--name` | Custom run name |
| `--image` | Image mapping (named or default) |
| `--copy-style` | `loaded_modules` (default), `all`, `none` |
| `--root-dir` | Set root directory for code bundling |
| `--raw-data-path` | Override raw data path |
| `--service-account` | K8s service account |
| `--follow` | Follow run progress |
| `--no-sync-local-sys-paths` | Skip local sys path sync |

### Passing inputs by type

```bash
# List
flyte run pipeline.py main --data '[1,2,3]'

# Dict
flyte run pipeline.py main --config '{"lr": 0.001, "epochs": 10}'

# Boolean
flyte run pipeline.py main --flag true

# Datetime
flyte run pipeline.py main --date '2025-01-01T00:00:00'

# Duration
flyte run pipeline.py main --timeout '1h'

# File
flyte run pipeline.py main --input-file s3://bucket/data.parquet

# DataFrame (via file path)
flyte run pipeline.py main --data-file /path/to/data.parquet
```

## Interacting with Runs

### Using Flyte MCP

If Flyte MCP tools are available, prefer them for all of the above — listing runs,
fetching a run's details, polling until it completes, and reading its inputs and outputs.


### Using CLI

```bash
# List runs
flyte get run --project fl

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-sdk-ship
description: 'Generates flyte.Image specs, Dockerfiles, dependency management, image tagging strategy, and reproducible build instructions for Flyte tasks. Use when the user needs to build container images for Flyte tasks, configure custom images, manage dependencies, or set up reproducible builds. Trigger words: "image", "Docker", "build", "dependency", "pip package", "debian base", "image builder", "push image", "container", "Dockerfile", "uv", "requirements".'
---

# Flyte 2 SDK Ship Skill

Generate images, Dockerfiles, and dependency configurations for Flyte 2 tasks.

## 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.)

## flyte.Image — Programmatic Image Definition

Use `flyte.Image` to define task container images in Python. Flyte builds and pushes them automatically.

### From Debian Base

```python
import flyte

env = flyte.TaskEnvironment(
    name="etl-pipeline",
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "pandas", "polars", "pyarrow", "boto3",
    ).with_system_packages(
        "git", "curl", "wget",
    ),
)
```

### From Existing Image

```python
env = flyte.TaskEnvironment(
    name="ml-training",
    image=flyte.Image.from_base(
        "ghcr.io/flyteorg/flyte:py3.12-v2",  # base image
    ).with_pip_packages(
        "torch", "transformers", "datasets",
    ),
)
```

### Image from uv Script Metadata

When using a `# /// script` header, Flyte can derive the image from the script's dependencies:

```python
# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "pandas",
#   "polars",
# ]
# ///

# Flyte reads the script metadata and builds the image automatically
```

## Image Configuration Methods

### with_pip_packages

```python
image = flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
    "pandas",           # latest version
    "torch>=2.0",       # version constraint
    "transformers==4.40",  # pinned version
)
```

### with_system_packages

```python
image = image.with_system_packages(
    "git", "curl", "wget", "jq", "ffmpeg",
)
```

### with_commands (apt-get run)

```python
image = image.with_commands(
    "apt-get update && apt-get install -y libgl1-mesa-glx",  # for OpenCV
    "pip install --upgrade pip",
)
```

### with_env_vars

```python
image = image.with_env_vars({
    "HF_HUB_DISABLE_TELEMETRY": "1",
    "PYTHONDONTWRITEBYTECODE": "1",
})
```

### with_local_rs_controller

```python
# For development: bake the Rust controller wheel into the image
image = image.with_local_rs_controller()
```

## Custom Dockerfile

For complex builds, use a custom `Dockerfile`:

```dockerfile
# Dockerfile
FROM python:3.12-slim

RUN apt-get update && apt-get install -y \
    git \
    libgl1-mesa-glx \
    && rm -rf /var/lib/apt/lists/*

RUN pip install --no-cache-dir \
    torch \
    transformers \
    datasets

WORKDIR /app
COPY . /app

ENV HF_HUB_DISABLE_TELEMETRY=1
```

```python
env = flyte.TaskEnvironment(
    name="ml-training",
    image=flyte.Image.from_dock

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: flyte-sdk-types
description: 'Guides correct types, I/O, and serialization for common data (Pandas, Arrow, Parquet, images, audio, HF datasets), including data locality and storage best practices. Use when the user needs help with type annotations, data serialization, file I/O between tasks, custom type transformers, DataFrame handling, or choosing the right Flyte type for their data. Trigger words: "type", "serialize", "deserialize", "DataFrame", "File", "Directory", "custom type", "data format", "Parquet", "Arrow", "Pandas", "type transformer".'
---

# Flyte 2 SDK Types Skill

Guide correct type annotations, I/O patterns, and serialization for Flyte 2 workflows.

## 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.)

## Type System Overview

Flyte 2 uses **Python type hints** for serialization. Every task input/output must have a type annotation. Flyte's type transformer system handles conversion between Python types and remote storage (S3, GCS, etc.).

### Supported Native Types

| Python Type | Flyte Type | Remote Transport |
|---|---|---|
| `int`, `float`, `bool`, `str` | Literal | Inline (JSON) |
| `list[T]`, `dict[K, V]` | Collection / Map | Inline (JSON) for small, blob for large |
| `flyte.io.File` | Blob | Uploaded to metadata bucket |
| `flyte.io.Dir` | Blob (directory) | Uploaded to metadata bucket |
| `flyte.io.DataFrame` | DataFrame | Parquet in metadata bucket |
| `dataclass` | Struct | JSON in metadata bucket |
| `pydantic.BaseModel` | Struct | JSON in metadata bucket |
| `datetime`, `timedelta` | DateTime / Duration | Inline |

## flyte.io.File — Single Files

Use `flyte.io.File` for any single file that flows between tasks. Files are **automatically uploaded** to the metadata bucket at runtime.

```python
import flyte
import flyte.io

@env.task
async def download_url(url: str) -> flyte.io.File:
    """Download a file and return as flyte.io.File."""
    import urllib.request
    local_path = f"/tmp/{url.split('/')[-1]}"
    urllib.request.urlretrieve(url, local_path)
    return flyte.io.File(path=local_path)

@env.task
async def process(file: flyte.io.File) -> dict:
    """Read a file — flyte.io.File downloads it automatically."""
    # file.path gives the local path (already downloaded)
    with open(file.path) as f:
        content = f.read()
    return {"lines": len(content.splitlines())}

@env.task
async def main(url: str) -> dict:
    downloaded = await download_url(url)
    return await process(downloaded)  # type: flyte.io.File flows as remote reference
```

### flyte.io.File best practices

- **Always pass `flyte.io.File` between tasks** — never pass file paths as strings. Flyte serializes the reference to the remote blob.
- **`file.path`** gives the local download path inside the task container.
- **Don't hardcode paths** — let Flyte manage the download/upload lifecycle.
- **Compression**: Flyte infers format from extension (`.parquet`, `.csv`, `.json`, `.pt`, `.png`, etc.).

## flyte.io.Dir — Directories

Use `flyte.io.Dir` for a collection of files

Preview of the first 4,000 characters. Read full instructions →

View saved version →

---
name: start-dex-local
description: Deploy Dex as a local in-cluster OIDC provider (IdP stand-in) for a kind-based Flyte deployment, so oauth2-proxy can be tested with no cloud account or real users. Use when the user wants to stand up Dex locally for testing Flyte authentication. For local testing only — in-memory storage, static test passwords.
---

# Start a local IdP with Dex (for testing)

Replace the **external** OIDC provider (Okta, Google, …) that oauth2-proxy
expects with [Dex](https://dexidp.io/) running **inside the same kind cluster**,
so you can test the whole Flyte auth flow with no cloud account and no real
users.

This assumes a kind cluster with Flyte and Traefik already up (steps 1–6 of the
kind deployment / the `deploy-flyte-kind` skill's OIDC section), about to wire
oauth2-proxy. Deploy Dex first, then point oauth2-proxy at it.

> **For local testing only.** Dex here uses in-memory storage and a static test
> password baked into its config. Never use this configuration anywhere real.

## The issuer-URL constraint (why the setup looks the way it does)

OIDC requires the **issuer URL to be identical everywhere it's seen**:
- oauth2-proxy (in-cluster) reaches Dex over a Kubernetes service name.
- The browser reaches Dex to log in, and must land on the *same* issuer the
  token was minted for, or validation fails.

A service name (`dex.flyte.svc.cluster.local`) isn't resolvable from the
browser; a `localhost` URL isn't resolvable from inside the cluster. The fix:
serve Dex under the **same host as Flyte** (`flyte.local`) at a sub-path
(`/dex`), routed through Traefik. One URL — `http://flyte.local/dex` — works
from both sides.

## Step 0: Prerequisites

```bash
for t in kubectl helm; do command -v $t >/dev/null || echo "MISSING: $t"; done
kubectl --context kind-flyte -n traefik get deploy traefik >/dev/null 2>&1 || echo "MISSING: traefik"
kubectl --context kind-flyte -n flyte get svc flyte-http >/dev/null 2>&1 || echo "MISSING: flyte"
docker ps --filter name=flyte-control-plane --format '{{.Ports}}' | grep -q '80->30080' \
  || echo "MISSING: hostPort 80->30080 mapping (recreate the cluster — see deploy-flyte-kind Step 1)"
```

If anything is `MISSING`, stop. Flyte + Traefik must already be deployed, and the
cluster must have been created with the `hostPort: 80 → 30080` mapping — Dex's
issuer (`http://flyte.local/dex`) is unreachable from the browser without it, and
the mapping can't be added to an existing cluster. Also confirm `127.0.0.1
flyte.local` is in `/etc/hosts`.

## Step 1: Write the Dex config

Write `dex-config.yaml` in the working directory. The issuer is the
through-Traefik URL; the two static clients are oauth2-proxy (confidential, with
a secret) and the Flyte CLI (public, for SDK login). `staticPasswords` gives a
login with no external user store:

```yaml
# dex-config.yaml
issuer: http://flyte.local/dex

storage:
  type: memory

web:
  http: 0.0.0.0:5556

oauth2:
  skipApprovalScreen: true        # auto-approve, no consent screen in dev

staticClients:
  # oauth2-proxy — confidential client (matches the secret passed to oauth2-proxy)
  - id: oauth2-proxy
    name: oauth2-proxy
    secret: oauth2-proxy-secret
    redirectURIs:
      - 'http://flyte.local/oauth2/callback'
      - 'https://flyte.local/oauth2/callback'   # console opened over TLS (websecure)

  # Flyte CLI — public client for SDK/CLI PKCE login
  - id: flytectl
    name: 'Flyte CLI'
    public: true
    redirectURIs:
      - 'http://localhost:53593/callback'

enablePasswordDB: true
staticPasswords:
  # login: admin@example.com / password
  - email: "admin@example.com"
    username: "admin"
    userID: "08a8684b-db88-4b73-90a9-3cd1661f5466"
    # bcrypt hash of the literal string "password" — see the warning below
    hash: "$2a$10$wi77Jcsjw08l416Q4./OCu6qNvYMaNSvA3Jbo30QeyZAvq9b4BSRK"
```

**`hash` must be a complete 60-character bcrypt string.** Dex crashes
(`CrashLoopBackOff`) with `malformed bcrypt hash: hashedSecret too shor

Preview of the first 4,000 characters. Read full instructions →