Short answer: not directly. In Prefect 3.x, the task runner is chosen and configured when the flow is defined (import time), so flow/deployment parameters (which are only available at runtime) can’t change

max_workers

. Recommended ways to vary

max_workers

per environment/deployment: - Use a Prefect setting/env var and don’t hardcode

max_workers

in code - In your flow, remove the explicit value:

from prefect import flow
    from prefect.task_runners import ThreadPoolTaskRunner

    @flow(task_runner=ThreadPoolTaskRunner())  # defer to settings
    def flow_snowflake2():
        print("Run flow")

- Then set one of these before the flow is imported (e.g., in your worker/job env): -

PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS=2

(canonical) -

PREFECT_TASK_RUNNER_THREAD_POOL_MAX_WORKERS=2

(alias) - You can persist it in your active Prefect profile as well:

prefect config set tasks.runner.thread_pool_max_workers=2

- For deployments, set the env var in: - Your work pool’s default job variables (env) - The deployment’s

job_variables

(env) - Your container/VM environment - Create flow variants with different runners at definition time (if you prefer code-level control)

from prefect import flow
  from prefect.task_runners import ThreadPoolTaskRunner

  @flow
  def flow_snowflake2():
      print("Run flow")

  flow_snowflake2_small = flow_snowflake2.with_options(
      task_runner=ThreadPoolTaskRunner(max_workers=2)
  )
  flow_snowflake2_large = flow_snowflake2.with_options(
      task_runner=ThreadPoolTaskRunner(max_workers=16)
  )
  # Create separate deployments pointing at each variant

- Factory that reads env at import time (handy if you need logic)

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

  def make_runner():
      return ThreadPoolTaskRunner(
          max_workers=int(os.getenv("MAX_WORKERS", "2"))
      )

  @flow(task_runner=make_runner())
  def flow_snowflake2():
      print("Run flow")

Key points: - Passing

max_workers

in the decorator overrides settings/env. - To control via env/settings, leave

max_workers

unset in code. - Flow/deployment parameters cannot affect task runner configuration because the runner is instantiated before parameters are available. If you can share how you’re running deployments (Docker, Kubernetes, local process worker, etc.), I can show exactly where to set the env var for your setup.

Great question. In Prefect 3.x you can set a deployment’s job_variables in three main ways. These determine the infrastructure config the worker will use for that deployment (e.g., env vars, image, working_dir, etc.). 1) Python API (Flow.deploy) - Pass a dict to the job_variables parameter:

from prefect import flow

@flow
def my_flow():
    ...

if __name__ == "__main__":
    my_flow.deploy(
        name="my-deployment",
        work_pool_name="my-pool",
        job_variables={
            "env": {
                "PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS": "2",
            },
            # other variables based on your pool type, e.g. image/working_dir/etc.
        },
    )

2) CLI (prefect deploy) - Use one or more --job-variable (-jv) flags. You can set nested keys with dot notation, or pass a JSON object:

# Simple key=value
prefect deploy \
  --name my-deployment \
  --pool my-pool \
  --job-variable env.PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS=2

# Or pass a JSON object
prefect deploy \
  --name my-deployment \
  --pool my-pool \
  --job-variable '{"env": {"PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS": "2"}}'

3) prefect.yaml (version-controlled) - Define job_variables under the deployment’s work_pool section:

deployments:
- name: my-deployment
  entrypoint: flows.py:my_flow
  work_pool:
    name: my-pool
    job_variables:
      env:
        PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS: "2"
      # add other variables supported by your pool type

Helpful tips - What keys can I set? It depends on your work pool type. Inspect the base job template to see supported keys:

prefect work-pool get-default-base-job-template --type process
prefect work-pool get-default-base-job-template --type docker
prefect work-pool get-default-base-job-template --type kubernetes

- Precedence (lowest → highest): - Work pool base job template defaults - Deployment job_variables (set as above) - Per-run overrides (prefect deployment run ... --job-variable ...) - For your specific need (ThreadPoolTaskRunner max_workers), set the env var - Canonical:

PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS

- Alias:

PREFECT_TASK_RUNNER_THREAD_POOL_MAX_WORKERS

- Ensure your flow’s decorator doesn’t hardcode `max_workers`; use

ThreadPoolTaskRunner()

so it reads from env. If you tell me which worker type you’re using (process, docker, kubernetes) I can give a tailored example of the exact job_variables available and a ready-to-paste snippet.