Short answer: there isn’t a built‑in “per‑worker, per‑task” concurrency limit in Prefect 3.x today. Task concurrency limits are enforced via tags (or via global concurrency limits) and are global across all workers. The

worker --limit

flag is flow-run concurrency per worker, not task-level. What exists today - Tag-based task limits (global): Docs - Global concurrency limits: Docs - CLI to manage tag limits (verified):

prefect concurrency-limit create <tag> <limit>
  prefect concurrency-limit ls
  prefect concurrency-limit inspect <tag>
  prefect concurrency-limit delete <tag>

Open issues / discussions - I couldn’t find an open issue/discussion specifically requesting “per-worker task concurrency.” If you’d like, I can open a feature request on your behalf, or you can use New issue. Practical workarounds (no local semaphore required) 1) Per-worker tags + tag limits - Give each worker a unique identifier (env var) and add a worker-specific tag to the task at submit time. Then create a concurrency limit for each worker’s tag. - Example:

# Set per-worker ID via environment on each worker, e.g.
  # PREFECT_WORKER_ID=worker-a

  from prefect import flow, task
  from prefect.context import tags
  import os

  @task
  def critical_op(i: int):
      ...

  @flow
  def main():
      worker_id = os.getenv("PREFECT_WORKER_ID", "default")
      # Add a worker-scoped tag
      with tags(f"db-crit:{worker_id}"):
          for i in range(10):
              critical_op.submit(i)

Then set limits per worker:

prefect concurrency-limit create db-crit:worker-a 2
  prefect concurrency-limit create db-crit:worker-b 2

Notes: - You can also do

critical_op.with_options(tags=[f"db-crit:{worker_id}"]).submit(...)

. - How to set the env var depends on worker type: - Process worker: set it in the shell/systemd before

prefect worker start

. - Kubernetes worker: add an env var in the work pool’s base job template. See work pools: Docs. 2) Global Concurrency Limits with a worker-specific key (still Prefect-managed, not a “local” semaphore) - If you want the limiting inside the task body while still using Prefect’s central leases:

import os
  from prefect import task
  from prefect.concurrency.sync import concurrency

  @task
  def critical_op():
      key = f"db-crit:{os.getenv('PREFECT_WORKER_ID', 'default')}"
      with concurrency(key, occupy=1):
          # do restricted work
          ...

Then create per-worker limits for those keys:

prefect concurrency-limit create db-crit:worker-a 2
  prefect concurrency-limit create db-crit:worker-b 2

Caveats - Both approaches still use Prefect’s orchestration for limits, but the scoping is achieved by using different tag/limit names per worker. There’s no native “scope to this worker instance” flag on concurrency limits today. If you share which worker type you’re on (process/Kubernetes/ECS), I can give you a ready-to-paste snippet to inject the worker ID env var. Also, happy to file a feature request for true per‑worker task limits if you’d like.

Totally agree: use Global Concurrency Limits (GCLs) for this. They’re the first‑class mechanism in Prefect 3.x and not a workaround. For per‑worker, per‑task limits, the recommended pattern is: - Use a GCL key that includes the worker’s identity (so each worker gets its own independent limit) - Acquire the lease inside the task using Prefect’s server‑backed concurrency context manager (this is not a local semaphore) Example

# Set on each worker (env var or job template)
# PREFECT_WORKER_ID=worker-a | worker-b | ...

from prefect import flow, task
from prefect.concurrency.sync import concurrency
import os, time

@task
def my_task(i: int):
    key = f"my-task:{os.getenv('PREFECT_WORKER_ID', 'default')}"
    # Server-backed lease; scoped by worker via key naming
    with concurrency(key, occupy=1):
        time.sleep(2)
        return i

@flow
def run():
    for i in range(10):
        my_task.submit(i)

Then create independent limits per worker:

prefect gcl create my-task:worker-a 2
prefect gcl create my-task:worker-b 2
prefect gcl ls
prefect gcl inspect my-task:worker-a

Notes - This is fully orchestrated by Prefect (leases stored/managed server‑side), so you’re not implementing a local semaphore. - If you can’t inject a worker ID as an env var, tell me your worker type (process/k8s/ECS) and I’ll share a snippet to set it in your pool template. Docs - Global concurrency limits: https://docs.prefect.io/v3/concepts/global-concurrency-limits - CLI (verified):

prefect gcl --help

If you need a zero‑code alternative, tag-based limits still exist but are mainly for compatibility; GCLs are preferred going forward.

i guess i didn't read as closely into "per worker" in your original ask as i could have

maybe im also confused by this

but at a task level

is this prefect "Task" or something else

map of task def -> limiter instances

ah yea i need to read better

something that hasn't really got enough use-case to justify the added complexity / effort

i think short answer, yes

Appreciate its a fairly niche case, our other flows are much more sensible and use ECS push pools with global limits etc but unfortunately this one needs to use windows ;(

this now exists in the real ™️ docs