Short answer: use a flow-level state change hook and query the API for the current flow run’s task runs using the Prefect client. The hook receives

(flow, flow_run, state)

. You can then filter task runs by

flow_run.id

. Here are minimal, working patterns. Async hook (recommended inside flows)

from prefect import flow, get_run_logger
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterId

async def collect_task_details(flow, flow_run, state):
    logger = get_run_logger()
    async with get_client() as client:
        task_runs = await client.read_task_runs(
            flow_run_filter=FlowRunFilter(
                id=FlowRunFilterId(any_=[flow_run.id])
            )
        )

    # Example: build a compact summary
    details = []
    for tr in task_runs:
        tr_state = tr.state
        details.append(
            {
                "task_run_id": str(tr.id),
                "task_name": tr.name,
                "state_type": tr_state.type.value if tr_state else None,
                "state_name": tr_state.name if tr_state else None,
                "start_time": tr.start_time,
                "end_time": tr.end_time,
                "run_time_seconds": getattr(tr, "total_run_time", None),
                "attempt": (tr_state.state_details.run_count if tr_state else None),
            }
        )

    <http://logger.info|logger.info>(f"Collected {len(details)} task runs for flow {flow_run.name}: {details}")

@flow(on_completion=[collect_task_details], on_failure=[collect_task_details])
def my_flow():
    ...

Synchronous hook (if you prefer not to use async)

from prefect import flow
from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterId

def collect_task_details_sync(flow, flow_run, state):
    with SyncPrefectClient() as client:
        task_runs = client.read_task_runs(
            flow_run_filter=FlowRunFilter(
                id=FlowRunFilterId(any_=[flow_run.id])
            )
        )
    # ...process task_runs...

@flow(on_failure=[collect_task_details_sync])
def my_flow():
    ...

Notes and options: - The hook runs in the flow’s runtime environment. On Prefect Cloud, your flow run already has the right API URL and key; no extra setup needed. - The query returns only task runs that belong to the current flow run. If you need task runs from nested subflows, first read subflow runs and then query their task runs:

from prefect.client.schemas.filters import FlowRunFilterParentFlowRunId

async with get_client() as client:
    subflows = await client.read_flow_runs(
        flow_run_filter=FlowRunFilter(
            parent_flow_run_id=FlowRunFilterParentFlowRunId(any_=[flow_run.id])
        )
    )
    # Then loop over subflows and call read_task_runs for each subflow.id

- Be aware that flow on_failure hooks can run on intermediate failures (e.g., before retries complete). Check `state.type`/`state.name` if you only want final outcomes. Docs: - State change hooks: State change hooks (Prefect 2) - Client usage: Using the Prefect client - States overview: States Would you like the example to also include nested subflow task runs, or a richer summary (e.g., logs, retries, exception messages)?

Short answer: not directly. In Prefect 2.x,

get_run_context()

gives you the current run context (for a flow:

FlowRunContext

with

flow

and `flow_run`; for a task:

TaskRunContext

with

task

and

task_run

). It does not contain all task runs for the flow. To retrieve all task run details, use the Prefect client and query by

flow_run.id

. You can combine both: use

get_run_context()

to get the current

flow_run.id

, then query the API. Example inside a flow-level state change hook

from prefect import flow, get_run_context
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterId

async def collect_task_details(flow, flow_run, state):
    # You already get flow_run in the hook, but get_run_context() works too
    ctx = get_run_context()
    assert str(ctx.flow_run.id) == str(flow_run.id)

    async with get_client() as client:
        task_runs = await client.read_task_runs(
            flow_run_filter=FlowRunFilter(
                id=FlowRunFilterId(any_=[flow_run.id])
            )
        )

    # task_runs now has all task run objects for this flow run
    ...

@flow(on_completion=[collect_task_details], on_failure=[collect_task_details])
def my_flow():
    ...

If you’re inside a task and want details about just that task,

get_run_context().task_run

is available; but for “all tasks in the flow,” you must query via the client as above. Docs: - State change hooks: State change hooks (Prefect 2) - Runtime context: Runtime context - Using the client: Using the Prefect client

Short answer: No —

get_run_context().task_run_futures

isn’t a supported/public attribute in Prefect 2.x. The flow runtime context does not expose a complete list of task futures. If you’ve seen similar attributes like

task_run_states

or

task_run_results

, those are internal and not reliable for this use-case; they’re often empty during runtime and don’t represent all task runs. There’s an open report about this behavior: FlowRunContext.task_run_states is empty whereas I can see tasks in the UI. Supported approach in Prefect 2.x: - Use

get_run_context()

only to get the current flow run id. - Use the Prefect client to read task runs filtered by that

flow_run.id

. Example inside a flow-level hook

from prefect import flow, get_run_context
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterId

async def collect_task_details(flow, flow_run, state):
    ctx = get_run_context()
    async with get_client() as client:
        task_runs = await client.read_task_runs(
            flow_run_filter=FlowRunFilter(id=FlowRunFilterId(any_=[ctx.flow_run.id]))
        )
    # ...process task_runs...

@flow(on_completion=[collect_task_details], on_failure=[collect_task_details])
def my_flow():
    ...

If you prefer not to call the API, the only alternative is to explicitly track futures yourself in the flow (e.g., store each

task.submit()

future in a list/dict as you create them). However, that registry won’t be available in a state change hook unless you persist it somewhere (e.g., block, external store) or query the API as above. Docs: - State change hooks: State change hooks (Prefect 2) - Runtime context: Runtime context - Using the client: Using the Prefect client

Great question. In Prefect 2.x, your flow state-change hook receives the new

State

object, which contains the flow’s return value. The simplest, robust way to get it is to resolve the state with

state.result()

(or

get_state_result(state)

). Quick example:

from prefect import flow
from prefect.states import get_state_result

def on_success(flow, flow_run, state):
    # This hook is only attached to completion, so the state will be COMPLETED
    try:
        value = state.result()  # returns the Python object your flow returned
        # alternatively: value = get_state_result(state, raise_on_failure=False)
        print(f"Flow {flow.name} returned: {value}")
    except Exception as exc:
        # Should be rare for a COMPLETED state, but this guards against missing results
        print(f"Could not retrieve result: {exc}")

@flow(on_completion=[on_success])
def my_flow():
    return {"foo": 1, "bar": 2}

my_flow()

Notes and gotchas: -

state.result()

returns the actual Python value your flow returned. If the state is failed and you pass the default

raise_on_failure=True

, it will raise that exception; for success it returns the value. - In most cases, hooks run in the same process as the flow and will have access to the result immediately. - If you ever see a “missing result” error (e.g., very large objects, custom types, or cross-process scenarios), enable result persistence so Prefect can fetch it from storage. Since you’re on GCP, here’s how to persist results to GCS so they’re always retrievable:

from prefect import flow
from prefect.serializers import PickleSerializer
from prefect_gcp.cloud_storage import GcsBucket

gcs = GcsBucket.load("my-gcs-bucket-block")  # configure this block in Prefect first

def on_success(flow, flow_run, state):
    value = state.result()
    print(f"Flow {flow.name} returned: {value}")

@flow(
    persist_result=True,
    result_storage=gcs,
    result_serializer=PickleSerializer(),
    on_completion=[on_success],
)
def my_flow():
    return {"foo": 1, "bar": 2}

If you can share what you plan to do with the output in the hook (e.g., send to Slack, write to BigQuery, etc.), I can tailor the example to that.

Yes. In Prefect 2.x you can query all child (subflow) runs from inside a flow state-change hook and inspect their states (and results, if persisted). Two common ways: 1) Directly filter subflows by parent flow run ID (simplest)

from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterParentFlowRunId

def on_completion(flow_obj, flow_run, state):
    client = get_client(sync_client=True)

    # Get all direct child subflows for this flow run
    subflow_runs = client.read_flow_runs(
        flow_run_filter=FlowRunFilter(
            parent_flow_run_id=FlowRunFilterParentFlowRunId(any_=[flow_run.id])
        )
    )

    for sf in subflow_runs:
        sf_state = sf.state  # or sf.state_type for a quick summary
        print(f"Subflow {sf.name} ({sf.id}) -> {sf_state.type.value if sf_state else sf.state_type}")

        # Optionally fetch the subflow's return value
        if sf_state and sf_state.is_completed():
            try:
                value = sf_state.result()
                print(f"  result: {value}")
            except Exception as e:
                print(f"  could not retrieve result: {e}")

@flow(on_completion=[on_completion])
def parent_flow():
    # call your subflows here
    pass

2) Derive subflows from task runs in the parent

from prefect.client.schemas.filters import TaskRunFilter, TaskRunFilterFlowRunId

def on_completion(flow_obj, flow_run, state):
    client = get_client(sync_client=True)

    task_runs = client.read_task_runs(
        task_run_filter=TaskRunFilter(
            flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run.id])
        )
    )

    child_flow_run_ids = [
        tr.state.state_details.child_flow_run_id
        for tr in task_runs
        if tr.state and tr.state.state_details and tr.state.state_details.child_flow_run_id
    ]

    for child_id in child_flow_run_ids:
        sf = client.read_flow_run(child_id)
        sf_state = sf.state
        print(f"Subflow {sf.name} ({sf.id}) -> {sf_state.type.value if sf_state else sf.state_type}")

Notes: - For results:

sf.state.result()

will return the subflow’s return value if it’s available. To reliably fetch across processes, enable result persistence on the subflows (e.g.,

persist_result=True

with GCS storage and a serializer). - Nested subflows: repeat the first approach by querying with each subflow’s

id

as the parent to crawl deeper levels. - Prefect Cloud: ensure your hook runs where

PREFECT_API_URL

and

PREFECT_API_KEY

are set (your worker environment), so

get_client()

can query the API. Do you want only direct children or all descendants? And do you need just states or also the subflow return values? I can tailor the snippet accordingly.

Short answer: not automatically. In a flow state-change hook you only get the parent’s final `State`; there’s no built-in list of subflows attached to it. Without the Prefect client, the only way to see subflow states in the hook is to capture and propagate that information yourself during the run (e.g., by calling subflows with

return_state=True

and returning a summary as part of the parent flow’s result). Here’s a pattern that needs no client: - Call each subflow with

return_state=True

- Build a lightweight summary (run_id, state, message, result if you want) - Return those summaries as part of the parent flow’s return value - In your completion hook, read them from

state.result()

Example:

from prefect import flow

@flow
def child_a(x: int) -> int:
    return x + 1

@flow
def child_b(y: int) -> int:
    return y * 2

def on_parent_completion(flow_obj, flow_run, state):
    payload = state.result()  # parent’s return value
    subflows = payload.get("subflows", [])
    print(f"Found {len(subflows)} subflows recorded by the parent")
    for sf in subflows:
        print(
            f"- {sf['name']} run_id={sf.get('run_id')} "
            f"state={sf.get('state_type')} message={sf.get('message')}"
        )
        if "result" in sf:
            print(f"  result: {sf['result']}")
        if "result_error" in sf:
            print(f"  result_error: {sf['result_error']}")

@flow(on_completion=[on_parent_completion])
def parent():
    subflow_summaries = []

    def run_and_record(f, *args, **kwargs):
        # force return_state so we can inspect it without the client
        st = f(*args, return_state=True, **kwargs)
        info = {
            "name": f.__name__,
            "run_id": st.state_details.flow_run_id,   # subflow run id
            "state_type": st.type.value,
            "message": st.message,
        }
        try:
            # For COMPLETED states this is the subflow's Python return value
            info["result"] = st.result()
        except Exception as e:
            # If the subflow failed and you still want to continue the parent
            info["result_error"] = str(e)
        subflow_summaries.append(info)

        # Preserve original semantics: return the subflow value to the parent
        # If the subflow failed, you can decide to raise here instead.
        return st.result(raise_on_failure=False)

    a = run_and_record(child_a, 1)
    b = run_and_record(child_b, 2)
    return {
        "final": a + b,
        "subflows": subflow_summaries,  # hook will read these
    }

if __name__ == "__main__":
    parent()

Why this works: - Hooks run in the same process as the flow, immediately after the state transition, so anything you include in the parent’s return value is accessible in the hook via

state.result()

. - By returning only simple summaries (str/ints/dicts), you avoid serialization issues that could arise if you tried to return raw

State

objects. If you need to discover subflows that you did not capture at runtime, you’ll need the Prefect client to query them. But if you can modify the parent flow, the pattern above keeps everything “direct Python” with no client calls.