Prefect MCP Server

Instructions

Args:
    filter_criteria: Dictionary with filter criteria according to Prefect API.
                     Example: {"flows": {"tags": {"all_": ["production"]}}}

Implementation Reference

• prefect_mcp_server_pkg/server.py:391-402 (handler)

  The main handler function for the 'filter_flows' tool. It is registered via the @mcp.tool() decorator and implements filtering of Prefect flows using the provided filter criteria with the Prefect client.

  @mcp.tool()
  async def filter_flows(ctx: Context, filter_criteria: Dict[str, Any]) -> Dict[str, Any]:
      """Filter flows based on specified criteria.
  
      Args:
          filter_criteria: Dictionary with filter criteria according to Prefect API.
                           Example: {"flows": {"tags": {"all_": ["production"]}}}
      """
      async with get_client() as client:
          flow_filter = FlowFilter(**filter_criteria)
          flows = await client.read_flows(flow_filter=flow_filter)
          return {"flows": [flow.model_dump() for flow in flows]}