Obsidian MCP Server

Instructions

Implementation Reference

• src/obsidian_mcp/server.py:1626-1701 (handler)

  Main handler function for the batch_append_notes tool. Performs batch appends to multiple notes with automatic backup creation, concurrent execution, error handling with rollback, and confirmation safety checks.
  name="batch_append_notes", description="Append content to multiple notes atomically", ) async def batch_append_notes( appends: list[NoteAppend], confirm: bool = False, ) -> str: """ Append content to multiple notes. Args: appends: List of NoteAppend objects with path and content confirm: Must be true to apply changes Returns: Success message with append summary """ if not appends: return "Error: No appends provided" # Check batch size limit if len(appends) > MAX_BATCH_SIZE: return ( f"Error: Batch size ({len(appends)}) exceeds maximum ({MAX_BATCH_SIZE}).\n" f"Split into smaller batches to avoid server timeouts." ) logger.info(f"Starting batch_append_notes: {len(appends)} notes") # Extract paths (Pydantic already validated) paths = [append.path for append in appends] if not confirm: return ( f"Error: Batch append to {len(appends)} notes requires explicit confirmation. " f"Set confirm=true to proceed." ) context = _get_context() try: # Create backup (async) backup_id = await context.vault.create_batch_backup(paths) # Apply all appends appended = [] failed = [] for append in appends: try: await context.vault.append_to_note(append.path, append.content) appended.append(append.path) except Exception as e: failed.append((append.path, str(e))) # Rollback on failure (async) if failed: await context.vault.restore_batch_backup(backup_id) result = ["❌ Batch append failed - all changes rolled back\n"] result.append("**Failed appends:**") for path, error in failed: result.append(f"- `{path}`: {error}") return "\n".join(result) # Success logger.info(f"Completed batch_append_notes: {len(appended)} notes updated successfully") result = [f"✅ Appended to {len(appended)} notes\n"] for path in appended: result.append(f"- `{path}`") result.append(f"\n**Backup:** `.batch_backups/{backup_id}/`") return "\n".join(result) except Exception as e: logger.exception("Error in batch append") return f"Error: {e}"
• src/obsidian_mcp/server.py:114-119 (schema)

  Pydantic schema defining the input structure for each append operation in the batch: path and content.
  class NoteAppend(BaseModel): """Schema for appending to a single note.""" path: str = Field(description="Relative path to the note") content: str = Field(description="Content to append to the note")
• src/obsidian_mcp/vault.py:551-582 (helper)

  Supporting method in ObsidianVault class that implements the core single-note append logic, reading existing content, ensuring newline separation, and asynchronously writing the updated content.
  async def append_to_note(self, relative_path: str, content: str) -> None: """ Append content to an existing note. Args: relative_path: Path to the note content: Content to append Raises: VaultSecurityError: If path is invalid FileNotFoundError: If note doesn't exist """ file_path = self._validate_path(relative_path) if not file_path.exists(): raise FileNotFoundError(f"Note not found: {relative_path}") # Read existing content async with aiofiles.open(file_path, encoding="utf-8") as f: existing = await f.read() # Append new content (with newline separator if needed) if not existing.endswith("\n"): existing += "\n" existing += content # Write back async with aiofiles.open(file_path, "w", encoding="utf-8") as f: await f.write(existing) logger.info(f"Appended to note: {relative_path}")
• src/obsidian_mcp/vault.py:917-971 (helper)

  Batch backup creation helper used by batch_append_notes for atomicity. Creates timestamped backups of specified notes concurrently with metadata preservation.
  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
• src/obsidian_mcp/vault.py:972-1022 (helper)

  Batch restore helper used for rollback on failure in batch_append_notes. Restores notes from backup concurrently preserving structure and metadata.
  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)