Short answer: your tag usage is correct, but the automatic tag-based limit is only enforced when tasks are orchestrated by the Prefect server (e.g., via a deployment/worker or a local run that’s actually connected to your API). If you run the flow “in-process” locally or against the ephemeral API, the server-side tag concurrency rule won’t kick in—so all tasks start immediately. Calling

rate_limit("tag:limited")

works because it explicitly acquires the limit before your code proceeds. A few checks and ways to make it work the way you expect: - How are you running the flow? Locally via

python script.py

or as a deployment run by a worker? If local, are you connected to Prefect Cloud/Server? - Verify your connection/profile:

prefect config view --show-sources

Ensure

PREFECT_API_URL

(and

PREFECT_API_KEY

for Cloud) are set so your local run uses the same API where you created

tag:limited

. - If you want automatic tag-based enforcement without calling

rate_limit

, run the flow as a deployment on a worker connected to the same API. The server will enforce the GCL on task start. - If you prefer to keep running locally/in-process, continue using

rate_limit("tag:limited")

inside the task; it will block until a slot is available and give you the rate-limit semantics you want. Your GCL is named correctly for tags:

prefect global-concurrency-limit create tag:limited --limit 1 --slot-decay-per-second 1

- With

limit=1

and

slot_decay_per_second=1

, you’ll get roughly one task start per second once the limit is saturated. Useful docs: - Global concurrency limits (GCL): Concepts — Global concurrency limits - Tag-based limits: Concepts — Tag-based concurrency limits - How-to guides: How to apply global concurrency limits and How to limit concurrent task runs with tags -

rate_limit

helper: API — prefect.concurrency.sync If you can share how you’re executing the flow (local script vs deployment, and whether your local run is pointed at Cloud/Server), I can suggest the exact steps to get server-enforced tag limiting working in your setup.

Thanks — since you’re running it as a deployment via

prefect deployment run

, the server-side tag-based enforcement is active. What you’re seeing is expected behavior, though. Key distinction: - Tag-based global concurrency limits (GCLs) limit concurrency — how many tasks with that tag can be RUNNING at the same time across your whole environment. - They do not inherently “pace” sequential calls inside a single flow loop. - The

rate_limit("tag:...")

helper does implement a token-bucket rate limit (using slot decay), so it will introduce waits even when tasks are invoked one-after-another in the same process. In your example, the loop calls tasks one after another. With a tag-based GCL at 1, there’s no contention (only one task attempts to run at a time), so the server allows each task to start immediately. That’s why it looks “not limited.” When you add

rate_limit("tag:limited")

, Prefect actively waits for the decayed slot and you get once-per-second pacing. You have two options depending on your goal: - You want strict 1-per-second pacing (even for a single flow loop): keep the explicit helper.

from prefect.concurrency.sync import rate_limit

  @task()
  def limited_task(i: int):
      rate_limit("tag:limited")  # waits for decayed slot
      ...

- You want to cap concurrent API calls across many flows/workers: keep the tag + GCL, and run tasks concurrently so the server has something to throttle. For example:

from prefect import flow, task
  from prefect.task_runners import ConcurrentTaskRunner

  @task(tags=["limited"])
  def limited_task(i: int):
      ...

  @flow(task_runner=ConcurrentTaskRunner())
  def sample_flow():
      futures = [limited_task.submit(i) for i in range(10)]
      for f in futures:
          f.result()

With a GCL

tag:limited

at 1, only one of those tasks will be RUNNING at any time across your deployment(s). Note this still doesn’t enforce “one per second” spacing; it enforces “one at a time”. Helpful checks: - Make sure the GCL exists and is active:

prefect global-concurrency-limit ls

You should see

tag:limited

with

limit=1

and

slot_decay_per_second=1

. Docs for reference: - Concepts — Global concurrency limits: Global concurrency limits - Concepts — Tag-based concurrency limits: Tag-based concurrency limits - How-to — Global limits: How to apply global concurrency limits - How-to — Tag limits: How to limit concurrent task runs with tags - API —

prefect.concurrency.sync

(includes

rate_limit

): prefect.concurrency.sync If your end goal is “exactly one call per second across all flows,” the most reliable approach is to keep the

rate_limit("tag:limited")

in the task or at the API call site. If you want me to help tune it (e.g., async version, handling bursts, etc.), happy to suggest a pattern.

Correct — you don’t need to decorate that function with

@task

if the only Prefect feature you’re using is

rate_limit("tag:limited")

. Details: -

rate_limit()

works fine in plain Python code (no flow/task context required). It talks to the Prefect API to increment a decaying global limit and returns when a slot is available. - Requirements: - Your process must be connected to a Prefect API (Cloud/Server). Ensure

PREFECT_API_URL

(and

PREFECT_API_KEY

for Cloud) are set for your deployment/worker. - The target limit must exist and be configured with

slot_decay_per_second > 0

(rate-limit mode requires decay). - Caveats: -

rate_limit()

does not create/maintain a lease; it increments the decaying counter and returns. There’s no “release” on exit—decay handles pacing. - If you want slots to be held for the duration of a code block and explicitly released, use the

concurrency(...)

context manager instead. So you can simplify to something like:

from prefect.concurrency.sync import rate_limit

def limited_task(i: int):
    rate_limit("tag:limited")  # waits until a decayed slot is available
    # do the work

If later you want retries, logging, caching, or task-level visibility, you can re-add

@task

— but it’s not required for the rate limiting itself.

Great question. By default, both helpers behave like the server’s tag-based limits: they do not fail if the specified limit(s) don’t exist. - Default behavior (strict=False, the default): -

rate_limit("name")

and

concurrency("name")

will no-op if the limit doesn’t exist. - They log a warning and proceed; no exception is raised. - To make them fail when the limit doesn’t exist: - Pass

strict=True

. Both functions support it and will raise

ConcurrencySlotAcquisitionError

if any named limit is missing. - Example:

from prefect.concurrency.sync import rate_limit, concurrency
    from prefect.concurrency.asyncio import ConcurrencySlotAcquisitionError

    try:
        rate_limit("tag:limited", strict=True)  # raises if missing
    except ConcurrencySlotAcquisitionError:
        ...

- Naming note: - Use the exact limit name you created. For tag-based limits, that means including the

tag:

prefix (e.g.,

tag:limited

). There’s no auto-prefixing. - Empty names: - Passing

[]

or

None

is a no-op (returns immediately, no error).

Yes. You can combine a global limit with a local (flow-run-scoped) limit. Common patterns: - Deployment-level concurrency (per-deployment, per-flow-run cap) - Limits how many runs of this deployment can execute concurrently. - Configure on deploy: - CLI: use the deployment’s concurrency flags - Python: pass

concurrency_limit=...

when you call

flow.deploy(...)

or

flow.serve(...)

- Works alongside any global limits you use in tasks. - Local, per-flow-run limit using a unique name - Create a limit name that’s unique to the flow run and acquire it with

concurrency(...)

around the work you want to bound. For example:

from prefect import flow, task, get_run_context
    from prefect.concurrency.sync import concurrency

    @task
    def do_work(i: int):
        ctx = get_run_context()
        flow_run_id = ctx.flow_run.id
        # Local limit name for this run only
        with concurrency(f"flow:{flow_run_id}:local", occupy=1, strict=False):
            # do bounded work
            ...

    @flow
    def my_flow():
        for i in range(10):
            do_work.submit(i)

- For best control, create the GCL at flow start (limit N for this run), and optionally delete it at the end:

from prefect.client.orchestration import get_client
    from prefect.client.schemas.actions import GlobalConcurrencyLimitCreate

    @flow
    async def my_flow():
        flow_run_id = get_run_context().flow_run.id
        name = f"flow:{flow_run_id}:local"
        async with get_client() as client:
            await client.upsert_global_concurrency_limit_by_name(
                GlobalConcurrencyLimitCreate(name=name, limit=3)
            )
        try:
            # submit tasks that use with concurrency(name) ...
            ...
        finally:
            # optional cleanup
            async with get_client() as client:
                await client.delete_global_concurrency_limit_by_name(name)

- This gives you a “local” cap for this flow run while your tasks can also acquire true global limits (e.g.,

database-pool

) at the same time. - Note: if you don’t upsert the limit first,

concurrency(name)

will no-op by default unless you pass

strict=True

. - Keep a separate global limit too - Inside the same task you can acquire multiple limits:

with concurrency(["database-pool", f"flow:{flow_run_id}:local"], occupy=1, strict=True):
        ...

- This enforces both the global cap and the per-run cap simultaneously. - If you need pacing (one per second) within the flow run - Use

rate_limit(f"flow:{flow_run_id}:local")

and set

slot_decay_per_second

on that per-run limit. This gives you a per-run rate limit, while still acquiring a broader global limit elsewhere for cross-flow control. Which to choose? - If you only need “how many flow runs at once”: deployment concurrency is simplest. - If you need “how many tasks for this run at once” (or per-run rate), use a per-run GCL name and acquire it with `concurrency(...)`/`rate_limit(...)` in your tasks. - Combine both if you want run-level caps and cross-flow/global caps together. If you share your exact target (e.g., “max 3 tasks from this flow run at once, max 20 across all flows, and 1 per second pacing”), I can draft the minimal code/config to achieve it.

Short answer: - You don’t have to create a per-run global limit if you just want a local (per-flow-run) cap. The simplest “local limit” is to set the flow’s task runner concurrency, e.g.

ConcurrentTaskRunner(max_workers=N)

. That caps how many tasks run concurrently within that flow run and requires no GCLs or cleanup. - If you specifically want a server-enforced per-run limit that tasks must acquire (e.g., combined with global limits or per-run rate limiting across multiple threads), then yes—create a GCL with a unique name per flow run. You can create it via the Python client at the start of the flow and delete it at the end. There’s no built-in auto-TTL; you own cleanup. - You can attach cleanup to flow hooks (on_completion/on_failure/on_cancellation/on_crashed) and/or use a try/finally in your flow body. Options in practice: 1) Local per-run cap without GCLs (recommended when you only need per-flow-run concurrency)

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

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

@flow(task_runner=ConcurrentTaskRunner(max_workers=3))
def my_flow():
    futures = [do_work.submit(i) for i in range(10)]
    for f in futures:
        f.result()

- This yields “max 3 tasks at once in this flow run,” no global objects created. 2) Per-run GCL you create and delete (for server-enforced per-run cap or per-run rate-limit)

from prefect import flow, task, get_run_context
from prefect.concurrency.sync import concurrency, rate_limit
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import GlobalConcurrencyLimitCreate

@task
def do_work(i: int, local_limit_name: str):
    # Combine with global limit if needed:
    with concurrency(["global-api", local_limit_name], strict=True):
        # or for per-run rate: rate_limit(local_limit_name)
        ...

@flow
async def my_flow():
    flow_run_id = get_run_context().flow_run.id
    local_limit_name = f"flow:{flow_run_id}:local"
    # Create or ensure the limit exists
    async with get_client() as client:
        await client.create_global_concurrency_limit(
            GlobalConcurrencyLimitCreate(name=local_limit_name, limit=3)
        )

    try:
        for i in range(10):
            do_work.submit(i, local_limit_name)
    finally:
        # Cleanup
        async with get_client() as client:
            await client.delete_global_concurrency_limit_by_name(local_limit_name)

- If you want per-run rate limiting, create with

slot_decay_per_second

and use

rate_limit(local_limit_name)

in the task. 3) Hook-based cleanup (in addition to or instead of try/finally) - Flows support hooks you can use to delete the limit even if the flow fails/cancels: -

on_completion

,

on_failure

,

on_cancellation

,

on_crashed

- Define a small cleanup callable that deletes the GCL by name and attach it to those hooks. You can compute the name from

flow_run.id

in the hook, or stash it in a variable/parameter. CLI alternatives (manual or from automation): - Create:

prefect global-concurrency-limit create flow:<id>:local --limit 3

- Delete:

prefect global-concurrency-limit delete flow:<id>:local

Help: Global concurrency limits Summary: - For purely local per-flow-run concurrency: use

ConcurrentTaskRunner(max_workers=...)

. - If you need a server-enforced per-run limit/rate across tasks: yes, create a per-run GCL via the API at start and delete it at the end. There’s no automatic deletion; use try/finally and/or flow hooks for cleanup.

Short answer: ConcurrentTaskRunner is a flow-level setting. It limits the total number of tasks that can run concurrently in that flow run, not just a specific function. If you want to cap only one function, use Prefect’s concurrency controls on that function instead of (or in addition to) the task runner. Ways to limit just one function: - Tag-based global limit (server-enforced) - Tag the task, create a GCL for that tag, only that task is constrained.

@task(tags=["limited_func"])
  def limited_func(...):
      ...

  # GCL (once)
  prefect global-concurrency-limit create tag:limited_func --limit 2

- Direct concurrency context (server-enforced for that function only)

from prefect.concurrency.sync import concurrency

  @task
  def limited_func(...):
      with concurrency("my-func", occupy=1, strict=True):
          ...

- You can also combine with a global limit:

with concurrency(["global-db", "my-func"]):
        ...

- Per-flow-run only (local scope) for that function - Create a per-run GCL at flow start (e.g., name = f"flow{flow run id}my-func") and delete it at the end; then acquire it only in that function. - Or keep it simple and use ConcurrentTaskRunner for general concurrency plus the

concurrency("my-func")

wrapper on the one function to enforce a stricter cap. If you actually want once-per-second pacing for that function, use: -

rate_limit("name")

inside that function, with a limit configured with

slot_decay_per_second

. Docs you might find useful: - Global concurrency limits: Concepts — Global concurrency limits - Tag-based limits: Concepts — Tag-based concurrency limits - API for `rate_limit`/`concurrency`: API — prefect.concurrency.sync