Short answer: Yes — just not “inside” deploy itself. In Prefect 3.x you can create (or upsert) a global concurrency limit in the same script right before you call

flow.deploy()

, then have your flow/tasks use it via tags. Key points: - Global concurrency limits are named. To apply them to tagged runs, name them

tag:<your-tag>

. - Tasks/flows that carry

<your-tag>

will respect a global limit named

tag:<your-tag>

. - This is separate from a deployment’s own

concurrency_limit

(which limits concurrent flow runs of that deployment). Example: create the limit, then deploy a flow that uses it via tags

from prefect import flow, task
from prefect.client.orchestration import get_client

TAG = "db-write"  # the tag your tasks/flows will use

@task(tags=[TAG])
def do_work(i: int):
    # ... your logic ...
    pass

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

def deploy():
    # 1) Ensure the global concurrency limit exists
    import anyio
    async def _ensure_limit():
        async with get_client() as client:
            # create or update idempotently
            await client.upsert_global_concurrency_limit_by_name(
                name=f"tag:{TAG}",
                limit=5,  # max concurrent runs across anything using this tag
            )
    anyio.run(_ensure_limit)

    # 2) Deploy the flow; adding the same tag at the deployment level is optional
    #    (tasks tagged with TAG will be limited either way)
    my_flow.deploy(
        name="db-writes",
        work_pool_name="your-pool",
        tags=[TAG],  # helpful if you also want flow runs to carry the tag
    )

if __name__ == "__main__":
    deploy()

CLI alternative (verified): - Create the global concurrency limit:

prefect global-concurrency-limit create tag:db-write --limit 5

- Deploy as usual with

prefect deploy

or via

flow.deploy(...)

. Notes: - If your goal is to limit concurrent flow runs of a specific deployment, use the

concurrency_limit

parameter on

flow.deploy(...)

instead (this is different from global concurrency limits). - If you want the limit to apply to task runs, tag the tasks (as in the example) or add the tag to the deployment so it propagates to runs. Quick clarification to make sure you get the right behavior: - Are you trying to limit concurrent task runs across all deployments via a shared tag (global limit), or limit how many flow runs of this single deployment can run at once (deployment concurrency)?

@Marvin I'm not using flow.deploy(), I'm using

prefect deploy

. I would prefer to keep definitions of my rate limits in code or configuration, I'm open to where, but I prefer to not have an ad-hoc script that contains both rate limit configuration and deploy commands

Yes. With

prefect deploy

you can keep your concurrency/rate limit definitions in code/config and have them created/updated as part of the deploy pipeline—no separate ad‑hoc script required. Two clean patterns: Option A: A small custom step called from prefect.yaml (recommended) - Put a tiny helper in your repo that upserts global concurrency limits via the Prefect client. - Call it in the

build

(or

push

) steps of prefect.yaml so it runs during

prefect deploy

. Example step (steps/gcl.py):

from typing import Iterable, Mapping
import anyio
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import GlobalConcurrencyLimitUpdate

def upsert_global_concurrency_limits(limits: Iterable[Mapping[str, object]]):
    """
    limits: iterable of dicts like:
      {"name": "tag:db-write", "limit": 5, "active": True}
    """
    async def _go():
        async with get_client() as client:
            for item in limits:
                name = str(item["name"])
                limit = int(item["limit"])
                active = bool(item.get("active", True))
                # Upsert (create-or-update by name)
                await client.upsert_global_concurrency_limit_by_name(name=name, limit=limit)
                # Set 'active' if provided
                await client.update_global_concurrency_limit(
                    name=name,
                    concurrency_limit=GlobalConcurrencyLimitUpdate(limit=limit, active=active),
                )
    anyio.run(_go)
    return {"count": len(list(limits))}

prefect.yaml:

build:
  - steps.gcl.upsert_global_concurrency_limits:
      limits:
        - name: "tag:db-write"
          limit: 5
        - name: "tag:external-api"
          limit: 10
          active: true

deployments:
  - name: my-deployment
    entrypoint: flows/my_flow.py:my_flow
    tags: ["db-write", "external-api"]  # flows/tasks will carry these tags

Why this works - Global concurrency limits that should apply to tags must be named

tag:<your-tag>

. - By running this in

build

, the limits are ensured/updated every time you run

prefect deploy

. - Tasks/flows that carry those tags will respect the limits automatically. Option B: No-Python shell step in prefect.yaml - Use the built-in shell step to create or update via CLI.

build:
  - prefect.deployments.steps.run_shell_script:
      script: |
        set -e
        prefect global-concurrency-limit create tag:db-write --limit 5 || \
        prefect global-concurrency-limit update tag:db-write --limit 5
        prefect global-concurrency-limit create tag:external-api --limit 10 || \
        prefect global-concurrency-limit update tag:external-api --limit 10

Notes and reminders - Use

tag:<name>

for the global concurrency limit name to have tasks/flows with

<name>

tag respect it. - If you instead want to limit concurrent flow runs of a single deployment, set the deployment’s own

concurrency_limit

(separate feature). - Verified CLI:

prefect global-concurrency-limit create|update|enable|disable|ls|inspect|delete

(you can rely on these in the shell-step pattern). If you share how you’d like to structure the limits (inline in prefect.yaml vs a separate config file), I can sketch that out too (e.g., reading a YAML file and passing it into the step).

You’re running into this because, in Prefect 3.x,

prefect.deployments.steps.run_shell_script

executes each line as an independent command (no interactive shell). Shell operators like

||

,

&&

, and

set -e

only work when they’re interpreted by a shell (e.g., bash). Good news: you have two solid options. Option A — Execute a script file Run your own

.sh

file with bash (and any flags you want), instead of inline commands.

build:
  - prefect.deployments.steps.run_shell_script:
      id: run-deploy-script
      script: "bash -euo pipefail ./scripts/deploy.sh"
      directory: /path/to/project   # optional working dir
      env:
        MY_ENV: "value"             # optional env
      stream_output: true

Option B — Wrap inline commands with bash -c If you prefer inline, wrap your script in

bash -c '...'

so bash interprets operators and options like

set -e

.

build:
  - prefect.deployments.steps.run_shell_script:
      id: complex-bash
      script: |
        bash -c '
          set -euo pipefail
          echo "Start"
          make build || { echo "Build failed"; exit 1; }
          make test
          echo "Done"
        '
      stream_output: true

Notes and gotchas - Each line is run separately unless you wrap it in

bash -c '...'

. That’s why

||

,

&&

, and

set -e

appear to “not work.” - The step already raises if any single command (line) returns non-zero, but if you need compound command semantics (

cmd1 || cmd2

, pipelines, etc.), use

bash -c

or a script file. - You can set a working directory via

directory

, environment via

env

, and enable simple env expansion via

expand_env_vars: true

. Docs and source - API reference: Utility steps (run_shell_script) - Source: run_shell_script implementation If you share a small snippet of what you’re trying to run, I can help convert it to a

bash -c

form or a script-file layout.

Short answer: not directly. With

prefect deploy --all

, Prefect 3.x runs the build (and push) steps for each deployment. There isn’t a “project-level” or “run-once” stage, and step outputs aren’t shared across deployments. Ways to handle a one-time action (like setting limits): - Run it once outside of deploy (recommended) - Put a one-time command in your CI/CD before/after

prefect deploy --all

. For example, make concurrency limits idempotent:

# Run once before/after `prefect deploy --all`
  prefect concurrency-limit inspect my-tag >/dev/null 2>&1 || prefect concurrency-limit create my-tag 10

- CLI help:

prefect concurrency-limit --help

shows available subcommands. You can list/inspect/create/delete. - We verified the commands exist via CLI help. - Keep it in build, but make it idempotent - This still executes per deployment, but does no work after the first:

build:
    - prefect.deployments.steps.run_shell_script:
        id: ensure-limits
        script: |
          set -e
          if prefect concurrency-limit inspect my-tag >/dev/null 2>&1; then
            echo "Concurrency limit for 'my-tag' already exists"
          else
            prefect concurrency-limit create my-tag 10
          fi
        stream_output: true

- Use a small Python script (run once) that applies org-wide/project-wide limits via the API - Useful if you’re setting several limits at once or need more logic. Run this script once in CI/CD rather than in build steps. Why this is necessary: -

prefect deploy --all

loops through each deployment and runs the configured build/push for each. There is no built-in “run once per project” step and no flag to skip stages globally. See Prefect YAML/deploy guides: - Define deployments with prefect.yaml - Create deployments If you can share which limits you’re setting (task tag concurrency limits, deployment concurrency, work pool/queue limits), I can give you an exact one-time command or script tailored to that type.