flyte-sdk-app

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

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

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)