get_flow_runs_by_flow
Retrieve flow runs for a specific workflow in Prefect, with options to filter by state type, limit results, and paginate through large datasets.
Instructions
Get flow runs for a specific flow.
Args: flow_id: The flow UUID limit: Maximum number of flow runs to return offset: Number of flow runs to skip state_type: Filter by state type (e.g., "RUNNING", "COMPLETED", "FAILED")
Returns: A list of flow runs for the specified flow
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| flow_id | Yes | ||
| limit | No | ||
| offset | No | ||
| state_type | No |
Implementation Reference
- src/mcp_prefect/flow_run.py:110-152 (handler)The main handler function for the 'get_flow_runs_by_flow' tool. It uses the Prefect client to query flow runs filtered by flow_id and optional state_type, adds UI links, and returns the result as text content.
@mcp.tool async def get_flow_runs_by_flow( flow_id: str, limit: Optional[int] = None, offset: Optional[int] = None, state_type: Optional[str] = None, ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]: """ Get flow runs for a specific flow. Args: flow_id: The flow UUID limit: Maximum number of flow runs to return offset: Number of flow runs to skip state_type: Filter by state type (e.g., "RUNNING", "COMPLETED", "FAILED") Returns: A list of flow runs for the specified flow """ async with get_client() as client: # Build filter parameters filters = {"flow_id": {"eq_": UUID(flow_id)}} if state_type: filters["state"] = {"type": {"any_": [state_type.upper()]}} 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)Conditional import of the flow_run module in main.py, which triggers registration of all @mcp.tool functions in flow_run.py, including 'get_flow_runs_by_flow'.
if APIType.FLOW_RUN.value in apis: info("Loading Flow Run API...") from . import flow_run - src/mcp_prefect/flow_run.py:12-14 (helper)Helper function used by get_flow_runs_by_flow (and other flow run tools) 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}"