Fernando Müller
11/09/2025, 7:41 PMMarvin
11/09/2025, 7:41 PMMarvin
11/09/2025, 7:45 PMfrom prefect import flow, task
from prefect.runtime import flow_run
@task
def show_flow_run_name():
print(f"Flow run name: {flow_run.name}")
@flow
def my_flow():
show_flow_run_name()
if __name__ == "__main__":
my_flow()
Alternative: use the run context + client (useful if you need more than just the name). In a task, get_run_context() returns a TaskRunContext, so you’ll fetch the flow run by ID:
- Async task
from prefect import flow, task
from prefect.context import get_run_context
from prefect.client.orchestration import get_client
@task
async def show_flow_run_name():
ctx = get_run_context() # TaskRunContext
async with get_client() as client:
fr = await client.read_flow_run(ctx.task_run.flow_run_id)
print(f"Flow run name: {fr.name}")
@flow
def my_flow():
show_flow_run_name()
if __name__ == "__main__":
my_flow()
- Sync task
from prefect import flow, task
from prefect.context import get_run_context
from prefect.client.orchestration import get_client
@task
def show_flow_run_name():
ctx = get_run_context() # TaskRunContext
with get_client(sync_client=True) as client:
fr = client.read_flow_run(ctx.task_run.flow_run_id)
print(f"Flow run name: {fr.name}")
@flow
def my_flow():
show_flow_run_name()
if __name__ == "__main__":
my_flow()
Notes and caveats:
- prefect.runtime.flow_run.name is the easiest and works inside tasks; if there’s no active run it may be None, so guard if needed.
- In a task, get_run_context() returns a TaskRunContext which does not include the flow run object; use ctx.task_run.flow_run_id + client to fetch details.
- Prefer prefect.runtime.flow_run.name if you only need the name.
Docs:
- Access runtime info (runtime.*)
- Runtime and run context concepts
- get_run_context API