Great question — you can do this cleanly in Prefect 3.x with a Docker worker by using a human-in-the-loop “form” and suspending the flow so the container exits while waiting. That way, you don’t consume any worker or Docker resources during the approval wait. High level - pause_flow_run(): keeps the container running (uses resources) — good for short waits. - suspend_flow_run(): exits the container and frees the worker slot — best for approvals and long waits. Since you asked to avoid wasting resources, use suspend_flow_run with a RunInput form. Example: human approval form that frees resources

python
from prefect import flow
from prefect.logging import get_run_logger
from prefect.flow_runs import suspend_flow_run
from prefect.input import RunInput
from typing import Literal

class ApprovalForm(RunInput):
    action: Literal["approve", "reject"]
    approver: str
    comments: str | None = None

@flow
def deploy_with_approval(app: str, version: str):
    logger = get_run_logger()

    # Suspend the run here: Docker container exits; worker slot is freed
    result = suspend_flow_run(
        wait_for_input=ApprovalForm.with_initial_data(
            description=(
                f"Approval required to deploy\n\n"
                f"- App: {app}\n"
                f"- Version: {version}\n\n"
                f"Choose approve or reject and provide your name/comments."
            )
        ),
        timeout=24 * 3600,  # auto-fails if not resumed in 24 hours
        key="deploy-approval-gate",  # idempotency if the flow retries
    )

    # When resumed, a fresh container is started and execution continues here
    if result.action == "reject":
        raise RuntimeError(f"Rejected by {result.approver}: {result.comments or ''}")

    <http://logger.info|logger.info>(f"Approved by {result.approver}. Proceeding with deployment...")
    # ... continue your deployment logic ...

How to resume the run (UI, Python, or API) - Prefect UI: Open the paused run and click Resume, fill the form. - Python client:

python
from prefect.client.orchestration import get_client

async def approve(run_id: str):
    async with get_client() as client:
        await client.resume_flow_run(
            flow_run_id=run_id,
            run_input={
                "action": "approve",
                "approver": "<mailto:alice@example.com|alice@example.com>",
                "comments": "Ship it",
            },
        )

- HTTP API:

bash
curl -X POST "$PREFECT_API_URL/api/flow_runs/<flow-run-id>/resume" \
  -H "Content-Type: application/json" \
  -d '{"run_input": {"action": "approve", "approver": "<mailto:alice@example.com|alice@example.com>", "comments": "Ship it"}}'

Docker worker setup (so suspend truly frees resources) 1) Create a Docker work pool:

bash
prefect work-pool create "docker-pool" --type docker

2) Start a Docker worker pointing to that pool:

bash
prefect worker start --pool docker-pool

3) Ensure the worker environment has Docker support installed:

bash
pip install "prefect[docker]"

Notes and tips - Run from a deployment: suspend_flow_run reschedules the run; your flow should be executed via a deployment associated with a Docker work pool so a new container can be launched on resume. - Timeouts: set timeout to enforce a maximum wait; on expiry the run will fail automatically. - Idempotency: use key=... so the same approval gate doesn’t trigger multiple times if your flow retries. - Short waits: if you truly need to keep the container/process alive (e.g., for fast “approve now” UX), use pause_flow_run(wait_for_input=...) — but it will keep the Docker container running until resumed. Useful docs - Flow run pause/suspend API reference: prefect.flow_runs (pause/suspend) - RunInput forms API reference: prefect.input (RunInput) - Work pools and workers concepts: Work pools and workers

- Docker worker collection: prefect-docker collection Want me to tailor this for a specific approval channel (e.g., Slack button → resume via webhook), or show a minimal deployment file for your Docker pool?

Short answer - Use suspend_flow_run(wait_for_input=...) to show an approval form and free the Docker container/worker slot. - Resume the run from the UI, via the Python client, or the HTTP API. - After resume, the flow continues from the line after suspend_flow_run — that call returns the provided input. Because a new container is started, any in‑memory variables from before the suspend are gone, so persist anything you need across the boundary. How to resume - Prefect UI: open the SUSPENDED run, click Resume, fill the form. - Python client:

python
from prefect.client.orchestration import get_client

async def approve(run_id: str):
    async with get_client() as client:
        await client.resume_flow_run(
            flow_run_id=run_id,
            run_input={"approved": True, "approver": "<mailto:alice@example.com|alice@example.com>"}
        )

- HTTP API:

bash
curl -X POST "$PREFECT_API_URL/api/flow_runs/<flow-run-id>/resume" \
  -H "Content-Type: application/json" \
  -d '{"run_input": {"approved": true, "approver": "<mailto:alice@example.com|alice@example.com>"}}'

Example pattern

python
from prefect import flow
from prefect.flow_runs import suspend_flow_run
from prefect.input import RunInput
from typing import Literal

class Approval(RunInput):
    decision: Literal["approve", "reject"]
    approver: str
    notes: str | None = None

@flow
def gated_flow():
    # Do work here and persist anything you'll need after resume (e.g., to a block or as task results)

    result = suspend_flow_run(
        wait_for_input=Approval.with_initial_data(
            description="Approve or reject the next step."
        ),
        timeout=24 * 3600,
        key="human-approval-gate"
    )

    # New container starts on resume; code continues here
    if result.decision == "reject":
        raise RuntimeError(f"Rejected by {result.approver}: {result.notes or ''}")

    # Proceed with the rest of the flow

Important details - The Docker container exits while suspended; no resources are used. On resume, the worker launches a new container that picks up after the suspend call. - Persist any state you’ll need after resume (task results with persistence, block storage, or include in the form), since local variables won’t survive across containers. - There’s no CLI “resume” command; use UI, Python client, or the HTTP API. References - Flow pause/suspend API: prefect.flow_runs - RunInput forms: prefect.input - Workers and work pools: Work pools and workers - Docker worker collection: prefect-docker