It depends on what “it” is. In Prefect 3.x: - Top-level flow run with a deployment concurrency limit: yes — a slot is taken when the run enters PENDING and held through RUNNING/CANCELLING; released when it leaves those states. - Task run with a tag-based concurrency limit: yes, but only while the task is RUNNING. - Subflows: do not use deployment/work-pool slots; they only respect tag-based limits if you’ve applied tags. - Retries and any Scheduled/Awaiting states (e.g., AwaitingRetry, AwaitingConcurrencySlot): do not hold a slot; a new slot is acquired when the retry actually starts again. - Work pool concurrency (worker capacity): a worker consumes capacity when it pulls a run; subflows don’t go through the work pool, so they don’t use this capacity. Which case are you asking about?

Short answer: for the worker limit, only top‑level flow runs use a slot. - Top-level flow run: yes. A worker slot is taken as soon as the worker claims the run and successfully moves it to PENDING, and it stays held through PENDING/RUNNING/CANCELLING until the worker is done (then it’s released). - Subflows and tasks: no. They run inside the parent flow’s process and do not consume worker slots. - Retries: AwaitingRetry doesn’t use a slot; the new attempt uses a slot only when the worker picks it up again. - Scheduled/queued/awaiting states: no slot until the worker actually claims the run.

Got it — you don’t want to tie up a worker slot while your flow is just waiting on an API for a couple minutes. Use a pause to free the worker during the wait, then resume to continue: - pause_flow_run: puts the flow in PAUSED, frees the worker immediately, and resumes from the next line when resumed. Works for ad‑hoc runs and deployments. - suspend_flow_run: similar but reschedules the flow (requires a deployment and persisted results). Best for longer waits/greater resilience. Minimal example (async flow, 2‑minute wait without blocking a worker):

from prefect import flow
from prefect.flow_runs import pause_flow_run

@flow
async def poll_api_then_continue():
    # do your API call here
    # ... call API, get a job_id, etc.

    # Don't sleep — pause instead to release the worker
    await pause_flow_run(timeout=180)  # 3 minutes max wait

    # After resume (manually or via automation/webhook), continue
    # ... check API for completion, finish up

If you need to capture structured input on resume (e.g., webhook payload or approval), you can:

from prefect.flow_runs import pause_flow_run
from prefect.input import RunInput

class DoneSignal(RunInput):
    job_id: str

await pause_flow_run(wait_for_input=DoneSignal, timeout=600)

Then resume it with:

from prefect.client import get_client
from uuid import UUID

async with get_client() as client:
    await client.resume_flow_run(
        flow_run_id=UUID("..."),
        run_input={"job_id": "123"}
    )

Notes: - Subflows can pause, but suspend_flow_run is only for top‑level flows with deployments and persisted results. - Awaiting/paused states don’t consume worker slots; the slot is reacquired when the run resumes and becomes PENDING/RUNNING again. If you’d like, share how you plan to trigger the resume (manual click, webhook, automation), and I can tailor the pattern.

If the call is truly synchronous (e.g.,

requests.get(..., timeout=120)

that blocks for ~2 minutes), then yes — that flow run will hold a worker slot the entire time it’s waiting. There’s no way to “free” the slot in the middle of a blocking call. Options to avoid tying up the worker: - Restructure to “submit + wait” and pause in between: - Kick off the job (or a short request), then

pause_flow_run(timeout=...)

and resume later (manually, via webhook, or automation), then check status and continue. - Works in sync flows;

pause_flow_run

is synchronous.

from prefect import flow
from prefect.flow_runs import pause_flow_run

@flow
def api_job():
    job_id = start_job()           # returns quickly with an id
    pause_flow_run(timeout=180)    # frees the worker slot for up to 3 minutes
    status = check_status(job_id)  # after resume
    return status

- For longer/safer handoffs, use suspend (requires a deployment and persisted results):

from prefect import flow
from prefect.flow_runs import suspend_flow_run

@flow(persist_result=True)
def api_job():
    job_id = start_job()
    suspend_flow_run(timeout=180)  # frees worker and reschedules on resume
    status = check_status(job_id)
    return status

- If the API absolutely forces a single blocking call you can’t break up: - You will use a worker slot for the duration. Mitigations: - Run these flows in a dedicated work pool (so they don’t starve other work) and/or increase worker concurrency for that pool. - Offload the blocking call to an external service (e.g., serverless/job queue) and have your flow pause and wait for a callback to resume. - Polling alternative: lower the HTTP timeout (e.g., 10–20s), loop with

pause_flow_run

between polls so each wait frees the worker. If you can share whether your API supports “submit + status” or a webhook callback, I can sketch the exact pause/resume or automation wiring.