Great question — you can absolutely keep a Coiled Dask cluster alive and reuse it across Prefect flow runs. There are two solid patterns you can use in Prefect 3.x: 1) Run a long-lived Coiled cluster “out of band” and connect to it by address - Start the cluster once (manually or with a small bootstrap script) and keep it alive:

import coiled

cluster = coiled.Cluster(
    name="prefect-prod",
    shutdown_on_close=False,  # don't tear down the cluster when Python exits
    idle_timeout=None,        # or set a long timeout like "4h" to avoid idle shutdowns
    # software="...", n_workers=..., region=..., etc. as needed
)
print("Scheduler:", cluster.scheduler_address)

- Store the scheduler address somewhere your flow can read it (e.g., Prefect String block or an env var), then point your flow’s task runner at it:

from prefect import flow
from prefect.blocks.system import String
from prefect_dask.task_runners import DaskTaskRunner

SCHEDULER = String.load("coiled-scheduler").value  # store once via UI or code

@flow(task_runner=DaskTaskRunner(address=SCHEDULER))
def my_flow():
    # tasks here will execute on the long-lived Coiled cluster
    ...

Notes: - This is the simplest way to ensure runs reuse the same cluster and avoid the spin-up cost. - Depending on your Coiled setup, connecting by raw address may require the same TLS/security config present at runtime. If you want Coiled to handle that automatically, use option 2 below. 2) Let Prefect attach/create a named Coiled cluster per run, but keep it running Provide a callable to

DaskTaskRunner.cluster_class

that “attaches if exists, else creates” and make sure the cluster isn’t shut down on close:

import coiled
from prefect import flow
from prefect_dask.task_runners import DaskTaskRunner

CLUSTER_NAME = "prefect-prod"

def coiled_attach_or_create():
    try:
        # Attach to an existing cluster by name
        return coiled.Cluster.from_name(CLUSTER_NAME)
    except Exception:
        # Create it if it doesn't exist, and keep it alive after the run
        return coiled.Cluster(
            name=CLUSTER_NAME,
            shutdown_on_close=False,
            idle_timeout=None,   # or a long value like "4h"
            # software="...", n_workers=..., region=..., etc.
        )

@flow(task_runner=DaskTaskRunner(
    cluster_class=coiled_attach_or_create,
    # Optional: have Prefect auto-adapt the cluster if desired
    # adapt_kwargs={"minimum": 0, "maximum": 50}
))
def my_flow():
    ...

What happens here: - Prefect will close the Python handle at the end of the run, but because

shutdown_on_close=False

, the Coiled cluster keeps running and is reused on subsequent runs. - Using Coiled’s Python API inside the flow also ensures the correct security config is applied automatically. A few practical tips - Cost control: If you keep the cluster alive, consider autoscaling or a non-zero

idle_timeout

that fits your schedule. A common pattern is

idle_timeout="4h"

during working hours. - Reliability: Give the cluster a unique, stable

name

(e.g., per env:

prefect-prod

,

prefect-staging

) and use

from_name

to attach. - Secrets/config: If you connect by address, store it in a Prefect Block or env var so you’re not hardcoding it in your flow code. References - Prefect + Dask integration - How-to: Use Coiled with Prefect - Coiled provisioner API reference Questions to tailor this for you: - Do you prefer the “always-on, connect-by-address” model or the “attach-or-create by name” approach? - Do you have an autoscaling target or a desired idle policy (e.g., keep alive during business hours)?