Hui Wu
03/26/2026, 12:10 PMMarvin
03/26/2026, 12:10 PMMarvin
03/26/2026, 12:12 PMfrom uuid import UUID
from prefect import get_client
from prefect.client.schemas.filters import TaskRunFilter, TaskRunFilterFlowRunId
async def get_current_states(flow_run_id: UUID):
async with get_client() as client:
# 1) Flow run (includes current state)
flow_run = await client.read_flow_run(flow_run_id)
# 2) All task runs in this flow run (each includes its current state)
task_runs = await client.read_task_runs(
task_run_filter=TaskRunFilter(
flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id])
)
)
return {
"flow_run_state": flow_run.state,
"task_run_states": {tr.id: tr.state for tr in task_runs},
}
If you need state history, call:
- await client.read_flow_run_states(flow_run_id) for the flow run
- await client.read_task_run_states(task_run_id) for each task run
Single HTTP request option (graph endpoint):
- There’s a server endpoint that returns a graph of the flow run including task runs and their state information. There isn’t a dedicated SDK wrapper, but you can use the client’s low-level `request()`:
from uuid import UUID
from prefect import get_client
async def get_run_graph(flow_run_id: UUID):
async with get_client() as client:
resp = await client.request(
"GET",
"/flow_runs/{id}/graph-v2",
path_params={"id": str(flow_run_id)},
)
data = resp.json()
return data # includes nodes for the flow run and task runs with state details
Clarifying question: Do you need just the current states, or full state histories for each run? If it’s just current states, the two-call pattern above is the simplest and most efficient. If you must do it in a single HTTP request, the graph endpoint is the closest fit.