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}"
```

Prefect MCP Server