list_flow_runs
Retrieve flow runs from Prefect API with optional filtering by flow ID, limit, and offset parameters.
Instructions
Get a list of flow runs from the Prefect API.
Args:
limit: Maximum number of flow runs to return (default 20).
offset: Number of flow runs to skip (default 0).
flow_id: Optional ID of the flow to filter runs by.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No | ||
| offset | No | ||
| flow_id | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- prefect_mcp_server_pkg/server.py:163-188 (handler)The handler function for the 'list_flow_runs' MCP tool, registered via @mcp.tool() decorator. It lists flow runs from Prefect API, optionally filtered by flow_id, with pagination support.
@mcp.tool() async def list_flow_runs( ctx: Context, limit: int = 20, offset: int = 0, flow_id: Optional[str] = None ) -> Dict[str, Any]: """Get a list of flow runs from the Prefect API. Args: limit: Maximum number of flow runs to return (default 20). offset: Number of flow runs to skip (default 0). flow_id: Optional ID of the flow to filter runs by. """ filter_dict = {} if flow_id: filter_dict["flow_id"] = {"equals": flow_id} async with get_client() as client: flow_run_filter = FlowRunFilter(**filter_dict) if filter_dict else None flow_runs = await client.read_flow_runs( flow_run_filter=flow_run_filter, limit=limit, offset=offset ) return { "flow_runs": [run.model_dump() for run in flow_runs], "count": len(flow_runs), }