Moondream MCP Server

  • src
  • mcp_server_everything_search
"""Everything SDK wrapper class.""" import ctypes import datetime import struct import sys from typing import Any, List from pydantic import BaseModel # Everything SDK constants EVERYTHING_OK = 0 EVERYTHING_ERROR_MEMORY = 1 EVERYTHING_ERROR_IPC = 2 EVERYTHING_ERROR_REGISTERCLASSEX = 3 EVERYTHING_ERROR_CREATEWINDOW = 4 EVERYTHING_ERROR_CREATETHREAD = 5 EVERYTHING_ERROR_INVALIDINDEX = 6 EVERYTHING_ERROR_INVALIDCALL = 7 # Request flags EVERYTHING_REQUEST_FILE_NAME = 0x00000001 EVERYTHING_REQUEST_PATH = 0x00000002 EVERYTHING_REQUEST_FULL_PATH_AND_FILE_NAME = 0x00000004 EVERYTHING_REQUEST_EXTENSION = 0x00000008 EVERYTHING_REQUEST_SIZE = 0x00000010 EVERYTHING_REQUEST_DATE_CREATED = 0x00000020 EVERYTHING_REQUEST_DATE_MODIFIED = 0x00000040 EVERYTHING_REQUEST_DATE_ACCESSED = 0x00000080 EVERYTHING_REQUEST_ATTRIBUTES = 0x00000100 EVERYTHING_REQUEST_FILE_LIST_FILE_NAME = 0x00000200 EVERYTHING_REQUEST_RUN_COUNT = 0x00000400 EVERYTHING_REQUEST_DATE_RUN = 0x00000800 EVERYTHING_REQUEST_DATE_RECENTLY_CHANGED = 0x00001000 EVERYTHING_REQUEST_HIGHLIGHTED_FILE_NAME = 0x00002000 EVERYTHING_REQUEST_HIGHLIGHTED_PATH = 0x00004000 EVERYTHING_REQUEST_HIGHLIGHTED_FULL_PATH_AND_FILE_NAME = 0x00008000 # Sort options EVERYTHING_SORT_NAME_ASCENDING = 1 EVERYTHING_SORT_NAME_DESCENDING = 2 EVERYTHING_SORT_PATH_ASCENDING = 3 EVERYTHING_SORT_PATH_DESCENDING = 4 EVERYTHING_SORT_SIZE_ASCENDING = 5 EVERYTHING_SORT_SIZE_DESCENDING = 6 EVERYTHING_SORT_EXTENSION_ASCENDING = 7 EVERYTHING_SORT_EXTENSION_DESCENDING = 8 EVERYTHING_SORT_TYPE_NAME_ASCENDING = 9 EVERYTHING_SORT_TYPE_NAME_DESCENDING = 10 EVERYTHING_SORT_DATE_CREATED_ASCENDING = 11 EVERYTHING_SORT_DATE_CREATED_DESCENDING = 12 EVERYTHING_SORT_DATE_MODIFIED_ASCENDING = 13 EVERYTHING_SORT_DATE_MODIFIED_DESCENDING = 14 EVERYTHING_SORT_ATTRIBUTES_ASCENDING = 15 EVERYTHING_SORT_ATTRIBUTES_DESCENDING = 16 EVERYTHING_SORT_FILE_LIST_FILENAME_ASCENDING = 17 EVERYTHING_SORT_FILE_LIST_FILENAME_DESCENDING = 18 EVERYTHING_SORT_RUN_COUNT_ASCENDING = 19 EVERYTHING_SORT_RUN_COUNT_DESCENDING = 20 EVERYTHING_SORT_DATE_RECENTLY_CHANGED_ASCENDING = 21 EVERYTHING_SORT_DATE_RECENTLY_CHANGED_DESCENDING = 22 EVERYTHING_SORT_DATE_ACCESSED_ASCENDING = 23 EVERYTHING_SORT_DATE_ACCESSED_DESCENDING = 24 EVERYTHING_SORT_DATE_RUN_ASCENDING = 25 EVERYTHING_SORT_DATE_RUN_DESCENDING = 26 # Windows time conversion constants WINDOWS_TICKS = int(1/10**-7) WINDOWS_EPOCH = datetime.datetime.strptime('1601-01-01 00:00:00', '%Y-%m-%d %H:%M:%S') POSIX_EPOCH = datetime.datetime.strptime('1970-01-01 00:00:00', '%Y-%m-%d %H:%M:%S') EPOCH_DIFF = (POSIX_EPOCH - WINDOWS_EPOCH).total_seconds() WINDOWS_TICKS_TO_POSIX_EPOCH = EPOCH_DIFF * WINDOWS_TICKS class SearchResult(BaseModel): """Model for search results.""" path: str filename: str extension: str | None = None size: int created: str | None = None modified: str | None = None accessed: str | None = None attributes: int | None = None run_count: int | None = None highlighted_filename: str | None = None highlighted_path: str | None = None class EverythingError(Exception): """Custom exception for Everything SDK errors.""" def __init__(self, error_code: int): self.error_code = error_code super().__init__(self._get_error_message()) def _get_error_message(self) -> str: error_messages = { EVERYTHING_ERROR_MEMORY: "Failed to allocate memory", EVERYTHING_ERROR_IPC: "IPC failed (Everything service not running?)", EVERYTHING_ERROR_REGISTERCLASSEX: "Failed to register window class", EVERYTHING_ERROR_CREATEWINDOW: "Failed to create window", EVERYTHING_ERROR_CREATETHREAD: "Failed to create thread", EVERYTHING_ERROR_INVALIDINDEX: "Invalid index", EVERYTHING_ERROR_INVALIDCALL: "Invalid call" } return error_messages.get(self.error_code, f"Unknown error: {self.error_code}") class EverythingSDK: """Wrapper for Everything SDK functionality.""" def __init__(self, dll_path: str): """Initialize Everything SDK with the specified DLL path.""" try: self.dll = ctypes.WinDLL(dll_path) self._configure_dll() except Exception as e: print(f"Failed to load Everything SDK DLL: {e}", file=sys.stderr) raise def _configure_dll(self): """Configure DLL function signatures.""" # Search configuration self.dll.Everything_SetSearchW.argtypes = [ctypes.c_wchar_p] self.dll.Everything_SetMatchPath.argtypes = [ctypes.c_bool] self.dll.Everything_SetMatchCase.argtypes = [ctypes.c_bool] self.dll.Everything_SetMatchWholeWord.argtypes = [ctypes.c_bool] self.dll.Everything_SetRegex.argtypes = [ctypes.c_bool] self.dll.Everything_SetMax.argtypes = [ctypes.c_uint] self.dll.Everything_SetSort.argtypes = [ctypes.c_uint] self.dll.Everything_SetRequestFlags.argtypes = [ctypes.c_uint] # Query function self.dll.Everything_QueryW.argtypes = [ctypes.c_bool] self.dll.Everything_QueryW.restype = ctypes.c_bool # Result getters self.dll.Everything_GetNumResults.restype = ctypes.c_uint self.dll.Everything_GetLastError.restype = ctypes.c_uint self.dll.Everything_GetResultFileNameW.argtypes = [ctypes.c_uint] self.dll.Everything_GetResultFileNameW.restype = ctypes.c_wchar_p self.dll.Everything_GetResultExtensionW.argtypes = [ctypes.c_uint] self.dll.Everything_GetResultExtensionW.restype = ctypes.c_wchar_p self.dll.Everything_GetResultPathW.argtypes = [ctypes.c_uint] self.dll.Everything_GetResultPathW.restype = ctypes.c_wchar_p self.dll.Everything_GetResultFullPathNameW.argtypes = [ ctypes.c_uint, ctypes.c_wchar_p, ctypes.c_uint ] self.dll.Everything_GetResultDateCreated.argtypes = [ ctypes.c_uint, ctypes.POINTER(ctypes.c_ulonglong) ] self.dll.Everything_GetResultDateModified.argtypes = [ ctypes.c_uint, ctypes.POINTER(ctypes.c_ulonglong) ] self.dll.Everything_GetResultDateAccessed.argtypes = [ ctypes.c_uint, ctypes.POINTER(ctypes.c_ulonglong) ] self.dll.Everything_GetResultSize.argtypes = [ ctypes.c_uint, ctypes.POINTER(ctypes.c_ulonglong) ] self.dll.Everything_GetResultAttributes.argtypes = [ctypes.c_uint] self.dll.Everything_GetResultRunCount.argtypes = [ctypes.c_uint] self.dll.Everything_GetResultHighlightedFileNameW.argtypes = [ctypes.c_uint] self.dll.Everything_GetResultHighlightedFileNameW.restype = ctypes.c_wchar_p self.dll.Everything_GetResultHighlightedPathW.argtypes = [ctypes.c_uint] self.dll.Everything_GetResultHighlightedPathW.restype = ctypes.c_wchar_p def _check_error(self): """Check for Everything SDK errors and raise appropriate exception.""" error_code = self.dll.Everything_GetLastError() if error_code != EVERYTHING_OK: raise EverythingError(error_code) def _get_time(self, filetime: int) -> datetime.datetime: """Convert Windows filetime to Python datetime.""" microsecs = (filetime - WINDOWS_TICKS_TO_POSIX_EPOCH) / WINDOWS_TICKS return datetime.datetime.fromtimestamp(microsecs) def search_files( self, query: str, max_results: int = 100, match_path: bool = False, match_case: bool = False, match_whole_word: bool = False, match_regex: bool = False, sort_by: int = EVERYTHING_SORT_NAME_ASCENDING, request_flags: int | None = None ) -> List[SearchResult]: """Perform file search using Everything SDK.""" print(f"Debug: Setting up search with query: {query}", file=sys.stderr) # Set up search parameters self.dll.Everything_SetSearchW(query) self.dll.Everything_SetMatchPath(match_path) self.dll.Everything_SetMatchCase(match_case) self.dll.Everything_SetMatchWholeWord(match_whole_word) self.dll.Everything_SetRegex(match_regex) self.dll.Everything_SetMax(max_results) self.dll.Everything_SetSort(sort_by) # Set request flags if request_flags is None: request_flags = ( EVERYTHING_REQUEST_FILE_NAME | EVERYTHING_REQUEST_PATH | EVERYTHING_REQUEST_EXTENSION | EVERYTHING_REQUEST_SIZE | EVERYTHING_REQUEST_DATE_CREATED | EVERYTHING_REQUEST_DATE_MODIFIED | EVERYTHING_REQUEST_DATE_ACCESSED | EVERYTHING_REQUEST_ATTRIBUTES | EVERYTHING_REQUEST_RUN_COUNT | EVERYTHING_REQUEST_HIGHLIGHTED_FILE_NAME | EVERYTHING_REQUEST_HIGHLIGHTED_PATH ) self.dll.Everything_SetRequestFlags(request_flags) # Execute search print("Debug: Executing search query", file=sys.stderr) if not self.dll.Everything_QueryW(True): self._check_error() raise RuntimeError("Search query failed") # Get results print("Debug: Getting search results", file=sys.stderr) num_results = min(self.dll.Everything_GetNumResults(), max_results) results = [] filename_buffer = ctypes.create_unicode_buffer(260) date_created = ctypes.c_ulonglong() date_modified = ctypes.c_ulonglong() date_accessed = ctypes.c_ulonglong() file_size = ctypes.c_ulonglong() for i in range(num_results): try: self.dll.Everything_GetResultFullPathNameW(i, filename_buffer, 260) # Get timestamps self.dll.Everything_GetResultDateCreated(i, date_created) self.dll.Everything_GetResultDateModified(i, date_modified) self.dll.Everything_GetResultDateAccessed(i, date_accessed) self.dll.Everything_GetResultSize(i, file_size) # Get other attributes filename = self.dll.Everything_GetResultFileNameW(i) extension = self.dll.Everything_GetResultExtensionW(i) attributes = self.dll.Everything_GetResultAttributes(i) run_count = self.dll.Everything_GetResultRunCount(i) highlighted_filename = self.dll.Everything_GetResultHighlightedFileNameW(i) highlighted_path = self.dll.Everything_GetResultHighlightedPathW(i) results.append(SearchResult( path=filename_buffer.value, filename=filename, extension=extension, size=file_size.value, created=self._get_time(date_created.value).isoformat() if date_created.value else None, modified=self._get_time(date_modified.value).isoformat() if date_modified.value else None, accessed=self._get_time(date_accessed.value).isoformat() if date_accessed.value else None, attributes=attributes, run_count=run_count, highlighted_filename=highlighted_filename, highlighted_path=highlighted_path )) except Exception as e: print(f"Debug: Error processing result {i}: {e}", file=sys.stderr) continue print("Debug: Resetting Everything SDK", file=sys.stderr) self.dll.Everything_Reset() return results