Peter Žáčik
11/14/2025, 12:50 PMMarvin
11/14/2025, 12:50 PMMarvin
11/14/2025, 12:54 PMfrom prefect import flow
from prefect.context import get_run_context
@flow
def my_flow():
fr = get_run_context().flow_run
if fr.auto_scheduled:
trigger = "schedule"
else:
trigger = "manual or other (e.g. automation/API)"
who = getattr(fr.created_by, "type", None) # 'scheduler', 'user', 'automation', 'api', etc.
print(f"Triggered by: {trigger} (created_by.type={who}, deployment_id={fr.deployment_id})")
Notes:
- If you click “Run” on a deployment in the UI (or use CLI to run a deployment), deployment_id will be set but auto_scheduled will be False and created_by.type will generally be 'user' → that’s a manual run of a deployment.
- If the scheduler created the run from the deployment’s schedule, auto_scheduled will be True and created_by.type will typically be 'scheduler'.
- Ad-hoc runs (not from a deployment) will have deployment_id=None.
If you just want a simple boolean for “scheduled vs not”:
from prefect.context import get_run_context
def is_scheduled_run() -> bool:
return get_run_context().flow_run.auto_scheduled
Optional runtime helpers:
- prefect.runtime.deployment.id() returns the current deployment id (None for ad-hoc runs)
- prefect.runtime.flow_run.parent_flow_run_id() is set for subflows
- prefect.runtime.flow_run.parent_deployment_id() can help when nesting is involved
Docs:
- Access runtime info
- Runtime context
- prefect.runtime.flow_run API
- FlowRunResponse schema
Do you also need to distinguish “manual” vs “automation rule” vs “API call”? If so, we can branch on created_by.type more explicitly.