get_flow_runs
Retrieve and filter workflow execution records by criteria like status, flow name, tags, or time range to monitor and analyze Prefect flow runs.
Instructions
Get a list of flow runs with optional filtering.
Args: limit: Maximum number of flow runs to return offset: Number of flow runs to skip flow_name: Filter by flow name state_type: Filter by state type (e.g., "RUNNING", "COMPLETED", "FAILED") state_name: Filter by state name deployment_id: Filter by deployment ID tags: Filter by tags start_time_before: ISO formatted datetime string start_time_after: ISO formatted datetime string
Returns: A list of flow runs with their details
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| deployment_id | No | ||
| flow_name | No | ||
| limit | No | ||
| offset | No | ||
| start_time_after | No | ||
| start_time_before | No | ||
| state_name | No | ||
| state_type | No | ||
| tags | No |
Implementation Reference
- src/mcp_prefect/flow_run.py:17-85 (handler)The handler function implementing the 'get_flow_runs' tool logic. It queries the Prefect client for flow runs with applied filters, enriches with UI URLs, and returns as text content.
@mcp.tool async def get_flow_runs( limit: Optional[int] = None, offset: Optional[int] = None, flow_name: Optional[str] = None, state_type: Optional[str] = None, state_name: Optional[str] = None, deployment_id: Optional[str] = None, tags: Optional[List[str]] = None, start_time_before: Optional[str] = None, start_time_after: Optional[str] = None, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: """ Get a list of flow runs with optional filtering. Args: limit: Maximum number of flow runs to return offset: Number of flow runs to skip flow_name: Filter by flow name state_type: Filter by state type (e.g., "RUNNING", "COMPLETED", "FAILED") state_name: Filter by state name deployment_id: Filter by deployment ID tags: Filter by tags start_time_before: ISO formatted datetime string start_time_after: ISO formatted datetime string Returns: A list of flow runs with their details """ async with get_client() as client: # Build filter parameters filters = {} if flow_name: filters["flow_name"] = {"like_": f"%{flow_name}%"} if state_type: filters["state"] = {"type": {"any_": [state_type.upper()]}} if state_name: filters["state"] = {"name": {"any_": [state_name]}} if deployment_id: filters["deployment_id"] = {"eq_": UUID(deployment_id)} if tags: filters["tags"] = {"all_": tags} if start_time_after: filters["start_time"] = {"ge_": start_time_after} if start_time_before: if "start_time" in filters: filters["start_time"]["le_"] = start_time_before else: filters["start_time"] = {"le_": start_time_before} flow_runs = await client.read_flow_runs( limit=limit, offset=offset, **filters ) # Add UI links to each flow run flow_runs_result = { "flow_runs": [ { **flow_run.dict(), "ui_url": get_flow_run_url(str(flow_run.id)) } for flow_run in flow_runs ] } return [types.TextContent(type="text", text=str(flow_runs_result))] - src/mcp_prefect/main.py:42-44 (registration)The import statement in main.py that loads the flow_run module, triggering registration of the 'get_flow_runs' tool via its @mcp.tool decorator.
if APIType.FLOW_RUN.value in apis: info("Loading Flow Run API...") from . import flow_run - src/mcp_prefect/flow_run.py:12-15 (helper)Helper function used by the handler to generate UI URLs for flow runs.
def get_flow_run_url(flow_run_id: str) -> str: base_url = PREFECT_API_URL.replace("/api", "") return f"{base_url}/flow-runs/{flow_run_id}" - src/mcp_prefect/flow_run.py:17-18 (registration)The @mcp.tool decorator that registers the get_flow_runs function as an MCP tool.
@mcp.tool async def get_flow_runs(