Short answer: you can’t set max_scheduled_runs in prefect.yaml today. Details: - In Prefect 3.x, max_scheduled_runs is supported on deployment schedules via the Python client API, but it is not a supported field in prefect.yaml’s schedule entries. - YAML-supported schedule fields include things like cron/interval/rrule, timezone, active, parameters, slug, replaces, etc., but not max_scheduled_runs. If you need it, set it programmatically after creating the deployment:

from prefect.client import get_client
from prefect.client.schemas.actions import DeploymentScheduleCreate
from prefect.client.schemas.schedules import CronSchedule
import asyncio

async def main():
    async with get_client() as client:
        dep = await client.read_deployment_by_name("my-flow/my-deployment")
        schedule = DeploymentScheduleCreate(
            schedule=CronSchedule(cron="0 9 * * *"),
            active=True,
            slug="morning-run",
            max_scheduled_runs=10,
        )
        await client.create_deployment_schedules(
            deployment_id=dep.id,
            schedules=[schedule],
        )

asyncio.run(main())

Notes and links: - Deployment schedule API (supports max_scheduled_runs): DeploymentScheduleCreate in source - Schedule how-to guides: Create schedules and Manage schedules Related settings: - There is a server-side cap (default 50) enforced by validation; if you need to raise the ceiling cluster-wide, configure the server setting PREFECT_DEPLOYMENT_SCHEDULE_MAX_SCHEDULED_RUNS in your server environment. Validation reference: schedule max_scheduled_runs validator. If you actually meant “limit how many runs can be running at once,” use a deployment concurrency limit in prefect.yaml (different from max_scheduled_runs):

deployments:
  - name: my-deployment
    entrypoint: flow.py:my_flow
    concurrency_limit: 3

If you want, share your current prefect.yaml and the behavior you’re targeting; I can suggest the exact steps for your setup.

✅ thought for 380.4 seconds

Short answer: yes, you can stack other decorators with

@task

in Prefect 3.x — but not “any” decorator, and order matters. Rules of thumb: - Use decorators that preserve metadata with

functools.wraps

. Without it, Prefect can lose your function’s name/signature. - Keep

@task

closest to the function definition (innermost). This lets Prefect wrap the real function and avoids issues with wrappers that don’t look like normal Python callables. - For async tasks, any wrapper around an

async def

must itself be

async

and

await

the wrapped call. - Avoid decorators that turn the function into something else (e.g.,

click.command

,

typer

) or heavy C-level wrappers on the task itself (e.g.,

lru_cache

). Apply those to a helper function instead, then call it from the task. - Prefer Prefect’s built-in task features (e.g.,

retries=

) over external retry decorators to avoid double-retry confusion. Good pattern (custom decorators):

from functools import wraps
from prefect import task, flow

def log_calls(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        print(f"Calling {fn.__name__}")
        return fn(*args, **kwargs)
    return wrapper

@task(retries=2, timeout_seconds=30)   # innermost
@log_calls                              # outer
def compute(x: int) -> int:
    return x * 2

@flow
def my_flow():
    return compute(5)

Async-safe wrapper:

from functools import wraps
from prefect import task

def async_log(fn):
    @wraps(fn)
    async def wrapper(*args, **kwargs):
        print(f"Starting {fn.__name__}")
        result = await fn(*args, **kwargs)
        print(f"Finished {fn.__name__}")
        return result
    return wrapper

@task
@async_log
async def fetch(url: str) -> str:
    ...

lru_cache pattern (wrap a helper, not the task):

from functools import lru_cache
from prefect import task

@lru_cache(maxsize=128)
def _expensive(x: int) -> int:
    return x * 2

@task
def cached_task(x: int) -> int:
    return _expensive(x)

Avoid:

# Click/typer turn functions into CLI commands, not plain callables
from prefect import task
import click

@task
@click.command()           # ❌ don’t stack with @task
def cli_task(...):
    ...

Why this works: - Prefect inspects the wrapped function for metadata and signatures; using

@wraps

and keeping

@task

innermost helps Prefect see the original function cleanly and correctly detect async vs sync. See the source for how Prefect handles wrapped callables and signatures: signature handling and task implementation. Tasks concept docs: Tasks. If you share which decorator(s) you want to combine with

@task

, I can suggest the safest ordering or an alternative pattern.