How to add feature flags and experiments to a FastAPI backend

# Keep AMPLITUDE_API_KEY from the product analytics tutorial.
export AMPLITUDE_DEPLOYMENT_KEY="your-server-deployment-key"
export AMPLITUDE_SERVER_ZONE="US"

uv add "amplitude-experiment>=1.13,<2"

# Keep analytics and authentication from the product analytics tutorial.
import logging
import os
from threading import Lock
from amplitude_experiment import Experiment, RemoteEvaluationConfig, User
from amplitude_experiment.remote.fetch_options import FetchOptions
from amplitude_experiment.server_zone import ServerZone

FLAG_KEY = "report-summary"
_client = None
_lock = Lock()
log = logging.getLogger(__name__)


def get_experiment():
    global _client
    with _lock:
        if _client is None:
            zone = ServerZone.EU if os.environ.get("AMPLITUDE_SERVER_ZONE", "US") == "EU" else ServerZone.US
            _client = Experiment.initialize_remote(
                os.environ["AMPLITUDE_DEPLOYMENT_KEY"],
                RemoteEvaluationConfig(
                    server_zone=zone,
                    fetch_timeout_millis=500,
                    fetch_retries=0,
                    connection_pool_max_size=8,
                    fetch_pool_acquire_timeout_millis=100,
                ),
            )
        return _client


def choose_report_layout(user_id, account_type):
    # This experiment is for signed-in users. Auth determines eligibility.
    if user_id is None:
        return "control"
    client = get_experiment()
    try:
        variants = client.fetch_v2(
            User(user_id=user_id, user_properties={"account_type": account_type}),
            FetchOptions(tracksExposure=True, flagKeys=[FLAG_KEY]),
        ) or {}
        variant = variants.get(FLAG_KEY)
        if variant is not None and variant.value in {"control", "treatment"}:
            return variant.value
    except Exception:
        # Availability wins for this cosmetic layout. Keep authorization separate.
        log.warning("Experiment evaluation failed; serving the baseline report")
    return "control"


def close_experiment():
    # Call only at worker exit, after requests have drained.
    if _client is not None:
        _client.close()

from amplitude import BaseEvent
from .analytics import get_client


def record_report_export(user_id, device_id, framework):
    if not isinstance(device_id, str) or not 5 <= len(device_id) <= 1024:
        device_id = None
    if user_id is None and device_id is None:
        return False
    get_client().track(BaseEvent(
        event_type="Report Exported",
        user_id=user_id,
        device_id=device_id,
        event_properties={"report_type": "usage", "days": 7, "framework": framework},
    ))
    return True

# Keep routes, middleware, and lifespan work from /build/fastapi-product-analytics.
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, Request
from fastapi.responses import Response
from starlette.concurrency import run_in_threadpool
from .analytics import get_client, record_report_preview, shutdown_client
from .auth import AuthUser, get_current_user
from .experiments import choose_report_layout, close_experiment, get_experiment
from .report_events import record_report_export


@asynccontextmanager
async def lifespan(app: FastAPI):
    get_client()
    get_experiment()
    try:
        yield
    finally:
        await run_in_threadpool(close_experiment)
        await run_in_threadpool(shutdown_client)


app = FastAPI(lifespan=lifespan)


@app.post("/api/reports/preview")
async def report_preview(
    request: Request,
    user: Annotated[AuthUser | None, Depends(get_current_user)],
):
    user_id = f"user-{user.id}" if user is not None else None
    account_type = user.account_type if user is not None else None
    layout = await run_in_threadpool(choose_report_layout, user_id, account_type)
    preview = {"report_type": "usage", "days": 7, "layout": layout}
    if layout == "treatment":
        preview["summary"] = "Your last seven days of usage are ready to export."
    await run_in_threadpool(record_report_preview, user_id, request.headers.get("X-Amplitude-Device-Id"), "fastapi", account_type)
    return preview


@app.post("/api/reports/export")
async def report_export(
    request: Request,
    user: Annotated[AuthUser | None, Depends(get_current_user)],
):
    user_id = f"user-{user.id}" if user is not None else None
    response = Response(
        "report_type,days\nusage,7\n",
        media_type="text/csv",
        headers={"Content-Disposition": 'attachment; filename="usage.csv"'},
    )
    await run_in_threadpool(record_report_export, user_id, request.headers.get("X-Amplitude-Device-Id"), "fastapi")
    return response

# Keep your existing authenticated cookies in cookies.txt.
ORIGIN="http://127.0.0.1:8000"
curl --fail-with-body -b cookies.txt -X POST "$ORIGIN/api/reports/preview"   -H "X-Amplitude-Device-Id: existing-browser-device-id"

curl --fail-with-body -b cookies.txt -X POST "$ORIGIN/api/reports/export" -H "X-Amplitude-Device-Id: existing-browser-device-id" -o usage.csv

# -- app/experiments.py --
# Keep analytics and authentication from the product analytics tutorial.
import logging
import os
from threading import Lock
from amplitude_experiment import Experiment, RemoteEvaluationConfig, User
from amplitude_experiment.remote.fetch_options import FetchOptions
from amplitude_experiment.server_zone import ServerZone

FLAG_KEY = "report-summary"
_client = None
_lock = Lock()
log = logging.getLogger(__name__)


def get_experiment():
    global _client
    with _lock:
        if _client is None:
            zone = ServerZone.EU if os.environ.get("AMPLITUDE_SERVER_ZONE", "US") == "EU" else ServerZone.US
            _client = Experiment.initialize_remote(
                os.environ["AMPLITUDE_DEPLOYMENT_KEY"],
                RemoteEvaluationConfig(
                    server_zone=zone,
                    fetch_timeout_millis=500,
                    fetch_retries=0,
                    connection_pool_max_size=8,
                    fetch_pool_acquire_timeout_millis=100,
                ),
            )
        return _client


def choose_report_layout(user_id, account_type):
    # This experiment is for signed-in users. Auth determines eligibility.
    if user_id is None:
        return "control"
    client = get_experiment()
    try:
        variants = client.fetch_v2(
            User(user_id=user_id, user_properties={"account_type": account_type}),
            FetchOptions(tracksExposure=True, flagKeys=[FLAG_KEY]),
        ) or {}
        variant = variants.get(FLAG_KEY)
        if variant is not None and variant.value in {"control", "treatment"}:
            return variant.value
    except Exception:
        # Availability wins for this cosmetic layout. Keep authorization separate.
        log.warning("Experiment evaluation failed; serving the baseline report")
    return "control"


def close_experiment():
    # Call only at worker exit, after requests have drained.
    if _client is not None:
        _client.close()

# -- app/report_events.py --
from amplitude import BaseEvent
from .analytics import get_client


def record_report_export(user_id, device_id, framework):
    if not isinstance(device_id, str) or not 5 <= len(device_id) <= 1024:
        device_id = None
    if user_id is None and device_id is None:
        return False
    get_client().track(BaseEvent(
        event_type="Report Exported",
        user_id=user_id,
        device_id=device_id,
        event_properties={"report_type": "usage", "days": 7, "framework": framework},
    ))
    return True

# -- app/main.py --
# Keep routes, middleware, and lifespan work from /build/fastapi-product-analytics.
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, Request
from fastapi.responses import Response
from starlette.concurrency import run_in_threadpool
from .analytics import get_client, record_report_preview, shutdown_client
from .auth import AuthUser, get_current_user
from .experiments import choose_report_layout, close_experiment, get_experiment
from .report_events import record_report_export


@asynccontextmanager
async def lifespan(app: FastAPI):
    get_client()
    get_experiment()
    try:
        yield
    finally:
        await run_in_threadpool(close_experiment)
        await run_in_threadpool(shutdown_client)


app = FastAPI(lifespan=lifespan)


@app.post("/api/reports/preview")
async def report_preview(
    request: Request,
    user: Annotated[AuthUser | None, Depends(get_current_user)],
):
    user_id = f"user-{user.id}" if user is not None else None
    account_type = user.account_type if user is not None else None
    layout = await run_in_threadpool(choose_report_layout, user_id, account_type)
    preview = {"report_type": "usage", "days": 7, "layout": layout}
    if layout == "treatment":
        preview["summary"] = "Your last seven days of usage are ready to export."
    await run_in_threadpool(record_report_preview, user_id, request.headers.get("X-Amplitude-Device-Id"), "fastapi", account_type)
    return preview


@app.post("/api/reports/export")
async def report_export(
    request: Request,
    user: Annotated[AuthUser | None, Depends(get_current_user)],
):
    user_id = f"user-{user.id}" if user is not None else None
    response = Response(
        "report_type,days\nusage,7\n",
        media_type="text/csv",
        headers={"Content-Disposition": 'attachment; filename="usage.csv"'},
    )
    await run_in_threadpool(record_report_export, user_id, request.headers.get("X-Amplitude-Device-Id"), "fastapi")
    return response

Extend the existing FastAPI report example using https://amplitude.com/build/fastapi-feature-flags.
Keep app/analytics.py and app/auth.py from /build/fastapi-product-analytics unchanged, including real verified authentication and worker shutdown flushing. Merge route and lifespan changes; preserve unrelated code and authorization/CSRF controls.
Use amplitude-experiment>=1.13,<2 with one remote client per worker. Read AMPLITUDE_DEPLOYMENT_KEY only on the server; it must belong to the analytics project. Honor AMPLITUDE_SERVER_ZONE for both SDKs.
Use the tested FetchOptions import from amplitude_experiment.remote.fetch_options, with tracksExposure=True and flagKeys=["report-summary"]. Supply a new User with the canonical authenticated ID and trusted account_type per request.
Evaluate only at POST /api/reports/preview. Serve control or treatment; only treatment adds summary. Anonymous identity, missing/unknown variants, and evaluation errors use the baseline without inventing exposures. Keep authentication separate from the flag.
Add POST /api/reports/export to return CSV and send Report Exported after preparing the response. Do not fetch a flag or manually send exposure on export. Keep the same user ID as evaluation and product analytics.
Run synchronous evaluation and analytics in run_in_threadpool; merge client initialization/cleanup into the existing lifespan.
Verify both variants, property targeting context, scoped exposure requests, conversion identity, timeout/failure fallback, concurrency, and shutdown delivery. Confirm remote exposures and experiment analysis separately in Amplitude. Report every file changed.