MCP Filesystem Server

Instructions

Args:
    path: Starting directory
    recursive: Whether to search subdirectories
    min_size: Minimum file size to consider (bytes)
    exclude_patterns: Optional patterns to exclude
    max_files: Maximum number of files to scan
    format: Output format ('text' or 'json')
    ctx: MCP context

Returns:
    Duplicate file information

Implementation Reference

• mcp_filesystem/advanced.py:381-507 (handler)

  Core handler implementing the duplicate detection: validates path, scans directory recursively, groups files by size, computes MD5 hashes for potential duplicates, returns dict of hash to file paths.

  async def find_duplicate_files(
      self,
      root_path: Union[str, Path],
      recursive: bool = True,
      min_size: int = 1,
      exclude_patterns: Optional[List[str]] = None,
      max_files: int = 1000,
  ) -> Dict[str, List[str]]:
      """Find duplicate files by comparing file sizes and contents.
  
      Args:
          root_path: Starting directory
          recursive: Whether to search subdirectories
          min_size: Minimum file size to consider (bytes)
          exclude_patterns: Optional patterns to exclude
          max_files: Maximum number of files to scan
  
      Returns:
          Dictionary mapping file hash to list of identical files
  
      Raises:
          ValueError: If root_path is outside allowed directories
      """
      import hashlib
  
      abs_path, allowed = await self.validator.validate_path(root_path)
      if not allowed:
          raise ValueError(f"Path outside allowed directories: {root_path}")
  
      if not abs_path.is_dir():
          raise ValueError(f"Not a directory: {root_path}")
  
      # Compile exclude patterns if provided
      exclude_regexes = []
      if exclude_patterns:
          for exclude in exclude_patterns:
              try:
                  exclude_regexes.append(re.compile(exclude))
              except re.error:
                  logger.warning(f"Invalid exclude pattern: {exclude}")
  
      # First, group files by size
      size_groups: Dict[int, List[Path]] = {}
      files_processed = 0
  
      async def scan_for_sizes(dir_path: Path) -> None:
          nonlocal files_processed
  
          if files_processed >= max_files:
              return
  
          try:
              entries = await anyio.to_thread.run_sync(list, dir_path.iterdir())
  
              for entry in entries:
                  if files_processed >= max_files:
                      return
  
                  # Skip if matched by exclude pattern
                  path_str = str(entry)
                  excluded = False
                  for exclude_re in exclude_regexes:
                      if exclude_re.search(path_str):
                          excluded = True
                          break
  
                  if excluded:
                      continue
  
                  try:
                      if entry.is_file():
                          size = entry.stat().st_size
                          if size >= min_size:
                              if size not in size_groups:
                                  size_groups[size] = []
                              size_groups[size].append(entry)
                              files_processed += 1
  
                      elif entry.is_dir() and recursive:
                          # Check if this path is still allowed
                          (
                              entry_abs,
                              entry_allowed,
                          ) = await self.validator.validate_path(entry)
                          if entry_allowed:
                              await scan_for_sizes(entry)
  
                  except (PermissionError, FileNotFoundError):
                      # Skip entries we can't access
                      pass
  
          except (PermissionError, FileNotFoundError):
              # Skip directories we can't access
              pass
  
      await scan_for_sizes(abs_path)
  
      # Now, for each size group with multiple files, compute and compare hashes
      duplicates: Dict[str, List[str]] = {}
  
      for size, files in size_groups.items():
          if len(files) < 2:
              continue
  
          # Group files by hash
          hash_groups: Dict[str, List[Path]] = {}
  
          for file_path in files:
              try:
                  # Compute file hash
                  file_bytes = await anyio.to_thread.run_sync(file_path.read_bytes)
                  file_hash = hashlib.md5(file_bytes).hexdigest()
  
                  if file_hash not in hash_groups:
                      hash_groups[file_hash] = []
                  hash_groups[file_hash].append(file_path)
  
              except (PermissionError, FileNotFoundError):
                  # Skip files we can't access
                  pass
  
          # Add duplicate groups to results
          for file_hash, hash_files in hash_groups.items():
              if len(hash_files) >= 2:
                  duplicates[file_hash] = [str(f) for f in hash_files]
  
      return duplicates

• mcp_filesystem/server.py:515-565 (registration)

  MCP tool registration via @mcp.tool() decorator. Wrapper function that delegates to Advanced component's handler and formats results for MCP response (text or JSON). Defines tool schema via parameters and docstring.

  @mcp.tool()
  async def find_duplicate_files(
      path: str,
      ctx: Context,
      recursive: bool = True,
      min_size: int = 1,
      exclude_patterns: Optional[List[str]] = None,
      max_files: int = 1000,
      format: str = "text",
  ) -> str:
      """Find duplicate files by comparing file sizes and contents.
  
      Args:
          path: Starting directory
          recursive: Whether to search subdirectories
          min_size: Minimum file size to consider (bytes)
          exclude_patterns: Optional patterns to exclude
          max_files: Maximum number of files to scan
          format: Output format ('text' or 'json')
          ctx: MCP context
  
      Returns:
          Duplicate file information
      """
      try:
          components = get_components()
          duplicates = await components["advanced"].find_duplicate_files(
              path, recursive, min_size, exclude_patterns, max_files
          )
  
          if format.lower() == "json":
              return json.dumps(duplicates, indent=2)
  
          # Format as text
          if not duplicates:
              return "No duplicate files found"
  
          lines = []
          for file_hash, files in duplicates.items():
              lines.append(f"Hash: {file_hash}")
              for file_path in files:
                  lines.append(f"  {file_path}")
              lines.append("")
  
          return f"Found {len(duplicates)} sets of duplicate files:\n\n" + "\n".join(
              lines
          )
  
      except Exception as e:
          return f"Error finding duplicate files: {str(e)}"