list_flow_runs
Retrieve flow runs from the Prefect API with customizable filters. Use limit, offset, and flow_id parameters to manage and query workflow execution data effectively.
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 |
|---|---|---|---|
| flow_id | No | ||
| limit | No | ||
| offset | No |
Implementation Reference
- prefect_mcp_server_pkg/server.py:163-188 (handler)The handler function implementing the 'list_flow_runs' MCP tool. It is registered via the @mcp.tool() decorator and fetches flow runs from the Prefect API using the prefect client, with optional filtering by flow_id.@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), }