search_flow_runs_by_state
Find Prefect workflow runs by their execution state to monitor progress, identify issues, or analyze performance. Filter by state type or name to focus on specific run categories.
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 |
|---|---|---|---|
| state_type | No | ||
| state_name | No | ||
| limit | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- prefect_mcp_server_pkg/server.py:190-220 (handler)The handler function implementing the 'search_flow_runs_by_state' tool. It is decorated with @mcp.tool() for automatic registration in the MCP server. The function builds a FlowRunFilter based on state_type or state_name, queries the Prefect client for matching flow runs, and returns a dictionary with the list of flow runs and count.
@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), }