For self-hosted Prefect 3.x, the best way to manage per-tenant concurrency is usually: Use a tenant-specific concurrency key and enforce limits with Prefect concurrency controls. There are a few ways to do it, depending on what exactly you want to limit. --- ## 1. Best fit: Global concurrency limits per tenant This is the most flexible approach for multi-tenant systems. You create one limit per tenant, for example: -

tenant:acme

→ 10 -

tenant:globex

→ 5 -

tenant:foo

→ 20 Then in your flow/task code, you acquire that tenant’s concurrency slot before doing work. Example pattern:

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

@task
def process_for_tenant(tenant_id: str, item: str):
    with concurrency(f"tenant:{tenant_id}", occupy=1, strict=True):
        print(f"Processing {item} for {tenant_id}")

@flow
def tenant_flow(tenant_id: str, items: list[str]):
    for item in items:
        process_for_tenant.submit(tenant_id, item)

If

tenant:acme

has a limit of 10, then across all running flows/tasks using that key, only 10 concurrent executions will be allowed for that tenant. ### Create the limits You can create them with the CLI:

prefect gcl create tenant:acme --limit 10
prefect gcl create tenant:globex --limit 5

You can inspect/update them too:

prefect gcl inspect tenant:acme
prefect gcl update tenant:acme --limit 15
prefect gcl ls

Why this is good for multi-tenancy: - works across all workers - works in self-hosted server - lets each tenant have a different quota - limits can be changed without redeploying code --- ## 2. If you want per-tenant per-resource limits Often you want more than “tenant total concurrency”. For example: - tenant A can run 20 jobs total - but only 3 DB-heavy jobs at once - and only 5 API calls at once Use names like: -

tenant:acme

-

tenant:acme:db

-

tenant:acme:api

Example:

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

@task
def db_work(tenant_id: str):
    with concurrency(f"tenant:{tenant_id}:db", occupy=1, strict=True):
        print("doing db work")

@task
def api_work(tenant_id: str):
    with concurrency(f"tenant:{tenant_id}:api", occupy=1, strict=True):
        print("doing api work")

This gives you much better isolation if different tenants compete for shared databases or external APIs. --- ## 3. Deployment concurrency limits are not enough for multi-tenant control Prefect also supports deployment-level concurrency limits, but those limit runs of a specific deployment, not a tenant across the whole platform. Useful when: - one deployment should only have N concurrent runs Not enough when: - tenant A may trigger multiple deployments/flows - you want one shared quota across them all So for multi-tenant quotas, global concurrency limits are usually the right primitive. --- ## 4. Work pools can help with isolation, but they are heavier Another option is one work pool per tenant with its own concurrency limit. That works well if you want stronger isolation, like: - separate worker infrastructure per tenant - dedicated Kubernetes namespaces / node pools - different execution environments But it’s operationally heavier. Good pattern: - Use work pools for infrastructure isolation - Use global concurrency limits for tenant quotas --- ## 5. Recommended architecture For most self-hosted multi-tenant setups, I’d recommend this: ### Option A: Shared workers + tenant concurrency keys Best if tenants share the same infrastructure. - one or a few shared work pools - one global concurrency limit per tenant - optionally one per tenant/resource Example: -

tenant:acme = 10

-

tenant:globex = 3

-

tenant:acme:db = 2

### Option B: Dedicated work pool per tenant + tenant concurrency keys Best if some tenants need stronger isolation. -

acme-pool

-

globex-pool

- plus

tenant:acme

,

tenant:globex

limits in code This gives both:

- infra isolation - app-level concurrency control --- ## 6. Important implementation detail: how to know the tenant You need a consistent way to derive

tenant_id

, for example from: - flow parameters - deployment parameters - tags - job variables / worker environment - your own app’s request metadata Simplest is usually passing

tenant_id

as a flow parameter. Example:

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

@task
def do_work(tenant_id: str, x: int):
    with concurrency(f"tenant:{tenant_id}", strict=True):
        print(f"{tenant_id}: {x}")

@flow
def tenant_pipeline(tenant_id: str, values: list[int]):
    for v in values:
        do_work.submit(tenant_id, v)

--- ## 7. Use

strict=True

I strongly recommend

strict=True

for tenant enforcement. That way if the concurrency limit does not exist or cannot be enforced, Prefect will fail instead of silently continuing. Example:

with concurrency(f"tenant:{tenant_id}", strict=True):
    ...

Without strict mode, missing limits may be tolerated depending on context, which is risky for quota enforcement. --- ## 8. CLI verification The current Prefect CLI supports the

gcl

commands for global concurrency limits. Useful commands: -

prefect gcl create

-

prefect gcl inspect

-

prefect gcl update

-

prefect gcl ls

If you want, I can also give you: 1. a complete example of tenant-aware flows/tasks, or 2. a bootstrap script that auto-creates concurrency limits for each tenant. --- ## 9. Practical recommendation If your goal is simply: "Tenant A should never consume more than X concurrent task slots" then do this: 1. Create a global concurrency limit per tenant 2. Pass

tenant_id

into all tenant work 3. Wrap the relevant task/flow sections with:

with concurrency(f"tenant:{tenant_id}", strict=True):

That is the cleanest Prefect-native solution. --- ## Relevant docs / references - Global concurrency limits - Tag-based concurrency limits - Deployment concurrency limits - `prefect.concurrency.sync` API If you want, I can next show you a full multi-tenant design example for: - shared workers - tenant ID propagation - tenant-specific limits - and how to configure it in a self-hosted Prefect setup.

Yes — deployment schedules work fine, but with an important distinction: A deployment schedule only decides when a flow run is created. It does not by itself enforce per-tenant concurrency across tenants. So if you use scheduled deployments, your tenant concurrency approach still works as long as the flow/tasks acquire the tenant-specific concurrency limit when they run. ## What happens Say you have: - Deployment

tenant-acme-daily

- schedule: every 5 minutes - flow parameter

tenant_id="acme"

When the schedule triggers, Prefect creates a flow run. When that flow run starts executing, your code does:

with concurrency(f"tenant:{tenant_id}", strict=True):
    ...

If

tenant:acme

is already at its limit, that work will wait until a slot is available. So the answer is: Yes, scheduled deployments are compatible with this pattern. ## Important nuance If you only use: - deployment schedules - deployment concurrency limits then you are limiting that deployment only, not all work for that tenant. Example: -

acme-etl

deployment limit = 3 -

acme-reporting

deployment limit = 2 That does not mean tenant

acme

is capped at 3 or 5 globally unless you also add a shared tenant concurrency key. So for real tenant-wide control, use: - schedules on deployments - plus tenant-specific global concurrency limits in flow/task code ## Typical pattern

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

@task
def do_work(tenant_id: str):
    with concurrency(f"tenant:{tenant_id}", strict=True):
        print(f"running work for {tenant_id}")

@flow
def tenant_job(tenant_id: str):
    do_work.submit(tenant_id)

Then create separate deployments/schedules like: - deployment 1 →

tenant_id="acme"

- deployment 2 →

tenant_id="globex"

Each scheduled deployment can run independently, but actual execution will still respect: -

tenant:acme

-

tenant:globex

## One more thing to watch If you want to limit the entire flow run per tenant, put the concurrency context at the flow level. If you want to limit only specific heavy sections, put it inside specific tasks. ### Flow-level example

from prefect import flow
from prefect.concurrency.sync import concurrency

@flow
def tenant_flow(tenant_id: str):
    with concurrency(f"tenant:{tenant_id}", strict=True):
        # entire flow counts against tenant quota
        ...

### Task-level example

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

@task
def heavy_task(tenant_id: str):
    with concurrency(f"tenant:{tenant_id}", strict=True):
        ...

## Recommendation If each deployment belongs to one tenant and is schedule-driven: - keep using deployment schedules - pass tenant ID as a deployment parameter - enforce concurrency with global concurrency limits That’s the right setup. If you want, I can show you a concrete scheduled multi-tenant deployment example with: - one flow - multiple tenant deployments - schedules - and per-tenant concurrency limits.