Short answer: In Prefect 3.x there isn’t a global setting to apply a timeout to every task. Timeouts are per-task via the

timeout_seconds

parameter on the

@task

decorator. To avoid repeating yourself, create a small wrapper around

prefect.task

that supplies a default. Example pattern (project-wide default):

# my_project/prefect_ext.py
from prefect import task as _prefect_task

DEFAULT_TASK_TIMEOUT = 300  # seconds

def task(*args, timeout_seconds=DEFAULT_TASK_TIMEOUT, **kwargs):
    # Only set the default if the caller did not specify one
    if "timeout_seconds" not in kwargs:
        kwargs["timeout_seconds"] = timeout_seconds
    return _prefect_task(*args, **kwargs)

Usage:

from my_project.prefect_ext import task

@task  # gets 300s timeout by default
def do_work():
    ...

@task(timeout_seconds=30)  # per-task override
def quick_task():
    ...

Why this approach - There’s no

PREFECT_*

setting for default task timeouts in 3.x. - The wrapper keeps your code DRY while still allowing per-task overrides. References - Task API (shows

timeout_seconds

): Task API - Settings and profiles (note there’s no global timeout setting): Settings & profiles, Settings reference If you’re on Prefect 2.x or want a different pattern (e.g., a decorator that injects defaults or applying

.with_options

programmatically), let me know and I can tailor it.

Yes — use ParamSpec/TypeVar plus overloads so your wrapper mirrors Prefect’s generic decorator and preserves the original function’s signature and return type. Here’s a typed wrapper you can drop in (supports both @task and @task(...)):

# my_project/prefect_ext.py
from typing import Any, Callable, Optional, overload, TypeVar

try:
    # Python 3.10+ has ParamSpec in typing
    from typing import ParamSpec
except ImportError:  # Python <3.10
    from typing_extensions import ParamSpec

from prefect import task as _prefect_task
from prefect.tasks import Task  # for typing the return

P = ParamSpec("P")
R = TypeVar("R")

DEFAULT_TASK_TIMEOUT: float = 300.0  # seconds

@overload
def task(
    __fn: None = ...,
    *,
    timeout_seconds: float | int | None = DEFAULT_TASK_TIMEOUT,
    **kwargs: Any,
) -> Callable[[Callable[P, R]], Task[P, R]]: ...
@overload
def task(
    __fn: Callable[P, R],
    *,
    timeout_seconds: float | int | None = DEFAULT_TASK_TIMEOUT,
    **kwargs: Any,
) -> Task[P, R]: ...

def task(
    __fn: Callable[P, R] | None = None,
    *,
    timeout_seconds: float | int | None = DEFAULT_TASK_TIMEOUT,
    **kwargs: Any,
) -> Task[P, R] | Callable[[Callable[P, R]], Task[P, R]]:
    # Only set the default if caller didn’t provide one
    if "timeout_seconds" not in kwargs:
        kwargs["timeout_seconds"] = timeout_seconds

    if __fn is None:
        # Used as @task(...)
        def _decorator(fn: Callable[P, R]) -> Task[P, R]:
            return _prefect_task(fn, **kwargs)
        return _decorator

    # Used as @task
    return _prefect_task(__fn, **kwargs)

Usage:

from my_project.prefect_ext import task

@task  # default 300s timeout, with full type inference on args/return
def do_work(x: int) -> str:
    return str(x)

@task(timeout_seconds=30)  # override per task
def quick_task() -> None:
    ...

Notes - This preserves the callable’s type parameters (P, R), so IDEs and type checkers still know your task’s argument and return types. - For Python <3.10, install typing_extensions (or vendor ParamSpec) for best type checking. API ref for the underlying decorator (shows

timeout_seconds

and generics): Task API