Jashaul
04/17/2026, 2:23 AMMarvin
04/17/2026, 2:23 AMMarvin
04/17/2026, 2:26 AMstate.result(), and that only works for terminal states (COMPLETED/FAILED/CRASHED/CANCELLED). For non-terminal states like PENDING or RUNNING, Prefect raises UnfinishedRun: Run is in PENDING state, its result is not available.
How to resolve
- If you have a flow_run_id:
from prefect.flow_runs import wait_for_flow_run
flow_run = await wait_for_flow_run(flow_run_id) # waits until terminal
result = await flow_run.state.result()
- If you have a task_run_id:
from prefect.task_runs import TaskRunWaiter
state = await TaskRunWaiter.wait_for_task_run(task_run_id) # waits until terminal
result = await state.result()
- From inside a flow/task (recommended):
- Synchronous flow:
from prefect import flow
@flow
def my_flow():
return "hello"
state = my_flow(return_state=True)
value = state.result() # "hello"
- Async task submission:
from prefect import flow, task
@task
async def my_task():
return 42
@flow
async def my_flow():
state = await my_task.submit(return_state=True)
value = await state.result()
return value
Ensure results are persisted if you’re trying to load them later (e.g., from the API after the run finishes). Otherwise you’ll hit a MissingResult even after completion:
from prefect import flow, task
@flow(persist_result=True)
def my_flow():
return "done"
@task(persist_result=True)
def my_task():
return 123
If you’re running on ephemeral infrastructure (Docker/Kubernetes), configure remote result storage (e.g., S3/GCS/Azure) on the flow/task to make results retrievable across workers.
A couple clarifications that will help me tailor the fix:
- Which Prefect version are you on (assuming 3.x unless you say otherwise)?
- What object are you calling get_result on (a State, a flow run view/id, a task run view/id)?
- Can you share a minimal snippet showing how you launch the run and where you call get_result?
If you paste a short code sample, I can suggest the exact change.