Obsidian MCP Server

Instructions

Implementation Reference

• src/obsidian_mcp/vault.py:972-1021 (handler)

  Core handler function that executes the restore logic: locates the backup directory, finds all .md files recursively, restores them concurrently using asyncio.gather with async file I/O, preserves metadata, and returns list of restored paths.
  async def restore_batch_backup(self, backup_id: str) -> list[str]: """ Restore notes from a batch backup asynchronously. Args: backup_id: Backup ID (timestamp) to restore from Returns: List of restored note paths Raises: FileNotFoundError: If backup doesn't exist """ backup_dir = self.vault_path / ".batch_backups" / backup_id if not backup_dir.exists(): raise FileNotFoundError(f"Backup not found: {backup_id}") logger.info(f"Restoring batch backup {backup_id}...") # Get all backup files backup_files = list(backup_dir.rglob("*.md")) # Restore all files asynchronously async def restore_file(i: int, backup_file: Path) -> str: # Get relative path from backup directory rel_path = backup_file.relative_to(backup_dir) target_file = self.vault_path / rel_path # Ensure parent directory exists target_file.parent.mkdir(parents=True, exist_ok=True) # Restore file async with aiofiles.open(backup_file, "rb") as src: content = await src.read() async with aiofiles.open(target_file, "wb") as dst: await dst.write(content) # Preserve metadata shutil.copystat(backup_file, target_file) logger.debug(f"Restored ({i}): {rel_path}") return str(rel_path) # Run all restores concurrently restored = await asyncio.gather( *[restore_file(i, backup_file) for i, backup_file in enumerate(backup_files, 1)] ) logger.info(f"Completed batch restore: {backup_id} ({len(restored)} notes)") return list(restored)
• src/obsidian_mcp/server.py:1703-1736 (registration)

  MCP tool registration using @mcp.tool(name="restore_batch_backup") and wrapper handler function that validates input, delegates to ObsidianVault.restore_batch_backup, formats the result as a human-readable string list, and handles errors.
  @mcp.tool( name="restore_batch_backup", description="Restore notes from a batch backup (undo batch operation)", ) async def restore_batch_backup(backup_id: str) -> str: """ Restore notes from a batch backup. Args: backup_id: Backup ID (timestamp) to restore from Returns: Success message with restored note count """ if not backup_id or not backup_id.strip(): return "Error: Backup ID cannot be empty" context = _get_context() try: restored = await context.vault.restore_batch_backup(backup_id) result = [f"✅ Restored {len(restored)} notes from backup `{backup_id}`\n"] for path in restored: result.append(f"- `{path}`") return "\n".join(result) except FileNotFoundError as e: return f"Error: {e}" except Exception as e: logger.exception("Error restoring backup") return f"Error: {e}"
• src/obsidian_mcp/vault.py:917-970 (helper)

  Companion helper method create_batch_backup that creates the timestamped backup directories used by restore_batch_backup, enabling atomic batch operations with rollback support.
  async def create_batch_backup(self, relative_paths: list[str]) -> str: """ Create a backup of multiple notes asynchronously. Args: relative_paths: List of note paths to backup Returns: Backup ID (timestamp) for later restoration Raises: VaultSecurityError: If any path is invalid FileNotFoundError: If any note doesn't exist """ # Validate all paths first file_paths = [] for rel_path in relative_paths: file_path = self._validate_path(rel_path) if not file_path.exists(): raise FileNotFoundError(f"Note not found: {rel_path}") file_paths.append((rel_path, file_path)) # Create backup directory with timestamp backup_id = datetime.now().strftime("%Y%m%d_%H%M%S") backup_dir = self.vault_path / ".batch_backups" / backup_id backup_dir.mkdir(parents=True, exist_ok=True) logger.info(f"Creating batch backup {backup_id}: {len(relative_paths)} files...") # Copy all files to backup asynchronously async def copy_file(i: int, rel_path: str, file_path: Path) -> None: backup_file = backup_dir / rel_path backup_file.parent.mkdir(parents=True, exist_ok=True) # Use async file operations async with aiofiles.open(file_path, "rb") as src: content = await src.read() async with aiofiles.open(backup_file, "wb") as dst: await dst.write(content) # Preserve metadata shutil.copystat(file_path, backup_file) logger.debug(f"Backed up ({i}/{len(file_paths)}): {rel_path}") # Run all copies concurrently await asyncio.gather( *[ copy_file(i, rel_path, file_path) for i, (rel_path, file_path) in enumerate(file_paths, 1) ] ) logger.info(f"Completed batch backup: {backup_id} ({len(relative_paths)} notes)") return backup_id