Great question! In Prefect 3.x,

task.submit(...)

(even for an async task) returns a

PrefectFuture

. You can inspect its current state via the

state

property, or wait for it to finish and then get the result. Quick patterns: - Read current state (non-blocking)

future = my_async_task.submit(arg1, arg2)
print(future.state.type)              # e.g. "RUNNING", "COMPLETED", "FAILED"
print(future.state.is_completed())    # True/False
print(future.state.is_failed())       # True/False
print(future.state.is_final())        # Completed/Failed/Crashed/Cancelled

- In an async flow: await completion and get the result

from prefect import task, flow
import asyncio

@task
async def my_async_task(x):
    await asyncio.sleep(1)
    return x * 2

@flow
async def my_flow():
    f = my_async_task.submit(5)

    # Option 1: poll the state if you want live updates
    while not f.state.is_final():
        print("Current state:", f.state.type)
        await asyncio.sleep(0.5)

    # Option 2: wait and get result (async)
    await f.wait_async()        # waits until done
    result = await f.result_async()
    print("Result:", result)

- In a sync flow: block until done and get the result

from prefect import task, flow

@task
async def my_async_task(x):
    return x * 2

@flow
def my_flow():
    f = my_async_task.submit(5)
    f.wait(timeout=30)          # blocks until completion (optional timeout)
    print(f.state.type)         # e.g. COMPLETED
    print(f.result())           # returns 10 (raises on failure by default)

- If you want the State object immediately instead of a future:

state = my_async_task.submit(5, return_state=True)
print(state.type)

Common state checks on `future.state`: -

is_running()

-

is_completed()

-

is_failed()

-

is_crashed()

-

is_cancelled()

-

is_paused()

-

is_final()

Do you want to check state inside an async flow (using

await

) or from a sync flow? I can tailor the example to your exact setup.

Short answer: there isn’t a mode where

submit()

returns two objects. But you don’t need it—every

PrefectFuture

has a

state

property. So you can “return both” by returning the future and reading

future.state

. Examples: - From an async flow

from prefect import task, flow
import asyncio

@task
async def my_task(x):
    await asyncio.sleep(1)
    return x * 2

@flow
async def my_flow():
    f = my_task.submit(5)
    initial = f.state            # snapshot (likely PENDING/SCHEDULED/RUNNING)

    await f.wait_async()         # ensure completion
    final = f.state              # now terminal (COMPLETED/FAILED/etc.)

    return f, final              # return both the future and its (final) state

- From a sync flow

from prefect import task, flow

@task
async def my_task(x):
    return x * 2

@flow
def my_flow():
    f = my_task.submit(5)
    f.wait()                     # block until done
    return f, f.state

Notes: -

return_state=True

makes

submit()

return a

State

instead of a

PrefectFuture

. There’s no built-in way to get both from one call—prefer returning the future and reading

future.state

. -

future.state

is a snapshot; call `wait()`/`wait_async()` (or `result()`/`result_async()`) before reading it if you want the final state. If you have a specific pattern in mind (e.g., background

.delay()

or checking mid-run progress), I can tailor the snippet.

Best options in Prefect 3.x with DaskTaskRunner: 1) Limit total parallelism via the Dask cluster (recommended for “only N tasks at once”)

from prefect import flow, task
from prefect_dask import DaskTaskRunner

@task
def work(i): ...
    
@flow(task_runner=DaskTaskRunner(
    cluster_kwargs={
        "n_workers": 5,            # total workers
        "threads_per_worker": 1,   # 1 task per worker
        "processes": False         # threads (True for processes if CPU-bound)
    }
))
def my_flow():
    futures = [work.submit(i) for i in range(100)]
    return [f.result() for f in futures]

- Total concurrency ≈ n_workers × threads_per_worker. - If connecting to a remote Dask cluster (

address=...

), set limits on the cluster side. 2) Server-side concurrency limits for specific task types (by tag) - Create a limit (verified CLI):

prefect concurrency-limit create database 3

- Tag tasks you want limited:

from prefect import task

@task(tags=["database"])
def db_task(...): ...

- Manage limits:

prefect concurrency-limit ls
prefect concurrency-limit inspect database
prefect concurrency-limit reset database
prefect concurrency-limit delete database

3) Dask resource annotations for per-task throttling inside the Dask cluster

import dask
from prefect import task, flow
from prefect_dask import DaskTaskRunner

@task
def api_call(i): ...

@flow(task_runner=DaskTaskRunner(
    cluster_kwargs={"n_workers": 10, "resources": {"API": 3}}
))
def run():
    futures = []
    for i in range(50):
        with dask.annotate(resources={"API": 1}):
            futures.append(api_call.submit(i))
    return [f.result() for f in futures]

- Only 3 tasks annotated with

API:1

run at once. 4) Fine-grained control inside task code via Prefect’s concurrency context manager - Sync tasks:

from prefect.concurrency.sync import concurrency

@task
def t():
    with concurrency("api_calls", occupy=1):
        ...

- Async tasks:

from prefect.concurrency.asyncio import concurrency

@task
async def t():
    async with concurrency("api_calls", occupy=1):
        ...

Notes: - DaskTaskRunner doesn’t have a

max_concurrent_tasks

knob; use worker/thread counts or the mechanisms above. - On some platforms, guard your entrypoint:

if __name__ == "__main__":
    my_flow()

Do you want to cap overall throughput for the whole flow, or only certain tasks (e.g., API/DB calls)? I can tailor the example accordingly.