Great question. Here’s the Prefect-recommended approach for concurrency “tags” in 3.x and how to handle them in dev vs prod. Assumptions: You’re on Prefect 3.x. Concurrency limits for tasks are “tag-based” and are backed by global concurrency limits on the server. Best-practice summary - Use resource-centric tags on tasks (e.g.

db

,

api:stripe

,

warehouse

) and keep the tag taxonomy small and stable across flows. - Create/manage the limits in your environment (Cloud/Server) via CLI/API/IaC, not in flow code. This avoids drift and makes limits auditable. - Scope by workspace or by tag naming if you need per-env isolation (e.g.,

db:dev

,

db:prod

) or just use separate workspaces. - Treat the server as the single source of truth; tasks without a limit on their tag will run unconstrained. How to set up and test 1) Tag your tasks in code

from prefect import flow, task

@task(tags=["db"])
def query_db(i: int):
    ...

@flow
def pipeline():
    for i in range(20):
        query_db.submit(i)

2) Local testing - Start a local API so limits can be enforced:

prefect server start

- Create a small limit for your tags:

prefect concurrency-limit create db 2
prefect concurrency-limit ls
prefect concurrency-limit inspect db

- Run your flow; you’ll see tasks wait in AwaitingConcurrencySlot when the limit is reached. 3) Production - Create and maintain limits via: - CLI (seeded by CI/CD bootstrap scripts) - Python client/API - Terraform (recommended for IaC) - Example CLI:

prefect concurrency-limit create db 10
prefect concurrency-limit inspect db
prefect concurrency-limit reset db      # recover if slots get stuck
prefect concurrency-limit delete db

Where these are tracked - Tags live in your code (on tasks). - The limits and active usage are tracked centrally in your Prefect API database (Cloud/Server). Tag-based limits are stored as global concurrency limits named `tag:<your_tag>`; list and inspect them with:

prefect concurrency-limit ls
prefect concurrency-limit inspect <tag>
# or to see all global limits (including the `tag:` ones)
prefect gcl ls

- You can also view/manage these in the Prefect UI (Cloud/Server) or via the REST/Python client. Notes and gotchas - Enforcement requires a running Prefect API (Cloud or local server). Fully offline runs will not enforce tag-based limits. - If a task has multiple tags with limits, it must acquire a slot for all of them; it will wait for the most constrained tag. - For rate-limiting or non-task code, use global concurrency limits (GCL) and the

concurrency()/rate_limit()

helpers, which also require the API. Useful links - Concept: Tag-based concurrency limits - How-to: Set task limits by tag - Concept: Global concurrency limits - How-to: Apply global concurrency and rate limits - CLI reference: prefect concurrency-limit and prefect gcl If you’d like, I can suggest a small bootstrap script (or Terraform config) to seed your required limits for dev/staging/prod. Are you on Prefect Cloud or running OSS Server?

Short answer: yes—treat concurrency limits as code. Two good options: - Recommended: Terraform provider for Prefect (drift detection, plan/apply, imports) - Also works: a YAML manifest in your repo + a CI job that applies it via CLI or Python client If you want YAML specifically, here’s a proven pattern. YAML as source of truth - Keep a single file per environment or a single file with env sections - Version in Git - Apply only via CI/CD with a service account/API key Example YAML

# prefect-concurrency.yaml
environments:
  dev:
    task_tags:
      - tag: db
        limit: 2
      - tag: api:stripe
        limit: 1
    global:
      - name: warehouse-writes
        limit: 4
        active: true
        slot_decay_per_second: 0
  prod:
    task_tags:
      - tag: db
        limit: 10
    global:
      - name: warehouse-writes
        limit: 8
        active: true

Apply script (idempotent) using CLI - Tag-based limits don’t have an “update” subcommand; update by delete+create - Global limits support updates

#!/usr/bin/env python
import subprocess, sys, json, yaml

def sh(cmd):
    return subprocess.run(cmd, text=True, capture_output=True)

def ensure_tag_limit(tag, limit):
    # Check if exists
    res = sh(["prefect", "concurrency-limit", "inspect", tag])
    if res.returncode != 0:
        subprocess.check_call(["prefect", "concurrency-limit", "create", tag, str(limit)])
        return
    # No update subcommand; re-create if limit changed
    current = res.stdout
    if f"concurrency_limit: {limit}" not in current:
        subprocess.check_call(["prefect", "concurrency-limit", "delete", tag])
        subprocess.check_call(["prefect", "concurrency-limit", "create", tag, str(limit)])

def read_gcl(name):
    res = sh(["prefect", "gcl", "inspect", name, "-o", "json"])
    if res.returncode != 0:
        return None
    return json.loads(res.stdout)

def ensure_gcl(lim):
    name = lim["name"]
    desired = {
        "limit": int(lim["limit"]),
        "slot_decay_per_second": float(lim.get("slot_decay_per_second", 0)),
        "active": bool(lim.get("active", True)),
    }
    current = read_gcl(name)
    if not current:
        cmd = ["prefect", "gcl", "create", name, "--limit", str(desired["limit"])]
        if desired["slot_decay_per_second"]:
            cmd += ["--slot-decay-per-second", str(desired["slot_decay_per_second"])]
        subprocess.check_call(cmd)
        if not desired["active"]:
            subprocess.check_call(["prefect", "gcl", "disable", name])
        return
    # Update if needed
    needs_update = (
        current["limit"] != desired["limit"] or
        float(current.get("slot_decay_per_second") or 0) != desired["slot_decay_per_second"]
    )
    if needs_update:
        cmd = ["prefect", "gcl", "update", name, "--limit", str(desired["limit"]),
               "--slot-decay-per-second", str(desired["slot_decay_per_second"])]
        subprocess.check_call(cmd)
    if desired["active"] and current.get("active") is False:
        subprocess.check_call(["prefect", "gcl", "enable", name])
    if not desired["active"] and current.get("active") is True:
        subprocess.check_call(["prefect", "gcl", "disable", name])

def main(manifest_path, environment):
    cfg = yaml.safe_load(open(manifest_path))
    env_cfg = cfg["environments"][environment]
    for tag_cfg in env_cfg.get("task_tags", []):
        ensure_tag_limit(tag_cfg["tag"], int(tag_cfg["limit"]))
    for gcl_cfg in env_cfg.get("global", []):
        ensure_gcl(gcl_cfg)
    # Optional: implement “prune” logic to delete limits not listed in YAML

if __name__ == "__main__":
    # Requires PREFECT_API_URL and PREFECT_API_KEY set in the CI environment
    main(sys.argv[1], sys.argv[2])

Example CI step (GitHub Actions) ``` - name: Apply Prefect concurrency limits env: PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }} PREFECT_API_KEY: ${{ secrets.PREFECT_API_KEY }} run: | pip install prefect pyyaml

python scripts/apply_concurrency.py prefect-concurrency.yaml prod

Notes
- Enforcement requires a running API (Cloud/Server). YAML is your source of truth; Prefect’s DB is the runtime state.
- Tag-based updates: delete+create. Global limits can be updated in place.
- Consider separate files per env/workspace, or one file with env sections as above.
- Optional: add “prune” behavior to delete limits that are no longer in your YAML, especially for GCL (be careful in prod).

If you prefer richer drift detection and a plan/apply workflow, use Terraform instead of custom YAML
- Provider: <https://registry.terraform.io/providers/PrefectHQ/prefect|PrefectHQ/prefect>
- Global limits resource: <https://registry.terraform.io/providers/PrefectHQ/prefect/latest/docs/resources/global_concurrency_limit|prefect_global_concurrency_limit>
- Tag-based (task) limits resource: <https://registry.terraform.io/providers/PrefectHQ/prefect/latest/docs/resources/task_run_concurrency_limit|prefect_task_run_concurrency_limit>
- Repo: <https://github.com/PrefectHQ/terraform-provider-prefect|github.com/PrefectHQ/terraform-provider-prefect>

Example Terraform

resource "prefect_task_run_concurrency_limit" "db_tag" { tag = "db" concurrency_limit = 10 } resource "prefect_global_concurrency_limit" "warehouse_writes" { name = "warehouse-writes" limit = 8 slot_decay_per_second = 0 active = true } ``` CLI references - Tag-based CLI: prefect concurrency-limit - Global CLI: prefect gcl Want me to tailor the YAML schema and apply script to your exact envs and naming (dev/staging/prod, workspace layout)?

Absolutely—Terraform works great for local testing with Docker. How to wire Terraform to your local Prefect Server - Run Prefect Server in Docker (exposed on 4200) - Example compose service:

services:
      prefect:
        image: prefecthq/prefect:3-latest
        ports:
          - "4200:4200"
        command: prefect server start --host 0.0.0.0 --port 4200

- Configure the Terraform provider to point at your local API:

terraform {
    required_providers {
      prefect = {
        source = "prefecthq/prefect"
      }
    }
  }

  provider "prefect" {
    endpoint = "<http://localhost:4200>"
  }

- Alternatively, set environment variables in your shell/CI:

export PREFECT_API_URL="<http://localhost:4200>"
  # No API key needed for local OSS

Manage your limits locally with Terraform - Tag-based (task) limits:

resource "prefect_task_run_concurrency_limit" "db_tag" {
    tag               = "db"
    concurrency_limit = 2
  }

- Global concurrency limits:

resource "prefect_global_concurrency_limit" "warehouse_writes" {
    name                  = "warehouse-writes"
    limit                 = 4
    slot_decay_per_second = 0
    active                = true
  }

Typical local workflow - Start server:

docker compose up -d

for your Prefect service - Initialize and apply Terraform:

terraform init
  terraform plan
  terraform apply

- Test your flows against the limits: - Tag your tasks in code (e.g.,

@task(tags=["db"])

) - Run your flow; the server enforces the limits - Inspect limits via CLI if you want:

prefect concurrency-limit ls
    prefect gcl ls

Notes and tips - The provider supports endpoint via

endpoint

or

PREFECT_API_URL

. No auth needed for local OSS. - If you enable CSRF on your local server, set in provider:

csrf_enabled = true

. - Limits are stored in the server’s database (in your Docker volume). Terraform state remains in your workspace, so you get drift detection locally just like in prod. - For environment parity, keep the same Terraform modules and pass different variables/workspaces for dev vs prod. You can also point

endpoint

to Cloud in prod and to

<http://localhost:4200>

locally. - Remember: concurrency enforcement requires a running API. Offline runs won’t enforce limits. Docs and provider links - Terraform provider: PrefectHQ/prefect - Global concurrency resource: prefect_global_concurrency_limit - Tag-based task limit resource: prefect_task_run_concurrency_limit - CLI refs: prefect concurrency-limit, prefect gcl If you share your compose file and desired limits, I can sketch a minimal Terraform module you can run locally and then promote to prod with variables.