vzehnder
08/12/2025, 10:54 PMfrom prefect import flow
@flow
def my_flow(timeout: int = 3600):
from prefect.context import get_run_context
context = get_run_context()
context.flow_run.timeout_seconds = timeout
...
# Call with different timeouts
my_flow(timeout=7200) # 2 hours
vzehnder
08/12/2025, 10:55 PMfrom prefect import flow
@flow
def my_flow(timeout: int = 3600):
from prefect.context import get_run_context
context = get_run_context()
context.flow_run.timeout_seconds = timeout
...
# Call with different timeouts
my_flow(timeout=7200) # 2 hours
Marvin
08/12/2025, 10:55 PMMarvin
08/12/2025, 11:00 PMget_run_context().flow_run.timeout_seconds
. That object isn’t the source of truth for timeout enforcement, and changing it won’t affect the engine.
Supported ways to set timeouts:
- At definition time on the decorator (Prefect 2.x and 3.x):
from prefect import flow
@flow(timeout_seconds=3600)
def my_flow():
...
- Per-call (dynamic) using with_options
(Prefect 2.x and 3.x):
from prefect import flow
@flow
def my_flow():
...
# Call with different timeouts per run
my_flow.with_options(timeout_seconds=7200)() # 2 hours
my_flow.with_options(timeout_seconds=300)() # 5 minutes
If you’re using deployments, set the timeout in code (as above) or create separate deployments with different configs. Changing it mid-run via context isn’t a thing—sadly, neither is my enthusiasm.