Great question. Assuming Prefect 3.x, the community’s go-to patterns for “long‑lived container + fast flow starts” are: Preferred options (fastest first) - Serve the flow inside the container with

flow.serve()

(or

serve(deployment, ...)

) - Ultra low-latency: the flow runs in-process/subprocess without provisioning new infrastructure. - No worker/work pool required; the container continuously polls for its own work. - Best when you want a single-purpose, always-on service that can be triggered frequently. - Run a Process worker inside the long-lived container pointing at a dedicated work pool - Very low-latency: each run is just a subprocess spawn inside the same container. - Best when you prefer standard deployments/work-pool orchestration and may run multiple different flows. When to avoid - Docker/Kubernetes workers for each run: great for isolation/scale but cold starts are seconds to tens of seconds (image pull + container start), so generally not “fast” for frequent triggers. How to set it up Option A: Serve the flow (fastest) - Container runs your code and “serves” the deployment in-process. - You trigger it via events/automations or

run_deployment

.

# serve_flow.py
from prefect import flow, serve

@flow
def low_latency_flow(name: str = "world"):
    return f"Hello, {name}"

if __name__ == "__main__":
    # Creates/updates a deployment and keeps the container polling for work
    serve(
        low_latency_flow.to_deployment(name="low-latency-deployment")
    )

Dockerfile sketch:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY serve_flow.py .
ENV PREFECT_API_URL=...
ENV PREFECT_API_KEY=...
CMD ["python", "serve_flow.py"]

Trigger it from code (keeps separate run tracking):

from prefect.deployments import run_deployment

# anywhere (e.g., in your API server)
state = run_deployment(
    name="low-latency-flow/low-latency-deployment",
    parameters={"name": "Marvin"}
)

Option B: Long-lived container running a Process worker - Create a process work pool; deploy your flow to that pool; keep a worker running in your container. Create a work pool:

prefect work-pool create "my-process-pool" --type process

Deploy your flow to that pool (example sketch):

# deploy.py
from prefect import flow
from prefect.deployments import deploy

@flow
def my_flow(x: int = 1):
    return x + 1

if __name__ == "__main__":
    deploy(
        entrypoint="./deploy.py:my_flow",
        name="fast-deployment",
        work_pool_name="my-process-pool",
        description="Low-latency process worker deployment",
    )

Run the worker in your container:

prefect worker start --pool my-process-pool

Trigger it:

from prefect.deployments import run_deployment

state = run_deployment(
    name="my_flow/fast-deployment",
    parameters={"x": 41}
)

Latency tradeoffs and tips - Directly calling the flow function (import and call) is fastest, but it will be tracked as a subflow of your calling flow/process, not as a separate deployment run. Use this if you need absolute minimal overhead and don’t need independent run tracking. -

run_deployment

is very fast for process/served containers (typically sub-200ms overhead plus your flow), and preserves independent run records, retries, and scheduling. - If you must use Docker/Kubernetes workers, expect seconds of cold start; mitigate with pre-pulled images, node-local registries, and keeping pools warm. For slow-start infra, tune worker prefetch to help readiness once the worker has polled. Which should you choose? - Need sub-second starts and a single purpose container? Serve the flow in that container (

serve(...)

). - Want standard pools/workers and might host multiple flows? Run a Process worker in the long-lived container and deploy to its work pool. - Need strict isolation per run or autoscaling? Use Docker/Kubernetes workers (accept slower starts).

Follow-up questions to tailor the setup - What’s your target p50/p95 latency? - Are you on Docker, Kubernetes, or a VM? - Do you need each invocation as a separately tracked deployment run, or is a subflow call sufficient? - Do you expect to run many different flows from the same container? Verified CLI references -

prefect deploy --help

-

prefect worker start --help

-

prefect work-pool create --help

If you want, share a bit more about your environment and latency goals, and I’ll sketch the exact Dockerfile/entrypoint and the cleanest trigger path for you.

Short answer: you can do it either way. For low-latency from a long‑lived container, the two common community patterns are: 1) Keep “classic” deployments and run a Process worker in the container - Yes, you’d still have a deploy script (or

prefect deploy

) that registers the deployment to a process work pool. - Your container just runs the worker and picks up runs instantly (no infra spin-up). - Example:

# create a process work pool once
prefect work-pool create "my-process-pool" --type process

# deploy script (runs in CI)
from prefect import flow
from prefect.deployments import deploy

@flow
def my_flow(x: int = 1): return x + 1

if __name__ == "__main__":
    deploy(
        entrypoint="./deploy.py:my_flow",
        name="fast-deployment",
        work_pool_name="my-process-pool",
    )

# container entrypoint
prefect worker start --pool my-process-pool

2) Skip a separate deploy step and “serve” the flow inside the container - The container calls

serve(...)

at startup; it registers/updates the deployment and continuously polls for its own work. - This is the simplest/fastest for a single-purpose, always-on service. - Example:

from prefect import flow, serve

@flow
def low_latency_flow(name: str = "world"):
    return f"Hello, {name}"

if __name__ == "__main__":
    serve(low_latency_flow.to_deployment(name="low-latency-deployment"))

Which should you choose? - Want the simplest path and a single dedicated container? Use

serve(...)

(no separate deploy step). - Want standard work-pool/worker orchestration, multiple flows per container, or team workflows via CI? Keep the classic deploy script and run a Process worker in the container. If you share which model you prefer (single served flow vs worker polling a pool) and your environment (Docker/K8s/VM), I can sketch the exact files and commands.

Totally doable. Use a single entrypoint that supports both: - Local dev: run the flow directly (no deployment, fastest feedback) - Container: serve the deployment (low-latency, long‑lived) Here’s a minimal pattern you can copy/paste. app.py

import os
import argparse
from prefect import flow, serve

@flow(log_prints=True)
def my_flow(name: str = "world"):
    print(f"Hello, {name}!")
    return f"Hello, {name}!"

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--name", default="world", help="Flow parameter for local runs")
    parser.add_argument("--serve", action="store_true", help="Start serving the deployment")
    parser.add_argument("--deployment-name", default="low-latency-deployment", help="Deployment name when serving")
    args = parser.parse_args()

    # You can toggle serving with either --serve or PREFECT_SERVE=1
    if args.serve or os.getenv("PREFECT_SERVE") == "1":
        deployment = my_flow.to_deployment(name=args.deployment_name)
        # Optionally add webserver=True for a health endpoint
        serve(deployment)
    else:
        # Local dev run: just call the flow directly
        my_flow(name=args.name)

if __name__ == "__main__":
    main()

Local dev - Fastest path, no deployment/build needed:

python app.py --name "Marvin"

Container (long‑lived, low‑latency serving) - Dockerfile sketch:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY app.py .
ENV PREFECT_API_URL=...      # set for Cloud/Server
ENV PREFECT_API_KEY=...      # set for Cloud
CMD ["python", "app.py", "--serve"]
# or: ENV PREFECT_SERVE=1 and just `CMD ["python", "app.py"]`

Notes - This pattern keeps local dev frictionless while giving you a long‑running container that serves the flow as a deployment for fast triggers. - If you prefer classic deployments + workers instead of

serve(...)

, you can keep the exact same main block: use

python app.py

for local dev, and have CI register the deployment with

prefect deploy

while your container runs a Process worker. But for the simplest/fastest single‑purpose service,

serve(...)

in the container is great. If you want, share your runtime (Docker/K8s/VM) and I’ll tailor the CMD/health checks and any extras (e.g.,

webserver=True

for probes).

You’ve got two clean ways to pass parameters to a deployed flow, and they both work with the “serve in container, run locally” pattern we set up. 1) From code using run_deployment (recommended) - Keeps each invocation tracked as its own flow run. - Pass a dict of parameters that matches your flow signature.

from prefect.deployments import run_deployment

# The deployment name is "<flow_name>/<deployment_name>"
state = run_deployment(
    name="my_flow/low-latency-deployment",
    parameters={"name": "Marvin", "count": 3},  # match your flow args
    flow_run_name="ad-hoc-run-1",               # optional
    tags=["api"],                                # optional
)

2) From the CLI (helpful for quick tests) - Same parameter dict as JSON.

prefect deployment run "my_flow/low-latency-deployment" --params '{"name": "Marvin", "count": 3}'

Serving with default parameters - If you often need defaults, set them when you create the deployment:

from prefect import flow, serve

@flow
def my_flow(name: str, count: int = 1):
    ...

if __name__ == "__main__":
    serve(
        my_flow.to_deployment(
            name="low-latency-deployment",
            parameters={"name": "default-name", "count": 1},  # deployment defaults
        )
    )

Local development (no deployment) - Just run your script and pass args directly to the flow:

python app.py --name "Local Dev"  # from the earlier pattern

Tips - Parameter names must match your flow function’s signature; Prefect 3.x enforces schemas by default (

enforce_parameter_schema=True

in

to_deployment

). - You can also include

flow_run_name

,

tags

, or

scheduled_time

in

run_deployment

for more control. If you share your flow signature, I’ll show the exact

run_deployment

and CLI command you’d use.

Great question. Your client has three reliable ways to pass parameters to the served flow (deployment) running in your container: Option 1: Python SDK (easiest if your client is Python) - Use

run_deployment

with a parameters dict. This creates a run for your served deployment and passes parameters through.

from prefect.deployments import run_deployment

state = run_deployment(
    name="my_flow/low-latency-deployment",
    parameters={"name": "Marvin", "count": 3},
    flow_run_name="api-triggered",
    tags=["from-client"]
)

Option 2: Direct REST API call (any language) - POST to the Prefect API endpoint for your deployment. Include params in JSON and your API key in the header.

curl -X POST \
  "$PREFECT_API_URL/api/deployments/<DEPLOYMENT_ID>/create_flow_run" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "parameters": { "name": "Marvin", "count": 3 },
    "name": "api-triggered",
    "tags": ["from-client"]
  }'

- You can get the deployment ID via

prefect deployment ls

or the UI. Option 3: Automations/events with parameter templating (push-style triggers) - Create an Automation with a RunDeployment action and use Jinja templates to map event payload fields into parameters. - Example parameter mapping inside the automation action:

parameters={
  "user_id": "{{ event.payload['user_id'] }}",
  "timestamp": "{{ event.occurred }}"
}

- Then your client just emits an event; the automation fills in parameters and triggers the served deployment. Notes and tips - Parameter names must match your flow signature; Prefect 3.x enforces schemas by default. If needed, you can set defaults at deployment creation:

serve(
  my_flow.to_deployment(
    name="low-latency-deployment",
    parameters={"name": "default", "count": 1}
  )
)

- Auth: clients must include

Authorization: Bearer <PREFECT_API_KEY>

when using the REST API. The Python SDK picks this up from

PREFECT_API_KEY

automatically. - If you prefer fire-and-forget behavior, set

timeout=0

in

run_deployment

to return immediately after submitting the run. If you tell me what language your client is in, I can provide a ready-to-run snippet for that client calling your served deployment with parameters.

Yes. Your

serve(my_flow.to_deployment(...))

pattern still lets clients pass parameters at run time. How it works -

to_deployment(...)

can define default parameters, but any parameters you send when you trigger a run (via SDK/CLI/REST) will be merged and override those defaults. - Your deployment name will be

my_flow/<deployment-name>

, so in your example:

my_flow/low-latency-deployment

. Examples - From Python (recommended):

from prefect.deployments import run_deployment

state = run_deployment(
    name="my_flow/low-latency-deployment",
    parameters={"name": "Marvin"}  # overrides any defaults
)

- From the CLI:

prefect deployment run "my_flow/low-latency-deployment" --params '{"name": "Marvin"}'

- From the REST API:

curl -X POST "$PREFECT_API_URL/api/deployments/<DEPLOYMENT_ID>/create_flow_run" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "parameters": { "name": "Marvin" },
    "name": "api-triggered"
  }'

Optional: set deployment defaults when serving (still override-able at run time):

serve(
  my_flow.to_deployment(
    name="low-latency-deployment",
    parameters={"name": "default"}  # default only
  )
)

Local dev remains the same:

python app.py --name "Local Dev"

You’re not crazy—“serving” can feel different if you’re used to classic worker-based deployments. Here’s how orchestration and admin work with a served deployment in Prefect 3.x: What happens when the container starts and calls serve(...) - Your code calls

serve(my_flow.to_deployment(name="..."))

. - On startup, Prefect: - Registers the deployment with the Prefect API if it doesn’t exist; otherwise updates it (idempotent by name

flow_name/deployment_name

). - Starts a long-lived runner that polls the API for runs of that deployment and executes them in the same container. - There is no separate “register” step required—starting the container is the registration. How do you point it at the right Prefect environment? - Set these env vars in the container so it registers with the correct backend: -

PREFECT_API_URL

(Cloud: https://api.prefect.cloud/api or your self-hosted URL) -

PREFECT_API_KEY

(for Cloud) or relevant auth for self-hosted - Then just run the container. On first boot, the deployment appears in your Prefect environment. Typical admin workflow with served deployments - Start/stop: start/stop the container (the deployment remains in the API). - Update config: change code or serve() args and restart the container; it updates the deployment. - List/inspect: -

prefect deployment ls

-

prefect deployment inspect "my_flow/my-deployment"

- Schedules: - Add/change via code (serve with interval/cron/rrule) and restart, or manage via CLI: -

prefect deployment schedule create "my_flow/my-deployment" --cron "0 * * * *"

-

prefect deployment schedule ls|pause|resume|clear ...

- Delete: -

prefect deployment delete "my_flow/my-deployment"

How clients trigger it (and pass parameters) - Python SDK: -

run_deployment(name="my_flow/my-deployment", parameters={...})

- CLI: -

prefect deployment run "my_flow/my-deployment" --params '{"key": "value"}'

- REST: -

POST /deployments/<id>/create_flow_run

with JSON

{"parameters": {...}}

and

Authorization: Bearer <API_KEY>

Promoting across environments (dev/staging/prod) - Build and push your image via CI/CD. - In each environment, run the same image with environment-specific `PREFECT_API_URL`/`PREFECT_API_KEY`. - The first time the container starts in that environment, it registers/updates the deployment there. Promotion is just “start the container with the right API settings.” When to prefer pre-registration instead - If you want the “classic” CI step that registers deployments without running containers, or you want dynamic infra (Docker/K8s workers), use: -

flow.from_source(...).deploy(...)

(or

prefect deploy

) to register - Then run workers in your environment to pick up flow runs - That model separates registration from execution; the serve model combines them. Links to dig deeper - Deployments concept - Serve flows in Docker - Create schedules - Manage schedules If you share where this is running (Docker/K8s/VM) I can give you a minimal set of commands/manifests to stand it up and verify the deployment shows up in your Prefect environment.

Yes—you can administer pause/resume from Prefect, even when the flow is “served” in Docker or Kubernetes. Think of it in two layers: - Container/runner: your long‑lived process that polls for work - Orchestration: Prefect’s API decides whether there is any work to give it Ways to pause execution 1) Pause the whole deployment (recommended for quick stop/start) - Stops new auto-scheduled runs for that deployment. Already-running flows continue. Manual triggers via run_deployment still work. - CLI:

prefect deployment pause "my_flow/my-deployment"
prefect deployment resume "my_flow/my-deployment"

- Python:

from prefect.client.orchestration import get_client

async with get_client() as client:
    d = await client.read_deployment_by_name("my_flow/my-deployment")
    await client.pause_deployment(d.id)
    # later...
    await client.resume_deployment(d.id)

2) Pause specific schedules (granular) - Only affects selected schedules on that deployment. - CLI:

prefect deployment schedule ls "my_flow/my-deployment"
prefect deployment schedule pause "my_flow/my-deployment" <SCHEDULE_ID>
prefect deployment schedule resume "my_flow/my-deployment" <SCHEDULE_ID>

3) Concurrency limits (throttle, don’t fully pause) - Limit concurrent runs for a deployment or work queue. Extra runs will queue (or cancel if configured). - Useful to “slow down” without pausing. 4) Pause on shutdown behavior for served deployments - By default,

serve()

pauses deployments when the container stops (so the scheduler won’t create new runs while your runner is down). - You can control this when serving:

serve(
  my_flow.to_deployment(
    name="my-deployment",
    # default is True; set to False if you do NOT want auto-pause on exit
    # pause_on_shutdown=False
  )
)

Operational notes - Stopping the container stops execution immediately, but the deployment remains registered. If you don’t pause it, auto-scheduled runs may queue up while the runner is down. - Pausing a deployment deletes its future auto-scheduled runs that are still in Scheduled; already-running runs are not interrupted. - You can still trigger runs manually (e.g., for an emergency) even when the deployment is paused. Admin cheatsheet - List deployments:

prefect deployment ls

- Inspect deployment:

prefect deployment inspect "my_flow/my-deployment"

- Manage schedules:

prefect deployment schedule ls|create|pause|resume|clear "my_flow/my-deployment"

If you let me know whether you’ll run this in raw Docker or K8s, I can suggest health checks and a clean “pause during rollout” workflow for that platform.

You don’t have to cram everything into a CLI call. In Prefect 3.x there are clean, standard ways to inject config and secrets for both Docker and Kubernetes, whether you “serve” in a long‑lived container or use workers. Pick the right tool for the job: - OS environment variables (baked into containers or set at runtime) - Prefect Variables (non‑secret config managed in Prefect) - Prefect Secret/Block storage (encrypted secrets managed in Prefect) - Kubernetes Secrets/ConfigMaps (for K8s workloads) - job_variables overrides (per‑deployment or per‑run env overrides) Common patterns A) Long‑lived container with serve(...) - Put non-secret config in env vars in your Dockerfile or runtime:

# Dockerfile
ENV LOG_LEVEL=INFO
# or at run time
docker run -e LOG_LEVEL=INFO -e PREFECT_API_URL=... -e PREFECT_API_KEY=... my-image

- Read in your flow with

os.environ[...]

. - For secrets, prefer Prefect Secret blocks or K8s Secrets (see below). B) Kubernetes (served or worker-based) - Use standard K8s Secrets/ConfigMaps:

envFrom:
  - secretRef:
      name: my-app-secrets
  - configMapRef:
      name: my-app-config

- Or

valueFrom.secretKeyRef

for specific keys. - Your flow reads them like any env var. C) Prefect-managed config - Variables for non-sensitive values:

from prefect.variables import Variable
env = Variable.get("environment", default="prod")

Set them once:

prefect variable set environment prod

- Secret blocks for sensitive data (encrypted at rest):

from prefect.blocks.system import Secret

# One-time setup (in CI or admin step)
Secret(value="super-secret-token").save("external-api-token", overwrite=True)

# In your flow
api_token = Secret.load("external-api-token").get()

D) Per-deployment or per-run overrides with job_variables - If you deploy to a work pool (Docker/K8s), you can set env vars on the deployment and even override them at run time:

# At deploy time
my_flow.deploy(
  name="prod",
  work_pool_name="kubernetes",
  job_variables={"env": {"LOG_LEVEL": "DEBUG", "FEATURE_FLAG": "1"}}
)

# At run time (Python)
from prefect.deployments import run_deployment
run_deployment(
  name="my_flow/prod",
  job_variables={"env": {"LOG_LEVEL": "WARNING"}}
)

# Or CLI
prefect deployment run "my_flow/prod" --job-variable env.LOG_LEVEL=WARNING

Note: job_variables apply to worker-based infra. For “serve” inside a long-lived container, prefer container/K8s env injection and Prefect Variables/Secret blocks. What to use when - Serve in Docker/K8s and keep it simple: - Base config via container/K8s env vars - Secrets via Prefect Secret blocks (Cloud) or K8s Secrets (K8s) - Optional: non-secret Config via Prefect Variables - Workers (Docker/K8s pools): - Use job_variables for per-deployment/per-run env - Still use Secret blocks or K8s Secrets for sensitive values Minimal, secure example with serve(...) - Docker - Build image without secrets - Inject Prefect API creds and app config at runtime:

docker run -d \
  -e PREFECT_API_URL=... \
  -e PREFECT_API_KEY=... \
  -e DB_HOST=db.local \
  my-image:prod

- Store API tokens/passwords in Prefect Secret blocks; load them in your flow:

from prefect.blocks.system import Secret

token = Secret.load("external-api-token").get()

- Kubernetes - Create K8s Secret and reference it:

kubectl create secret generic app-secrets \
  --from-literal=DB_PASSWORD=... \
  --from-literal=API_TOKEN=...

- Deployment/pod spec:

envFrom:
  - secretRef:
      name: app-secrets

Key Prefect settings you’ll likely set as env vars -

PREFECT_API_URL

and

PREFECT_API_KEY

to point your container at the right Prefect workspace - Optional logging or runner settings, e.g.

PREFECT_LOGGING_LEVEL

,

PREFECT_RUNNER_SERVER_ENABLE=true

Links - Blocks & Secrets - Variables

- Customize job variables - Serve flows in Docker - Kubernetes deployments If you share where you’re running (pure Docker vs K8s) and what secrets/config you have, I can sketch the exact Dockerfile/K8s manifest and Prefect block creation commands.