Passing inputs into app environments

[[AppEnvironment]]s support various input types that can be passed at deployment time. This includes primitive values, files, directories, and delayed values like RunOutput and AppEndpoint.

Input types overview

There are several input types:

  • Primitive values: Strings, numbers, booleans
  • Files: flyte.io.File objects
  • Directories: flyte.io.Dir objects
  • Delayed values: RunOutput (from task runs) or AppEndpoint ( apps)

Basic input types

passing-inputs-examples.py 
# String inputs
app_env = flyte.app.AppEnvironment(
    name="configurable-app",
    inputs=[
        flyte.app.Input(name="environment", value="production"),
        flyte.app.Input(name="log_level", value="INFO"),
    ],
    # ...
)

# File inputs
app_env2 = flyte.app.AppEnvironment(
    name="app-with-model",
    inputs=[
        flyte.app.Input(
            name="model_file",
            value=flyte.io.File("s3://bucket/models/model.pkl"),
            mount="/app/models",
        ),
    ],
    # ...
)

# Directory inputs
app_env3 = flyte.app.AppEnvironment(
    name="app-with-data",
    inputs=[
        flyte.app.Input(
            name="data_dir",
            value=flyte.io.Dir("s3://bucket/data/"),
            mount="/app/data",
        ),
    ],
    # ...
)

Delayed values

Delayed values are inputs whose actual values are materialized at deployment time.

RunOutput

Use RunOutput to pass outputs from task runs as app inputs:

passing-inputs-examples.py 
# Delayed inputs with RunOutput
env = flyte.TaskEnvironment(name="training-env")

@env.task
async def train_model() -> flyte.io.File:
    # ... training logic ...
    return await flyte.io.File.from_local("/tmp/trained-model.pkl")

# Use the task output as an app input
app_env4 = flyte.app.AppEnvironment(
    name="serving-app",
    inputs=[
        flyte.app.Input(
            name="model",
            value=flyte.app.RunOutput(type="file", run_name="training_run", task_name="train_model"),
            mount="/app/model",
        ),
    ],
    # ...
)

The type argument is required and must be one of string, file, or directory. When the app is deployed, it will make the remote calls needed to figure out the actual value of the input.

AppEndpoint

Use AppEndpoint to pass endpoints from other apps:

passing-inputs-examples.py 
# Delayed inputs with AppEndpoint
app1_env = flyte.app.AppEnvironment(name="backend-api")

app2_env = flyte.app.AppEnvironment(
    name="frontend-app",
    inputs=[
        flyte.app.Input(
            name="backend_url",
            value=flyte.app.AppEndpoint(app_name="backend-api"),
            env_var="BACKEND_URL",  # app1_env's endpoint will be available as an environment variable
        ),
    ],
    # ...
)

The endpoint URL will be injected as the input value when the app starts.

This is particularly useful when you want to chain apps together (for example, a frontend app calling a backend app), without hardcoding URLs.

Overriding inputs at serve time

You can override input values when serving apps (this is not supported for deployment):

# Override inputs when serving
app = flyte.with_servecontext(
    input_values={"my-app": {"model_path": "s3://bucket/new-model.pkl"}}
).serve(app_env)

Input overrides are only available when using flyte.serve() or flyte.with_servecontext().serve(). The flyte.deploy() function does not support input overrides - inputs must be specified in the AppEnvironment definition.

This is useful for:

  • Testing different configurations during development
  • Using different models or data sources for testing
  • A/B testing different app configurations

Example: FastAPI app with configurable model

Here’s a complete example showing how to use inputs in a FastAPI app:

app-inputs-fastapi-example.py 
"""Example: FastAPI app with configurable model input."""

from contextlib import asynccontextmanager
from flyte.app.extras import FastAPIAppEnvironment
from fastapi import FastAPI
import os
import flyte
import joblib


state = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Access input via environment variable
    model = joblib.load(os.getenv("MODEL_PATH", "/app/models/default.pkl"))
    state["model"] = model
    yield


app = FastAPI(lifespan=lifespan)

app_env = FastAPIAppEnvironment(
    name="model-serving-api",
    app=app,
    inputs=[
        flyte.app.Input(
            name="model_file",
            value=flyte.io.File("s3://bucket/models/default.pkl"),
            mount="/app/models",
            env_var="MODEL_PATH",
        ),
    ],
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "fastapi", "uvicorn", "scikit-learn"
    ),
    resources=flyte.Resources(cpu=2, memory="2Gi"),
    requires_auth=False,
)

@app.get("/predict")
async def predict(data: dict):
    model = state["model"]
    return {"prediction": model.predict(data)}

Example: Using RunOutput for model serving

passing-inputs-examples.py 
# Example: Using RunOutput for model serving
import joblib
from sklearn.ensemble import RandomForestClassifier
from flyte.app.extras import FastAPIAppEnvironment
from fastapi import FastAPI

# Training task
training_env = flyte.TaskEnvironment(name="training-env")

@training_env.task
async def train_model_task() -> flyte.io.File:
    """Train a model and return it."""

    model = RandomForestClassifier()

    # ... training logic ...

    path = "./trained-model.pkl"
    joblib.dump(model, path)
    return await flyte.io.File.from_local(path)

# Serving app that uses the trained model
app = FastAPI()
serving_env = FastAPIAppEnvironment(
    name="model-serving-app",
    app=app,
    inputs=[
        flyte.app.Input(
            name="model",
            value=flyte.app.RunOutput(
                type="file",
                task_name="training-env.train_model_task"
            ),
            mount="/app/model",
            env_var="MODEL_PATH",
        ),
    ],
)

Accessing inputs in your app

How you access inputs depends on how they’re configured:

  1. Environment variables: If env_var is specified, the input is available as an environment variable
  2. Mounted paths: File and directory inputs are mounted at the specified path
  3. Flyte SDK: Use the Flyte SDK to access input values programmatically
import os

# Input with env_var specified
env = flyte.app.AppEnvironment(
    name="my-app",
    flyte.app.Input(
        name="model_file",
        value=flyte.io.File("s3://bucket/model.pkl"),
        mount="/app/models/model.pkl",
        env_var="MODEL_PATH",
    ),
    # ...
)

# Access in the app via the environment variable
API_KEY = os.getenv("API_KEY")

# Access in the app via the mounted path
with open("/app/models/model.pkl", "rb") as f:
    model = pickle.load(f)

# Access in the app via the Flyte SDK (for string inputs)
input_value = flyte.app.get_input("config")  # Returns string value

Best practices

  1. Use delayed inputs: Leverage RunOutput and AppEndpoint to create app dependencies between tasks and apps, or app-to-app chains.
  2. Override for testing: Use the input_values parameter when serving to test different configurations without changing code.
  3. Mount paths clearly: Use descriptive mount paths for file/directory inputs so your app code is easy to understand.
  4. Use environment variables: For simple constants that you can hard-code, use env_var to inject values as environment variables.
  5. Production deployments: For production, define inputs in the AppEnvironment rather than overriding them at deploy time.

Limitations

  • Large files/directories can slow down app startup.
  • Input overrides are only available when using flyte.with_servecontext(...).serve(...).