Short answer: Yes. In Prefect 3.x you can load persisted results for any completed flow run or task run by fetching its final State from the API, then calling

state.result()

(sync) or

await state.aresult()

(async). This works outside of any running flow. Important prerequisites - Results must have been persisted at run time (e.g.,

@flow(persist_result=True)

or

@task(persist_result=True)

, or via config). - If a run did not persist its result, or the result expired/was deleted, loading will raise

prefect.exceptions.MissingResult

. How to enable persistence (if you haven’t already) - Per-flow/task:

@flow(persist_result=True)
def my_flow(): ...
@task(persist_result=True)
def my_task(): ...

- Or globally in config:

# .env or environment
PREFECT_RESULTS_PERSIST_BY_DEFAULT=true
# optional: choose storage and serializer
PREFECT_RESULTS_DEFAULT_STORAGE_BLOCK=s3-bucket/my-results
PREFECT_RESULTS_DEFAULT_SERIALIZER=json

Load a flow run’s result by flow_run_id (async)

import asyncio
from uuid import UUID
from prefect import get_client
from prefect.exceptions import MissingResult, UnfinishedRun

async def load_flow_result(flow_run_id: UUID):
    async with get_client() as client:
        # Fetch all states for this flow run
        states = await client.read_flow_run_states(flow_run_id=flow_run_id)
        if not states:
            raise UnfinishedRun("No states found for this flow run yet.")
        final_state = states[-1]  # typically the last state is the final state

        try:
            return await final_state.aresult(raise_on_failure=True)
        except MissingResult:
            raise MissingResult("Result was not persisted or has expired.")

# usage
# result = asyncio.run(load_flow_result(UUID("...")))

Load a task run’s result by task_run_id (async)

import asyncio
from uuid import UUID
from prefect import get_client
from prefect.exceptions import MissingResult, UnfinishedRun

async def load_task_result(task_run_id: UUID):
    async with get_client() as client:
        states = await client.read_task_run_states(task_run_id=task_run_id)
        if not states:
            raise UnfinishedRun("No states found for this task run yet.")

        # Prefer a completed state; fall back to the last
        completed = next((s for s in reversed(states) if s.is_completed()), states[-1])

        try:
            return await completed.aresult(raise_on_failure=True)
        except MissingResult:
            raise MissingResult("Result was not persisted or has expired.")

# usage
# result = asyncio.run(load_task_result(UUID("...")))

Synchronous variants

from uuid import UUID
from prefect import get_client
from prefect.exceptions import MissingResult, UnfinishedRun

def load_flow_result_sync(flow_run_id: UUID):
    with get_client(sync_client=True) as client:
        states = client.read_flow_run_states(flow_run_id=flow_run_id)
        if not states:
            raise UnfinishedRun("No states found.")
        final_state = states[-1]
        try:
            return final_state.result(raise_on_failure=True)
        except MissingResult:
            raise MissingResult("Result was not persisted or has expired.")

def load_task_result_sync(task_run_id: UUID):
    with get_client(sync_client=True) as client:
        states = client.read_task_run_states(task_run_id=task_run_id)
        if not states:
            raise UnfinishedRun("No states found.")
        completed = next((s for s in reversed(states) if s.is_completed()), states[-1])
        try:
            return completed.result(raise_on_failure=True)
        except MissingResult:
            raise MissingResult("Result was not persisted or has expired.")

Notes and tips -

state.result()

and

state.aresult()

work both in-process and out-of-process: they inspect

state.data

and, when needed, transparently load from your configured result storage (local/S3/etc.) using the correct serializer.

- If you expect flakiness reading from external storage, you can pass

retry_result_failure=True

(default) to retry transient read failures. - If a state is failed/crashed and you still want to inspect the (possibly partial) result, pass

raise_on_failure=False

. Lower-level option (advanced) - If you prefer to bypass the helper on

State

, you can read directly from storage via

prefect.results.ResultStore

using the

ResultRecordMetadata

found in

state.data

. In most cases,

state.result()

is the recommended, simpler approach. If you want, share a specific flow_run_id or task_run_id and I can help you test one of these snippets against your environment.