Great question. In Prefect 3.x, you choose the task runner for a flow at invocation time; you can’t change it mid-run. There are a few good ways to set

max_workers

dynamically per run: Option 1 — Set per run with with_options (recommended) - Compute the number of workers at call time and pass a task runner instance.

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner
import os

@flow
def my_flow():
    # your tasks here
    ...

def run_flow(workers: int):
    my_flow.with_options(
        task_runner=ThreadPoolTaskRunner(max_workers=workers)
    )()

if __name__ == "__main__":
    workers = int(os.getenv("WORKERS", "8"))
    run_flow(workers)

- You can use the same pattern for subflows to give them different runners. Option 2 — Use an environment variable (no code change) - Prefect will default

ThreadPoolTaskRunner.max_workers

from the env var if you don’t set it explicitly: -

PREFECT_TASK_RUNNER_THREAD_POOL_MAX_WORKERS

Shell example:

export PREFECT_TASK_RUNNER_THREAD_POOL_MAX_WORKERS=16
python my_flow.py

In a deployment (

prefect.yaml

) via job variables:

deployments:
  - name: my-deployment
    entrypoint: flows/my_flow.py:my_flow
    job_variables:
      env:
        PREFECT_TASK_RUNNER_THREAD_POOL_MAX_WORKERS: "16"

Option 3 — Use temporary_settings around the call - Handy when you want to adjust the setting programmatically before a specific run.

from prefect import flow
from prefect.settings import temporary_settings, PREFECT_TASK_RUNNER_THREAD_POOL_MAX_WORKERS

@flow
def my_flow():
    ...

with temporary_settings({PREFECT_TASK_RUNNER_THREAD_POOL_MAX_WORKERS: 32}):
    my_flow()

Notes and gotchas - The task runner is created when the flow starts. Set the value before you call the flow (or use

with_options

)—changing settings inside a running flow won’t reconfigure its runner. - Passing

ThreadPoolTaskRunner(max_workers=...)

directly to

@flow(task_runner=...)

hard-codes it at import time, so prefer

with_options

if you need per-run control. References - Task runners (concepts) - Settings reference - ThreadPoolTaskRunner source If you’re aiming to set

max_workers

based on a flow parameter (e.g.,

my_flow(workers=...)

), I can show a small pattern that wires the parameter into

with_options

cleanly. Would you like that?