FastAPI Integrations and Webhook Apps

Choose the FastAPI integration

Use FastAPIAppEnvironment when you own the fastapi.FastAPI instance; use FlyteWebhookAppEnvironment when you want Flyte's generated task, run, app, trigger, build, and HuggingFace-prefetch routes.

Serve an application you define

The shipped local example creates the FastAPI application first and passes it to FastAPIAppEnvironment:

import httpx
from fastapi import FastAPI

import flyte
from flyte.app.extras import FastAPIAppEnvironment

app = FastAPI(
    title="Local Add One",
    description="A local FastAPI app that adds one to the input",
    version="1.0.0",
)

app_env = FastAPIAppEnvironment(
    name="local-add-one",
    app=app,
    description="Adds one to the input",
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages("fastapi", "uvicorn", "httpx"),
    resources=flyte.Resources(),
    requires_auth=False,
)

FastAPIAppEnvironment is a dataclass subclass of flyte.app.AppEnvironment. Its app field must be a non-null fastapi.FastAPI; initialization raises ValueError for None and TypeError for another object type. FastAPI and Starlette are imported when the environment is initialized, so missing packages produce explicit installation errors. Uvicorn is imported when the server starts.

The environment adds a relative Flyte link for /docs, registers its _fastapi_app_server as the environment server, and starts Uvicorn with the inherited app-environment port. If no uvicorn.Config is supplied, it creates one for the application; if a supplied config has port=None, it fills that value from the environment port.

A local app can be called through the served application's endpoint rather than constructing a URL:

@task_env.task
async def add_one_task(x: int) -> int:
    """Task that calls the local app endpoint."""
    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"]


if __name__ == "__main__":
    local_app = flyte.with_servecontext(mode="local").serve(app_env)
    local_app.activate(wait=True)
    result = flyte.with_runcontext(mode="local").run(add_one_task, x=5)
    print(f"Result: {result.outputs()[0]}")
    local_app.deactivate(wait=True)

The complete example defines task_env and the @app.get("/") handler before this task. serve(...).activate(wait=True) makes the app available, app_env.endpoint is the address used by the task, and deactivate(wait=True) shuts it down.

Pass Flyte authentication through FastAPI

For a custom webhook, initialize Flyte passthrough authentication during FastAPI startup and add FastAPIPassthroughAuthMiddleware:

from contextlib import asynccontextmanager

from fastapi import FastAPI

import flyte
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(
    title="Flyte Webhook Runner",
    description="A webhook service that triggers Flyte task runs",
    version="1.0.0",
    lifespan=lifespan,
)

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

The middleware subclasses Starlette's BaseHTTPMiddleware. By default it runs extract_authorization_header and extract_cookie_header for each non-excluded request. Each present header becomes a tuple—("authorization", value) or ("cookie", value)—and all resulting tuples are passed to flyte.remote.auth_metadata while the downstream handler runs. A handler can therefore call Flyte remote APIs without entering auth_metadata() itself:

@app.get("/me")
async def get_current_user():
    user = await remote.User.get.aio()
    return {
        "subject": user.subject(),
        "name": user.name(),
    }

In the real run_webhook_fastapi.py example, remote is imported from flyte, and the handler catches failures from remote.User.get.aio() and converts them to an HTTP 401 response.

Configure extractors and exclusions

Supplying header_extractors replaces the two defaults; it does not extend them. The static factory creates a case-insensitive extractor for a custom header:

api_key_extractor = FastAPIPassthroughAuthMiddleware.extract_custom_header("x-api-key")

app.add_middleware(
    FastAPIPassthroughAuthMiddleware,
    header_extractors=[api_key_extractor],
    excluded_paths={"/health"},
)

Each extractor receives a Starlette request and returns a (key, value) tuple or None. If an extractor raises, the middleware logs a warning containing the extractor name and continues with the other extractors. If no extractor returns a tuple, the middleware does not call the route: it returns status 401 with this JSON body and header:

{"detail": "Authentication credentials required"}
WWW-Authenticate: Bearer

excluded_paths uses exact URL-path matching. For example, /health excludes /health, but the middleware does not treat that value as a prefix rule for other paths.

requires_auth belongs to the inherited AppEnvironment and controls the deployed app's platform or gateway authentication policy. It does not remove FastAPIPassthroughAuthMiddleware. A generated webhook still requires a request header for every middleware-protected route even when the platform-level setting is configured separately.

Use the pre-built webhook environment

FlyteWebhookAppEnvironment creates its own FastAPI application, installs the passthrough middleware, and registers the selected routes. A minimal deployment only needs the inherited environment name:

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)

If image is omitted, the environment default is flyte.Image.from_debian_base().with_pip_packages("fastapi", "uvicorn"). Its type is "FlyteWebhookApp". The generated FastAPI metadata uses the configured title, or Flyte Webhook: {name}, the configured description, or A webhook service for Flyte operations, and version 1.0.0.

Its lifespan calls flyte.init_passthrough.aio(project=flyte.current_project(), domain=flyte.current_domain()). The generated app installs FastAPIPassthroughAuthMiddleware with these unauthenticated exact paths:

excluded_paths = {"/health", "/docs", "/openapi.json", "/redoc"}

The inherited Uvicorn server behavior is used, and container_command() returns []; startup is provided through the environment server callback rather than a generated container command.

Generated route families

The internal _EndpointConfig dataclass stores method, path, handler, and an optional name for one registry entry. _create_webhook_app() uses those records with FastAPI.add_api_route(); _EndpointConfig is an internal registration record, not the public way to add webhook endpoints.

The registry contains these routes:

Endpoint name	Method and path	Operation family
health	GET /health	Health check
me	GET /me	Current user
run_task	POST /run-task/{domain}/{project}/{name}	Start a task run
get_task	GET /task/{domain}/{project}/{name}	Task metadata
get_run	GET /run/{name}	Run status and metadata
get_run_io	GET /run/{name}/io	Run inputs and outputs
abort_run	POST /run/{name}/abort	Abort a run
get_app	GET /app/{name}	App status
activate_app	POST /app/{name}/activate	Activate an app
deactivate_app	POST /app/{name}/deactivate	Deactivate an app
call_app	POST /app/{name}/call	Call another app endpoint
activate_trigger	POST /trigger/{task_name}/{trigger_name}/activate	Activate a trigger
deactivate_trigger	POST /trigger/{task_name}/{trigger_name}/deactivate	Deactivate a trigger
build_image	POST /build-image	Start an image build
prefetch_hf_model	POST /prefetch/hf-model	Start HuggingFace prefetching
get_prefetch_hf_model	GET /prefetch/hf-model/{run_name}	Prefetch run status
get_prefetch_hf_model_io	GET /prefetch/hf-model/{run_name}/io	Prefetch inputs and outputs
abort_prefetch_hf_model	POST /prefetch/hf-model/{run_name}/abort	Abort a prefetch run

The handlers use Flyte remote resources including Task, Run, App, Trigger, and User. Task execution uses flyte.run; image building uses flyte.build.aio; and HuggingFace operations use flyte.prefetch.hf_model with flyte.Resources for the prefetch task.

Limit the exposed operations

Use either endpoint_groups or endpoints. FlyteWebhookAppEnvironment.__post_init__() raises ValueError("Cannot specify both endpoints and endpoint_groups") when both are supplied.

Groups are all, core, task, run, app, trigger, build, and prefetch. For example, this exposes health/user, task, and run operations but not app, trigger, build, or prefetch operations:

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

Individual names can be selected instead:

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

The tests verify that this selection creates /health, /run-task/{domain}/{project}/{name}, and /run/{name}, while omitting /me and the other unselected routes. With neither filter, _resolve_endpoints() enables all registered endpoints, including control-plane operations such as activation, trigger changes, image builds, and model prefetching. Unknown runtime group or endpoint strings are logged and skipped.

Apply allowlists

Task, app, and trigger allowlists are checked by the generated handlers. A task entry can use any of these forms:

webhook_env = FlyteWebhookAppEnvironment(
    name="restricted-webhook",
    endpoint_groups=["core", "task", "run"],
    task_allowlist=[
        "production/my-project/my-task",  # domain/project/name
        "my-project/another-task",        # project/name, any domain
        "name-only-task",                 # name, any domain/project
    ],
)

The executable tests confirm that a full production/my-project/my-task entry does not match the same task in development; my-project/my-task matches in both domains; and my-task matches regardless of domain and project. None permits all tasks. app_allowlist matches exact app names. trigger_allowlist accepts task_name/trigger_name or a bare trigger name.

App operations also reject targeting the webhook app itself, with errors such as Cannot get status of self, Cannot activate self, Cannot deactivate self, and Cannot call self.

Call a deployed webhook

The example in examples/apps/run_webhook_env.py uses FLYTE_API_KEY, sends it as a Bearer token, checks the public health route, then calls authenticated task and run routes:

import os

import httpx


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}",
    "User-Agent": "flyte-webhook-client/1.0",
}

served_app.activate()

with httpx.Client(headers=headers) as client:
    health_resp = client.get(endpoint.rstrip("/") + "/health")
    health_resp.raise_for_status()

    me_resp = client.get(endpoint.rstrip("/") + "/me")
    me_resp.raise_for_status()

    data = {"x": 42, "y": "hello"}
    route = "/run-task/development/flytesnacks/webhook-env-task.example_task"
    resp = client.post(endpoint.rstrip("/") + route, json=data)
    if resp.is_success:
        run_name = resp.json()["name"]
        run_resp = client.get(endpoint.rstrip("/") + f"/run/{run_name}")
        run_resp.raise_for_status()

The example obtains endpoint from the served app, after flyte.serve(webhook_env), and deploys the task environment before serving the webhook. /health is excluded from middleware authentication; /me, /run-task/..., and /run/... use the forwarded request credentials.

Troubleshoot deployment and route behavior

• Missing dependencies: FastAPIAppEnvironment reports missing fastapi or starlette during initialization. The server reports missing uvicorn when it starts. Custom images therefore need the web dependencies, while the webhook environment's default image includes FastAPI and Uvicorn.
• Port configuration: The inherited app-environment port defaults to 8080. Reserved ports 8012, 8022, 8112, 9090, and 9091 are rejected. A supplied uvicorn.Config is retained, with its port filled only when it is None.
• Serialization: Initialization replaces app.state with a locally defined picklable Starlette State subclass. Code that relies on the original state object should account for that replacement.
• Run I/O: Run I/O returns outputs=None when a run is incomplete or output retrieval fails; failed runs may have no outputs. Flyte input/output wrapper values are converted to JSON-compatible dictionaries and lists.
• Build and prefetch timing: The build-image handler calls flyte.build.aio(..., wait=False), so the build is asynchronous. HuggingFace prefetch has source defaults of secret key HF_TOKEN, CPU 2, memory 8Gi, disk 50Gi, task auto, modality ("text",), and force=0. The token key names a Flyte secret; it is not read directly from the webhook process environment.
• Calling another app: call_app constructs a raw urllib.request request. Although its source comment refers to forwarding current authentication, the visible request headers contain Content-Type and a generated User-Agent, not an explicitly copied Authorization or Cookie header; verify this behavior before relying on downstream credential forwarding.
• Authentication confusion: requires_auth does not disable the generated middleware. Add a request header or use one of the generated exact exclusions; for a custom app, explicitly exclude paths such as /health when appropriate.

tests/flyte/app/extras/test_webhook_app.py is the focused executable reference for webhook defaults, generated routes, middleware registration, endpoint filtering, allowlists, name and port validation, and health behavior.