Short answer: yes. In Prefect 3.x you can limit concurrency at a few levels without changing your flow/task logic: 1) Per work pool (caps total concurrent flow runs pulled by that pool) - No code changes. Applies across all queues in the pool. - CLI:

prefect work-pool update <pool-name> --concurrency-limit <N>

2) Per work queue (caps concurrent flow runs from that queue) - No code changes. Obeys the pool cap as well. - CLI:

prefect work-queue set-concurrency-limit <queue-name> <N> -p <pool-name>

3) Per deployment (caps concurrent runs of a single deployment) - Minimal change at deploy time:

my_flow.deploy(name="my-deployment", concurrency_limit=3)

If your goal is to stop hitting an external API’s rate limits, the most precise control is at the task level, but that does require a small code change: 4) Task-level concurrency via tags (shared limit across all tasks with that tag) - Code: add a tag to the relevant tasks, e.g.

@task(tags=["external-api"])

- CLI (no code change to set the limit once tasks are tagged):

prefect concurrency-limit create external-api 5

5) Global concurrency/rate limits (fine-grained, supports true rate limiting) - Code (inside the task that calls the API):

from prefect.concurrency.sync import rate_limit

@task
def call_api(...):
    rate_limit("external-api", occupy=1)
    ...

- Then create the limit (can be rate-based with decay):

prefect global-concurrency-limit create external-api --limit 5 --slot-decay-per-second 2.0

Recommendations based on your goals: - If you just want to bluntly reduce overall parallelism without code changes: set a per-queue or per-pool concurrency limit. That will reduce concurrent flow runs and usually lowers overall API pressure. - If you need to meet a hard API limit (e.g., 2 req/sec or max 5 concurrent calls), use task tags or global concurrency/rate limits for the specific API-calling tasks. Notes: - There isn’t a single “org-wide” flow-run cap across all pools; you can approximate that by routing flows through a single work pool and setting its limit, or by limiting each pool individually. - “Per work queue task concurrency” isn’t a concept; queues/pools control flow-run concurrency. Task concurrency is controlled by tags or the global concurrency/rate limit APIs. If you can share: - Which work pool(s) and queue(s) your deployments use - Whether you want a cap on concurrent flow runs or a precise API limit (X req/sec or Y concurrent calls) …I can propose exact commands (and/or a minimal code snippet) to implement it.

Great plan. To pick sane caps for work pools/queues and your task runner, here’s a pragmatic way to estimate your API call rate and translate it into limits. Two ways to size it A) Quick estimate (no instrumentation) - Let: - K = task runner max_workers you plan to use (e.g., ThreadPoolTaskRunner(max_workers=K)) - T_call = average API call latency in seconds (including retries if common) - m = average number of API calls per task run (often 1) - Rough per-flow rate: - R_flow_rps ≈ K * m / T_call - Example: K=6, each task makes m=1 call, avg T_call=0.8s ⇒ ~6 / 0.8 ≈ 7.5 rps per flow - Convert to work queue limit: - Given a vendor budget R_budget_rps, choose queue limit N so N * R_flow_rps ≤ R_budget_rps - Leave 20–30% headroom for variance: N ≈ floor(0.7 * R_budget_rps / R_flow_rps) B) Measure it (recommended) - Instrument your API caller once, run a flow for a few minutes, and read the observed per-flow rps. Then multiply by the number of concurrent flow runs you plan to allow. Here’s a drop-in tracker that works with a thread-based task runner

# metrics.py
import time
import threading
from collections import deque

class RateTracker:
    def __init__(self, window_seconds=30):
        self.window = window_seconds
        self._lock = threading.Lock()
        self._events = deque()
        self._latencies = deque()

    def record(self, latency=None):
        now = time.perf_counter()
        with self._lock:
            self._events.append(now)
            if latency is not None:
                self._latencies.append(latency)
            cutoff = now - self.window
            while self._events and self._events[0] < cutoff:
                self._events.popleft()
            while self._latencies and len(self._latencies) > len(self._events):
                self._latencies.popleft()

    def rps_current_window(self):
        with self._lock:
            return len(self._events) / self.window if self.window else 0.0

    def summary(self, start_time):
        dur = time.perf_counter() - start_time
        with self._lock:
            total = len(self._events)
            lat = list(self._latencies)
        avg_rps = total / dur if dur else 0.0
        p95 = None
        if lat:
            lat_sorted = sorted(lat)
            idx = max(0, int(0.95 * len(lat_sorted)) - 1)
            p95 = lat_sorted[idx]
        return {"total_calls": total, "duration_s": dur, "avg_rps": avg_rps, "p95_latency_s": p95}

tracker = RateTracker(window_seconds=30)

Wrap your API call and log a summary at the end of the flow

# api_client.py
import time
import httpx
from metrics import tracker

def call_api(endpoint: str):
    t0 = time.perf_counter()
    try:
        with httpx.Client(timeout=30) as client:
            r = client.get(endpoint)
            r.raise_for_status()
            return r.json()
    finally:
        tracker.record(latency=time.perf_counter() - t0)

Use a thread-based runner and log the summary

from prefect import flow, task, get_run_logger
from prefect.task_runners import ThreadPoolTaskRunner
from metrics import tracker
from api_client import call_api
import time

@task
def fetch_one(url):
    return call_api(url)

@flow(task_runner=ThreadPoolTaskRunner(max_workers=6))
def my_flow(urls):
    logger = get_run_logger()
    flow_start = time.perf_counter()

    for u in urls:
        fetch_one.submit(u)

    # wait for tasks to finish by collecting results
    # (or Prefect will await them implicitly at flow end)
    # _ = [f.result() for f in futures]  # optional

    summary = tracker.summary(flow_start)
    <http://logger.info|logger.info>(f"API usage summary: {summary}")

How to turn measurements into limits - After one representative run, note avg_rps from the summary, call it R_flow_rps. - If your vendor allows R_budget_rps total: - Work queue concurrency limit N ≈ floor(0.7 * R_budget_rps / R_flow_rps) - If that yields N < 1, reduce your max_workers (K) and/or add a task-level limit.

- If your API has minute-based quotas (Q per minute), convert to per-second R_budget_rps ≈ Q/60. Use the same formula. Set the limits - Work pool (caps all queues in the pool):

prefect work-pool update <pool-name> --concurrency-limit <N>

- Work queue (best for your case, caps concurrent flow runs from that queue):

prefect work-queue set-concurrency-limit <queue-name> <N> -p <pool-name>

- Task runner inside the flow:

from prefect.task_runners import ThreadPoolTaskRunner
@flow(task_runner=ThreadPoolTaskRunner(max_workers=K))
def my_flow(...): ...

Optionally: exact rate limiting with a token bucket (if you need hard guarantees) - If you end up needing strict caps regardless of latency variance, add a one-liner to the API-calling task and create a global rate limit: - In task:

from prefect.concurrency.sync import rate_limit

@task
def fetch_one(url):
    rate_limit("external-api", occupy=1)  # one token per request
    return call_api(url)

- Create the rate limit (token bucket): set refill rate to your rps and “bucket size” to your tolerated burst:

prefect global-concurrency-limit create external-api --limit 20 --slot-decay-per-second 5
# Above: allows ~5 req/sec sustained with up to 20 burst tokens

Caveats and tips - Use ThreadPoolTaskRunner for sync HTTP clients; ProcessPool won’t share the in-process tracker. - Retries/backoff increase call count—include them in your measurements. - Measure against realistic workloads for a few minutes; use the 70–80% headroom rule in production. - If you run multiple flows with different shapes, measure each flow type separately and size per-queue limits based on the heaviest per-flow rps on that queue. If you share: - Your vendor budget (e.g., 300/minute or 5/sec) - Your planned max_workers (K) - A measured avg call latency T_call, or I can suggest a default …I can calculate an initial queue limit N and K for you, plus a burst-friendly global rate limit config if you want hard caps.