search_flow_runs_by_state
Retrieve workflow flow runs by their state, such as 'COMPLETED' or 'FAILED', using optional filters and a customizable limit to streamline automation monitoring and analysis.
Instructions
Search for flow runs by state.
Args:
state_type: Optional state type (e.g., "COMPLETED", "FAILED", "CRASHED").
state_name: Optional state name (e.g., "Completed", "Failed").
limit: Maximum number of flow runs to return (default 20).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No | ||
| state_name | No | ||
| state_type | No |
Implementation Reference
- prefect_mcp_server_pkg/server.py:190-220 (handler)The handler function for the 'search_flow_runs_by_state' tool. It is decorated with @mcp.tool() for registration with the FastMCP server. Searches Prefect flow runs filtered by state type or name using the Prefect client API, returning a list of matching flow runs.@mcp.tool() async def search_flow_runs_by_state( ctx: Context, state_type: Optional[str] = None, state_name: Optional[str] = None, limit: int = 20, ) -> Dict[str, Any]: """Search for flow runs by state. Args: state_type: Optional state type (e.g., "COMPLETED", "FAILED", "CRASHED"). state_name: Optional state name (e.g., "Completed", "Failed"). limit: Maximum number of flow runs to return (default 20). """ filter_dict = {} if state_type: filter_dict["state"] = {"type": {"equals": state_type}} if state_name: filter_dict["state"] = {"name": {"equals": state_name}} 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 ) return { "flow_runs": [run.model_dump() for run in flow_runs], "count": len(flow_runs), }