Great question. Here’s a pragmatic, scalable way to manage both Prefect Variables and concurrency limits in Prefect 3.x. High‑level recommendations - Treat them as “config as code”: keep a manifest in your repo and apply via CI/CD to your Prefect workspace(s). - Use Variables for non-sensitive configuration. For secrets, prefer Prefect Secret blocks or an external secret manager. - Separate manifests per environment (dev/stg/prod) and namespace names clearly. - Add a plan/audit step in CI to detect drift and a periodic export for backup. Quick links - Variables: Concepts, How‑to, CLI, Python API - Tag-based concurrency limits: Concepts, How‑to, CLI - Global concurrency limits: Concepts, How‑to, CLI Recommended workflow (variables and limits) 1) Keep manifests in Git - variables.json (or yaml) - concurrency-limits.yaml (both tag-based and global) Example variables.json

{
  "database_url": {
    "value": "<postgresql://prod-db:5432/app>",
    "tags": [
      "database",
      "prod"
    ]
  },
  "api_timeout": {
    "value": 30,
    "tags": [
      "api"
    ]
  },
  "feature_flags": {
    "value": {
      "use_new_pipeline": true
    }
  }
}

Example concurrency-limits.yaml

tag_limits:
  - { tag: "database", limit: 10 }
  - { tag: "third_party_api", limit: 5 }

global_limits:
  - { name: "api_calls", limit: 5, slot_decay_per_second: 2.0, active: true }
  - { name: "big_query_pool", limit: 20, active: true }

2) Apply via CI (idempotent) - Use the CLI for simplicity, or the Python client for more control. - Verified CLI: - Variables:

prefect variable set|get|ls|inspect|delete

- Tag limits:

prefect concurrency-limit create|ls|inspect|reset|delete

- Global limits:

prefect gcl create|ls|inspect|update|enable|disable|delete

Sample GitHub Actions steps

- uses: actions/setup-python@v5
  with:
    python-version: "3.12"

- run: pip install prefect

- name: Apply variables
  env:
    PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }}
    PREFECT_API_KEY: ${{ secrets.PREFECT_API_KEY }}
  run: python scripts/apply_variables.py config/variables.json

- name: Apply concurrency limits
  env:
    PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }}
    PREFECT_API_KEY: ${{ secrets.PREFECT_API_KEY }}
  run: python scripts/apply_limits.py config/concurrency-limits.yaml

scripts/apply_variables.py (idempotent)

import json
from prefect.variables import Variable

import sys
path = sys.argv[1]

with open(path) as f:
    spec = json.load(f)

for name, cfg in spec.items():
    value = cfg["value"]
    tags = cfg.get("tags", [])
    Variable.set(name=name, value=value, tags=tags, overwrite=True)
    print(f"✓ set {name}")

scripts/apply_limits.py (idempotent) ``` import sys, yaml, asyncio from prefect import get_client from prefect.client.schemas.actions import GlobalConcurrencyLimitCreate, GlobalConcurrencyLimitUpdate path = sys.argv[1] async def main(): with open(path) as f: spec = yaml.safe_load(f) async with get_client() as client: # tag-based (task) limits for item in spec.get("tag_limits", []): tag = item["tag"]; limit = int(item["limit"]) try: existing = await client.read_concurrency_limit_by_tag(tag) if existing.concurrency_limit != limit: await client.update_concurrency_limit_by_tag(tag, limit)

print(f"✓ updated tag limit {tag} -> {limit}") else: print(f"• tag limit {tag} unchanged") except Exception: await client.create_concurrency_limit(tag=tag, concurrency_limit=limit) print(f"✓ created tag limit {tag} -> {limit}") # global limits for item in spec.get("global_limits", []): name = item["name"] payload = GlobalConcurrencyLimitCreate( name=name, limit=int(item["limit"]), slot_decay_per_second=item.get("slot_decay_per_second"), active=item.get("active", True), active_slots=item.get("active_slots"), ) try: existing = await client.read_global_concurrency_limit_by_name(name) update = GlobalConcurrencyLimitUpdate( limit=payload.limit, slot_decay_per_second=payload.slot_decay_per_second, active=payload.active, active_slots=payload.active_slots, ) await client.update_global_concurrency_limit(name, update) print(f"✓ upserted global limit {name}") except Exception: await client.create_global_concurrency_limit(payload) print(f"✓ created global limit {name}") asyncio.run(main())

3) Track what exists (audit and drift)
- List variables

prefect variable ls --limit 1000 prefect variable inspect <name> --output json

- List concurrency limits

prefect concurrency-limit ls prefect gcl ls

- Export snapshot (for backup/diff in CI)

python - <<'PY' import asyncio, json from prefect import get_client async def export(): async with get_client() as client: vars = await client.read_variables(limit=10000) print(json.dumps({v.name: {"value": v.value, "tags": v.tags} for v in vars}, indent=2)) asyncio.run(export()) PY ``` You can commit snapshots or push to object storage, then compare them against your manifests for drift and alert in CI. Notes and best practices - Versioning: Prefect Variables are mutable and do not keep history. Git is your source of truth—review changes via PRs. If you need “versioned values,” encode the version in the name (e.g.,

my_config_v2

) or keep version metadata inside the value. - Secrets: Prefer Prefect Secret blocks or your cloud secret manager. Variables are not designed for secret-at-rest guarantees. See:

prefect.blocks.system.Secret

. - Tag-based vs Global concurrency: - Tag-based limits control how many tasks with a tag can run concurrently; add

tags=[...]

on tasks. - Global limits work anywhere in code via context managers and support rate limiting with slot decay. - In recent Prefect 3 releases, tag-based limits are backed by global limits named

tag:{tag}

. - Namespacing: Prefix variable and limit names with env/team (e.g.,

prod_api_timeout

,

teamx_api_calls

) to avoid collisions across workspaces/environments. - Also consider execution-level limits: - Work pool concurrency (cap total running flow runs per pool) and deployment concurrency (cap per-deployment). Those are configured separately if you need them. If you share whether you’re on Prefect Cloud or self-hosted and how many environments you manage, I can tailor manifests and CI examples (GitHub Actions/GitLab/Argo) to your setup and include drift checks.