How to get root flow_run_id?

[50Bytes-dev]

How to get root flow run id by runtime context?

from prefect import flow, task

@task
def deep():
    ...  # how to get the flow_run_id of the root flow?

@flow
def child():
    deep()

@flow
def root():
    child()

[agammann]

In Prefect 3, prefect.runtime.flow_run.root_flow_run_id resolves the root of the parent flow-run chain:

from prefect import flow, task
from prefect.runtime import flow_run


@task
def deep() -> str:
    return flow_run.root_flow_run_id


@flow
def child() -> str:
    return deep()


@flow
def root() -> str:
    expected = flow_run.id
    actual = child()
    assert actual == expected
    return actual


assert root()

I ran this with Prefect 3.8.8 against a real temporary local API and verified that the nested task received the top-level flow's ID. The underlying accessor is documented as get_root_flow_run_id.

That runtime attribute is absent in the Prefect 2.20.25 version I checked. For that version, passing the root ID explicitly is a small alternative:

from prefect import flow, task
from prefect.runtime import flow_run


@task
def deep(root_id: str) -> str:
    return root_id


@flow
def child(root_id: str) -> str:
    return deep(root_id)


@flow
def root() -> str:
    root_id = flow_run.id
    assert child(root_id) == root_id
    return root_id


assert root()

I also ran this fallback on Prefect 2.20.25. Both tests used Python 3.12.14. The Prefect 2 environment used Pydantic 2.10.6 and AnyIO 4.8.0. These examples follow nested flow calls in the same orchestration chain; an unrelated deployment run started independently is a different root.