Yes — in Prefect 3.x, the simplest way is to use a task runner with

max_workers=10

. If you have 50 submitted tasks, Prefect will only run 10 at once, and start new ones as earlier ones finish. Example:

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import wait
import time

@task
def do_work(i):
    time.sleep(5)
    return i

@flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
def my_flow():
    futures = [do_work.submit(i) for i in range(50)]
    wait(futures)

if __name__ == "__main__":
    my_flow()

What this does: - 50 tasks are submitted - only 10 execute in parallel - when one finishes, another starts If your tasks are CPU-heavy, use

ProcessPoolTaskRunner

instead of

ThreadPoolTaskRunner

.

from prefect.task_runners import ProcessPoolTaskRunner

@flow(task_runner=ProcessPoolTaskRunner(max_workers=10))
def my_flow():
    futures = [do_work.submit(i) for i in range(50)]
    wait(futures)

When to use this approach - good if you want to limit concurrency within one flow run - especially useful for local execution or a single worker process If you need the limit enforced across multiple flow runs / workers, use a global concurrency limit or tag-based concurrency limit instead. Useful docs: - Run work concurrently - Global concurrency limits If you want, I can also show the cross-worker/global version where 50 tasks may come from multiple flows but only 10 total are allowed to run at once.