Short answer: - Calling a task directly (e.g.,

my_task()

) runs it synchronously and blocks the flow until it finishes, returning the actual result. - Calling

my_task.submit()

schedules it on the flow’s task runner (thread/process pool), returns immediately with a

PrefectFuture

, and enables concurrency; you call

future.result()

when you want the value. Details in Prefect 3.x: - Execution model - Direct call: Executes inline within the flow run, sequential and blocking. - submit(): Enqueued on the flow’s task runner (ThreadPoolTaskRunner by default). Non-blocking; multiple tasks can run concurrently. - Return type - Direct call: Returns the task’s result; raises immediately on failure. - submit(): Returns a `PrefectFuture[T]`; use

future.result()

to get the value or inspect

future.state

for status. - Concurrency/parallelism - Direct call: No parallelism; tasks run one after another. - submit(): Parallelism via the task runner (threads or processes), supports fan-out patterns and overlapping work. - Retries, caching, and limits - Both modes apply the same retries, retry delays, caching, and tag-based concurrency limits. The difference is only whether the flow is blocked while those happen. - State and observability - Both create a task run with full state tracking and logging in the UI. With submit, state progresses asynchronously; with direct calls, it happens inline. - Dependencies - Direct call: Natural ordering via code (one call after another). - submit(): Use

wait_for=[...]

or depend on other futures to control ordering when tasks don’t pass data. - Async nuances - In async flows, calling an async task directly lets you

await

it naturally. If you

submit()

and then immediately call

future.result()

in an async flow, you’ll block; prefer direct

await

or gather futures thoughtfully. When to use which - Use direct calls when: - You want simple, sequential logic with clear data dependencies - You’re fine with the flow blocking until the task finishes - You’re in an async flow calling async tasks (natural

await

) - Use submit when: - You want to run many independent tasks concurrently (fan-out) - You want the flow to continue doing work while tasks run/retry in the background - You need thread/process parallelism for I/O or CPU workloads Minimal examples

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def work(x):
    return x * 2

# Direct call: sequential
@flow
def sequential_flow():
    a = work(1)  # blocks
    b = work(a)  # runs after previous completes
    return b

# Submitted: concurrent
@flow(task_runner=ThreadPoolTaskRunner(max_workers=4))
def concurrent_flow():
    futures = [work.submit(i) for i in range(10)]  # returns immediately
    results = [f.result() for f in futures]        # wait when you choose
    return results

Great question. In Prefect 3.x, you’ll programmatically start runs from FastAPI using the Prefect Client. The typical pattern is to trigger a registered deployment. Below are ready-to-use snippets and notes. Key docs: - Prefect Client API (v3) - Deployments in Prefect 3.x - Workers and work pools Prereqs - Your FastAPI service must be configured to talk to Prefect: - Set

PREFECT_API_URL

and

PREFECT_API_KEY

in the FastAPI service environment. - Ensure a worker is running for the target work pool so the run is picked up. Trigger a deployment by ID (async FastAPI)

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.exceptions import ObjectNotFound

app = FastAPI()

class TriggerRequest(BaseModel):
    deployment_id: UUID
    parameters: dict = {}
    idempotency_key: str | None = None
    tags: list[str] = []

@app.post("/trigger", status_code=status.HTTP_202_ACCEPTED)
async def trigger(request: TriggerRequest):
    try:
        async with get_client() as client:
            # Optional: verify deployment exists
            try:
                await client.read_deployment(request.deployment_id)
            except ObjectNotFound:
                raise HTTPException(status_code=404, detail="Deployment not found")

            flow_run = await client.create_flow_run_from_deployment(
                deployment_id=request.deployment_id,
                parameters=request.parameters,     # merged with deployment defaults
                idempotency_key=request.idempotency_key,  # prevents dupes on retries
                tags=request.tags,
                # name="optional-custom-name",
                # scheduled_start_time=None,   # or datetime for delayed start
                # work_queue_name=None,        # optional override
                # job_variables=None,          # infra overrides if needed
            )
            return {
                "flow_run_id": str(flow_run.id),
                "name": flow_run.name,
                "state": flow_run.state.type,             # typically SCHEDULED
                "scheduled_start_time": flow_run.scheduled_start_time,
            }
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to create flow run: {e}")

Trigger a deployment by name (flow_name/deployment_name)

@app.post("/trigger-by-name", status_code=202)
async def trigger_by_name(flow_name: str, deployment_name: str, parameters: dict | None = None):
    try:
        async with get_client() as client:
            deployment = await client.read_deployment_by_name(f"{flow_name}/{deployment_name}")
            flow_run = await client.create_flow_run_from_deployment(
                deployment_id=deployment.id,
                parameters=parameters or {},
            )
            return {"flow_run_id": str(flow_run.id), "state": flow_run.state.type}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Get run status

@app.get("/runs/{flow_run_id}")
async def get_run_status(flow_run_id: UUID):
    try:
        async with get_client() as client:
            flow_run = await client.read_flow_run(flow_run_id)
            return {
                "flow_run_id": str(flow_run.id),
                "state": flow_run.state.type,
                "start_time": flow_run.start_time,
                "end_time": flow_run.end_time,
            }
    except ObjectNotFound:
        raise HTTPException(status_code=404, detail="Flow run not found")

Sync endpoint variant ``` from prefect.client.orchestration import get_client @app.post("/trigger-sync", status_code=202)

def trigger_sync(deployment_id: UUID, parameters: dict | None = None): try: with get_client(sync_client=True) as client: flow_run = client.create_flow_run_from_deployment( deployment_id=deployment_id, parameters=parameters or {}, ) return {"flow_run_id": str(flow_run.id), "state": flow_run.state.type} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) ``` Notes and best practices - Use

idempotency_key

if your endpoint might be retried (to avoid duplicate runs). - Parameters are merged with deployment defaults (they don’t replace them wholesale). - The new run will typically be in SCHEDULED state; a worker in the deployment’s work pool must be running to pick it up. - Return 202 Accepted quickly; don’t block your API by waiting for the flow to finish. - Prefer triggering deployments for remote execution. If you just want to run the flow locally inside the API process, call the flow function directly; but that bypasses orchestration and workers. If you share your exact input shape (IDs vs names, parameter schema), I can tailor the endpoint code further.

Yes.

get_client()

returns a Prefect client you can use to trigger a deployment. You don’t trigger from

get_client

itself; you create the client and call

create_flow_run_from_deployment(...)

on it. Minimal async example (recommended in FastAPI):

from prefect.client.orchestration import get_client

async def trigger_by_id(deployment_id, params=None):
    async with get_client() as client:
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id=deployment_id,
            parameters=params or {},
            # idempotency_key="optional-unique-key",
            # tags=["api-triggered"],
            # work_queue_name="optional-override",
        )
        return str(flow_run.id)

Lookup by name (flow_name/deployment_name):

async def trigger_by_name(flow_name: str, deployment_name: str, params=None):
    async with get_client() as client:
        deployment = await client.read_deployment_by_name(f"{flow_name}/{deployment_name}")
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id=deployment.id,
            parameters=params or {},
        )
        return str(flow_run.id)

Sync variant (if you’re not in an async context):

from prefect.client.orchestration import get_client

def trigger_sync(deployment_id, params=None):
    with get_client(sync_client=True) as client:
        flow_run = client.create_flow_run_from_deployment(
            deployment_id=deployment_id,
            parameters=params or {},
        )
        return str(flow_run.id)

Notes - Configure your service with

PREFECT_API_URL

and

PREFECT_API_KEY

. - The new run will usually be SCHEDULED; ensure a worker is running for the deployment’s work pool (or override queue via

work_queue_name

). - You can use

idempotency_key

to avoid duplicate runs on retries. Refs: - Prefect Client API (v3) - Deployments

Short answer: No, you don’t need a separate “task worker” process just because your flow uses

.map

. In Prefect 3.x: - By default, flows run with

ThreadPoolTaskRunner

, which executes submitted/mapped tasks in a thread pool inside the same flow process. That means

.submit()

and

.map()

will run concurrently without any extra worker processes. - You only need an additional process when you intentionally use

PrefectTaskRunner

(the distributed task runner) or APIs like `.delay()`/`.apply_async()` that create pending task runs for a task worker to pick up. What you do need for a deployment: - A flow worker to pick up and start the flow run itself (e.g.,

prefect worker start -p <your-work-pool>

). This is separate from the task runner inside your flow. - Once the flow run starts on a worker, your mapped tasks will execute using whatever task runner you configured on the flow: - Default:

ThreadPoolTaskRunner

→ tasks run in threads in the same process -

ProcessPoolTaskRunner

→ tasks run in processes in the same machine -

PrefectTaskRunner

→ tasks are deferred and require a task worker to execute them Rule of thumb: - If you didn’t configure

PrefectTaskRunner

, you don’t need a task worker. You only need a flow worker for the deployment to start.

Short answer: - Tasks: use tag-based limits (server-enforced) and/or global concurrency limits (code-enforced). - Flows: control how many flow runs can execute via deployment/work pool concurrency, not tags. Here’s how to set each in Prefect 3.x. Task concurrency (two complementary options) 1) Tag-based task limits (server-enforced) - Add tags to tasks (or via a

tags()

context) and create limits for those tags. - The server blocks a task from entering Running if any of its tags are out of slots. Code:

from prefect import flow, task, tags

@task(tags=["database"])
def query_db(q: str):
    ...

@flow
def my_flow():
    with tags("high-priority"):  # adds at runtime
        a = query_db.submit("select * from table")

Create/inspect limits:

prefect concurrency-limit create database 5
prefect concurrency-limit ls
prefect concurrency-limit inspect database
prefect concurrency-limit reset database

Notes: - Tags from decorator and

with tags(...)

are unioned. - Child tasks don’t inherit parent task tags automatically; use

with tags(...)

around the child call if you want propagation. - Limits are enforced at the server when a task transitions to Running; they don’t rely on your task runner. 2) Global concurrency limits (code-enforced) - Use global limits to control any critical section (DB pool size, API rate, GPU). - Works for sync and async tasks via context managers. Sync:

from prefect.concurrency.sync import concurrency, rate_limit

@task
def heavy_db():
    with concurrency("db_pool", occupy=1):   # blocks if no slots
        return do_db_work()

Async:

from prefect.concurrency.asyncio import concurrency, rate_limit

@task
async def call_api():
    await rate_limit("external-api")  # e.g., 10 tokens/sec
    return await http_call()

Create/update global limits:

prefect gcl create db_pool --limit 5
prefect gcl create external-api --limit 10 --slot-decay-per-second 10.0
prefect gcl inspect db_pool
prefect gcl update db_pool --limit 8
prefect gcl delete db_pool

Flow-level concurrency - There isn’t a “flow tag” concurrency limit. Instead, limit how many flow runs execute via: - Deployment concurrency limit (per-deployment) - Work pool concurrency limit (across deployments in a pool) Set deployment concurrency: - With serve:

from prefect import flow, serve

@flow
def etl():
    ...

if __name__ == "__main__":
    serve(
        etl,
        name="etl-deployment",
        concurrency_limit=3,  # max concurrent flow runs for this deployment
    )

- Or via CLI:

prefect deploy ./path/to/flow.py:etl --name etl-deployment --concurrency-limit 3

Work pool concurrency: - Set a limit on the pool to cap concurrent flow runs across deployments in that pool. Additional tips - Combine limits: A task can have tags (server limit) and use a global limit block (client limit) to coordinate multiple resources. - Multiple tags: all tagged limits must have available slots for the task to run. - If tasks appear “stuck,” inspect limits: -

prefect concurrency-limit inspect <tag>

-

prefect gcl inspect <name>

- Adjust the server retry wait when tags are saturated via server config: -

PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS

(default ~30s) If you share your exact flow/task layout and desired caps (e.g., “max 2 concurrent flow runs; within a run, at most 5 DB tasks and 10 API calls/sec”), I can draft a tailored configuration snippet.

Short answer: Don’t submit 20k mapped tasks all at once with the default ThreadPoolTaskRunner. Use batching/windowing or streaming patterns to control memory and server load. Why - ThreadPoolTaskRunner has no submission queue limit;

.map

will create 20k futures immediately. That can: - Bloat memory (futures, states, results) - Overwhelm the Prefect API/UI with state writes - Increase chance of hitting OS/thread/file descriptor limits - External services (APIs/DBs) may throttle you; without backpressure, you’ll see timeouts/retries. Better patterns - Batch mapping (recommended)

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import wait

@task
def process_item(x: int) -> str:
    return f"ok-{x}"

@flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
def process_large():
    items = list(range(20_000))
    batch_size = 1000  # tune 500–2000 based on result size

    results = []
    for i in range(0, len(items), batch_size):
        batch = items[i:i+batch_size]
        futs = process_item.map(batch)
        done, not_done = wait(futs, timeout=300)
        results.extend([f.result() for f in done])
        if not_done:
            # handle timeouts/failures per batch
            pass
    return results

- Stream results as they complete (low memory)

from prefect import flow, task
from prefect.futures import as_completed
from prefect.task_runners import ThreadPoolTaskRunner

@task
def work(i: int) -> str:
    return f"r-{i}"

@flow(task_runner=ThreadPoolTaskRunner(max_workers=16))
def stream_large():
    futs = work.map(range(20_000))
    for f in as_completed(futs):
        try:
            r = f.result()
            # process/store r immediately; don’t buffer
        except Exception as e:
            # handle single-item failure
            ...

- Windowed parallelism + rate limiting - Combine a modest thread pool with Prefect concurrency limits to cap effective concurrency even if you submit many tasks.

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

# one-time setup: prefect concurrency-limit create external-api 50

@task
def call_api(i: int):
    with concurrency("external-api", occupy=1):
        return do_http(i)

@flow(task_runner=ThreadPoolTaskRunner(max_workers=32))
def process_with_limit():
    futs = call_api.map(range(20_000))
    return [f.result() for f in futs]

Important tips - Use

unmapped()

for static parameters in

.map

to avoid per-item context changes that can spawn many threads. - Tune

max_workers

for your workload (I/O: 8–16; CPU-bound: 4–8). - If you truly need to scale beyond a single machine, consider

PrefectTaskRunner

with task workers to distribute mapped tasks across multiple workers. - If tasks return large payloads, prefer streaming (as_completed) or persist results as you go to avoid holding everything in memory. If you share more about the task’s cost (I/O vs CPU), result size, and any external rate limits, I can suggest concrete batch sizes and runner settings.