Serving Applications

Model a long-running application

When you need an HTTP API, dashboard, or model server to live beside Flyte tasks, use flyte.app.AppEnvironment rather than a task environment. AppEnvironment carries the deployment contract: an image, resources, port, authentication requirement, scaling policy, dependencies, links, parameters, cluster pool, and startup/health-check timeouts.

The base environment defaults to port 8080, authentication required, the default cluster pool, and Scaling() (scale-to-zero with a maximum of one replica). It inherits Environment fields such as resources, env_vars, secrets, depends_on, and include. App names are validated against a lowercase alphanumeric/hyphen/dot pattern; ports 8012, 8022, 8112, 9090, and 9091 are reserved and rejected.

Choose the process contract

Use args when the application runtime already has a container command and needs additional arguments. AppEnvironment.container_args() returns a list unchanged, and uses shlex.split() when args is a string. Use command when you need to replace the container command entirely; string commands are also split with shlex.split() and list commands are returned as supplied.

If you omit command, AppEnvironment.container_cmd() builds an fserve command containing the serialization version, project, domain, organization, optional image-cache and code-bundle arguments, serialized parameters, and the raw-data template {{.rawOutputDataPrefix}}. It appends resolver arguments when the app's declaring module can be resolved.

Parameters require the fserve runtime. Consequently, AppEnvironment.__post_init__ rejects parameters combined with a custom command whose first token is not fserve. Use args or register the process with @app_env.server instead:

@app_env.server
async def run_server():
    ...

The server, on_startup, and on_shutdown methods are decorators that store the corresponding callbacks on the environment. Startup runs before the server callback, and shutdown runs after it. These callbacks may be synchronous or asynchronous and accept values corresponding to the environment's Parameter definitions.

Adapt an existing FastAPI application

When you already have a FastAPI object, pass it to FastAPIAppEnvironment. The README's minimal deployment is:

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"
    ),
)

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

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

FastAPIAppEnvironment.__post_init__ verifies that FastAPI and Starlette are installed, requires app to be an actual fastapi.FastAPI instance, and rejects None or another ASGI application type. It replaces app.state with a picklable state implementation because Starlette's normal state object has circular-reference pickling behavior. Code should therefore rely on the normal FastAPI state interface, not on the original state object identity.

The adapter sets type to "FastAPI", prepends a relative /docs flyte.app.Link titled FastAPI OpenAPI Docs, and registers _fastapi_app_server as the environment server. container_command() deliberately returns []: the adapter does not describe the service as a literal uvicorn app:app container command. When the server callback starts, it imports Uvicorn, creates uvicorn.Config(self.app, port=self.port.port) if no configuration was supplied, fills a supplied configuration's missing port from the environment, and awaits uvicorn.Server(...).serve().

FastAPI and Starlette are checked when the environment is constructed; Uvicorn is checked when the server starts. Install fastapi and uvicorn in the image used by a remote deployment and make them available to the local Python process as well.

Serve remotely or locally

Call flyte.serve(env) for the direct serving path. It returns an AppHandle, whose common interface works for both remote flyte.remote.App and local _LocalApp handles:

flyte.init_from_config()
served = flyte.serve(env)
print(served.url)
print(served.endpoint)
served.activate()

The handle exposes name, url, endpoint, is_active(), is_deactivated(), activate(wait=False), deactivate(wait=False), and synchronous/asynchronous ephemeral context managers. Remote serving requires Flyte initialization. For more deployment controls, flyte.with_servecontext(...) accepts project and domain selection, environment-variable and parameter overrides, cluster-pool selection, logging settings, code-bundle settings, and local health-check settings.

For local development, select local mode explicitly. The local server is non-blocking; wait for readiness before making requests and deactivate it when finished:

local_app = flyte.with_servecontext(mode="local").serve(app_env)
local_app.activate(wait=True)
print(f"App is ready at {local_app.endpoint}")

local_app.deactivate(wait=True)

In local mode, AppEnvironment.endpoint is http://localhost:<port>. In remote mode, it first checks INTERNAL_APP_ENDPOINT_PATTERN; if set, the value is formatted with app_fqdn=self.name. Otherwise it ensures a Flyte client and resolves flyte.remote.App.get(name=self.name).endpoint. Reading .endpoint remotely therefore requires client initialization unless the internal endpoint pattern is configured.

A task can call a locally served app over HTTP. The repository's local example serves an app on /, then calls the endpoint returned by the environment:

import httpx

@task_env.task
async def add_one_task(x: int) -> int:
    print(f"Calling app at {app_env.endpoint}")
    async with httpx.AsyncClient() as client:
        response = await client.get(app_env.endpoint, params={"x": x})
        response.raise_for_status()
        return response.json()["result"]

In examples/apps/local_fastapi_app.py, the task is run with flyte.with_runcontext(mode="local"), after activate(wait=True), and the app is finally stopped with deactivate(wait=True).

Integrate deployment configuration

Use the app-specific fields to describe how the service is deployed rather than embedding those choices in the FastAPI process. resources accepts CPU, memory, GPU, disk, and shared-memory settings. scaling controls replicas; domain configures custom domain behavior, cluster_pool defaults to "default", and timeouts controls startup and health checks. clone_with() creates another environment while preserving the existing configuration and applying selected overrides such as name, image, port, scaling, parameters, or domain.

App parameters can connect deployment-time values to task outputs. The model-serving example uses a file RunOutput and requests that it be downloaded:

parameters=[
    flyte.app.Parameter(
        name="model",
        value=flyte.app.RunOutput(
            task_name="cv-export.export_model_to_pt2",
            type="file",
        ),
        download=True,
    )
]

Declare an app dependency with depends_on when one served app calls another. The serving planner includes dependent environments in the deployment plan:

caller_env = FastAPIAppEnvironment(
    name="caller",
    app=caller_app,
    depends_on=[target_env],
)

The caller can then use the target environment's endpoint through an HTTP client. The dependency declaration affects deployment planning; it does not itself make a route call or inject an HTTP client.

Pass through Flyte authentication in FastAPI

If an endpoint must use the caller's Flyte credentials for remote operations, install FastAPIPassthroughAuthMiddleware. Its default extractors look for Authorization and Cookie headers:

app.add_middleware(
    FastAPIPassthroughAuthMiddleware,
    excluded_paths={"/health"},
)

The middleware checks every non-excluded request. It calls all configured extractors, collects non-None (key, value) tuples, and returns a 401 JSON response with {"detail": "Authentication credentials required"} and WWW-Authenticate: Bearer when no extractor succeeds. Otherwise it wraps the downstream handler in flyte.remote.auth_metadata(*auth_tuples), making the extracted metadata available during that request.

The repository's manual webhook example initializes passthrough access in a FastAPI lifespan before yielding control:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from flyte.app.extras import FastAPIPassthroughAuthMiddleware

@asynccontextmanager
async def lifespan(app: FastAPI):
    await flyte.init_passthrough.aio(
        project=flyte.current_project(),
        domain=flyte.current_domain(),
    )
    yield

app = FastAPI(lifespan=lifespan)
app.add_middleware(FastAPIPassthroughAuthMiddleware, excluded_paths={"/health"})

For another credential header, create an extractor with extract_custom_header(). Header lookup is case-insensitive, and the returned extractor reports the lowercased header name:

api_key_extractor = FastAPIPassthroughAuthMiddleware.extract_custom_header("x-api-key")
app.add_middleware(
    FastAPIPassthroughAuthMiddleware,
    header_extractors=[api_key_extractor],
)

Extractor exceptions are logged and do not abort the middleware loop. Authentication metadata is scoped through Flyte's context mechanism, so concurrent requests use their own extracted metadata. Exclusions are exact URL paths; the generated webhook excludes /health, /docs, /openapi.json, and /redoc, not arbitrary path variants.

Use the pre-built Flyte webhook

FlyteWebhookAppEnvironment packages a FastAPI service for common Flyte control-plane operations. It creates the FastAPI app itself, uses type="FlyteWebhookApp", defaults its image to a Debian image with fastapi and uvicorn, installs passthrough authentication, and delegates serving to the FastAPI adapter.

The minimal construction and deployment path is:

import flyte
from flyte.app.extras import FlyteWebhookAppEnvironment

webhook_env = FlyteWebhookAppEnvironment(
    name="my-webhook",
    image=flyte.Image.from_debian_base().with_pip_packages("fastapi", "uvicorn"),
    resources=flyte.Resources(),
)

flyte.serve(webhook_env)

With neither selector supplied, all registered webhook endpoints are enabled. Use endpoint_groups to select a group, or endpoints to select individual routes:

Group	Operations
core	health, me
task	run_task, get_task
run	get_run, get_run_io, abort_run
app	get_app, activate_app, deactivate_app, call_app
trigger	activate_trigger, deactivate_trigger
build	build_image
prefetch	prefetch_hf_model, get_prefetch_hf_model, get_prefetch_hf_model_io, abort_prefetch_hf_model

For example, this enables only core, task, and run routes:

webhook_env = FlyteWebhookAppEnvironment(
    name="task-runner-webhook",
    endpoint_groups=["core", "task", "run"],
)

Individual selection has the same public construction shape:

webhook_env = FlyteWebhookAppEnvironment(
    name="minimal-webhook",
    endpoints=["health", "run_task", "get_run"],
)

Do not pass both endpoint_groups and endpoints: FlyteWebhookAppEnvironment.__post_init__ raises ValueError("Cannot specify both endpoints and endpoint_groups"). Unknown groups and endpoint names are skipped with warnings. The generated app registers the resolved endpoint functions with app.add_api_route.

Restrict webhook resources

Task, app, and trigger allowlists are checked by the generated handlers. Task entries support three exact forms:

• domain/project/name, such as production/my-project/my-task;
• project/name, matching that project and task across domains;
• name, matching that task name across domains and projects.

The allowlist tests demonstrate the distinctions:

# Full path match
allowlist_full = ["production/my-project/my-task"]
assert is_task_allowed(allowlist_full, "production", "my-project", "my-task") is True
assert is_task_allowed(allowlist_full, "development", "my-project", "my-task") is False

# Project/name match (any domain)
allowlist_proj = ["my-project/my-task"]
assert is_task_allowed(allowlist_proj, "production", "my-project", "my-task") is True
assert is_task_allowed(allowlist_proj, "development", "my-project", "my-task") is True

# Name-only match (any domain/project)
allowlist_name = ["my-task"]
assert is_task_allowed(allowlist_name, "production", "my-project", "my-task") is True
assert is_task_allowed(allowlist_name, "development", "other-project", "my-task") is True

app_allowlist matches app names exactly. trigger_allowlist accepts task_name/trigger_name or a bare trigger name. Disallowed operations return HTTP 403. The webhook also rejects attempts to query, activate, deactivate, or call the webhook app itself with HTTP 400; missing tasks, runs, or apps map to 404.

Webhook authentication is initialized in the generated app's FastAPI lifespan and uses FastAPIPassthroughAuthMiddleware. /health, /docs, /openapi.json, and /redoc bypass that middleware; other generated routes require at least one extracted credential. For a deployed webhook, the example reads FLYTE_API_KEY and sends it as Authorization: Bearer <token>:

token = os.getenv("FLYTE_API_KEY")
if not token:
    raise ValueError("FLYTE_API_KEY not set. Obtain with: flyte get api-key")

headers = {"Authorization": f"Bearer {token}"}

requires_auth=False on an app environment does not remove middleware installed by FlyteWebhookAppEnvironment; the middleware's request behavior is configured by the generated FastAPI app.

Troubleshoot serving failures

• Missing packages: constructing FastAPIAppEnvironment raises an explicit ModuleNotFoundError if FastAPI or Starlette is unavailable. Uvicorn is checked only when _fastapi_app_server() starts.
• Wrong app object: app must be a fastapi.FastAPI instance; an arbitrary ASGI application is rejected.
• Unexpected command assumptions: FastAPI serving uses the registered Uvicorn server callback and an empty container_command(), not a generated uvicorn app:app command.
• Invalid names or ports: use lowercase app names with valid hyphen/dot placement and avoid the five reserved ports enforced by AppEnvironment.
• Parameters with a command: use args or @app_env.server, or make a custom command start with fserve; otherwise construction fails.
• Remote endpoint lookup: initialize Flyte before reading a remote .endpoint, or configure INTERNAL_APP_ENDPOINT_PATTERN.
• Unauthenticated webhook requests: exact non-excluded paths return 401 unless an extractor finds an authorization or cookie header (or a configured custom header). Obtain the example token with flyte get api-key and provide it through FLYTE_API_KEY.
• Live integration requirements: the exhaustive webhook example exercises live Flyte services and credentials; its Hugging Face prefetch flow additionally requires an HF_TOKEN secret and is not a self-contained unit test.