MCP Filesystem Server

Instructions

Implementation Reference

• mcp_filesystem/server.py:852-967 (handler)

  The MCP tool handler for 'edit_file_at_line', decorated with @mcp.tool(). Defines the tool schema via parameters and docstring. Delegates execution to the operations component and formats the results as a human-readable string summary.
  @mcp.tool() async def edit_file_at_line( path: str, line_edits: List[Dict[str, Any]], ctx: Context, offset: int = 0, limit: Optional[int] = None, relative_line_numbers: bool = False, abort_on_verification_failure: bool = False, encoding: str = "utf-8", dry_run: bool = False, ) -> str: """Edit specific lines in a text file. Args: path: Path to the file line_edits: List of edits to apply. Each edit is a dict with: - line_number: Line number to edit (0-based if relative_line_numbers=True, otherwise 1-based) - action: "replace", "insert_before", "insert_after", "delete" - content: New content for replace/insert operations (optional for delete) - expected_content: (Optional) Expected content of the line being edited for verification offset: Line offset (0-based) to start considering lines limit: Maximum number of lines to consider relative_line_numbers: Whether line numbers in edits are relative to offset abort_on_verification_failure: Whether to abort all edits if any verification fails encoding: Text encoding (default: utf-8) dry_run: If True, returns what would be changed without modifying the file ctx: MCP context Returns: Edit results summary """ try: components = get_components() results = await components["operations"].edit_file_at_line( path, line_edits, offset, limit, relative_line_numbers, abort_on_verification_failure, encoding, dry_run, ) # Format as text summary mode = "Would apply" if dry_run else "Applied" # Check if we had verification failures that prevented editing if "success" in results and not results["success"]: summary = [ f"Failed to edit {results['path']} due to content verification failures:", "", ] if "verification_failures" in results: for failure in results["verification_failures"]: line_num = failure.get("line", "?") action = failure.get("action", "?") summary.append(f"Line {line_num}: {action} - Verification failed") summary.append(f" Expected: {failure.get('expected', '').strip()}") summary.append(f" Actual: {failure.get('actual', '').strip()}") summary.append("") if "message" in results: summary.append(f"Error: {results['message']}") return "\n".join(summary) # Normal success case summary = [ f"{mode} {results['edits_applied']} edits to {results['path']}:", "", ] # Add verification warnings if any if "verification_failures" in results and results["verification_failures"]: summary.append( "Warning: Some content verification checks failed but edits were applied:" ) for failure in results["verification_failures"]: line_num = failure.get("line", "?") summary.append(f" Line {line_num}: Content did not match expected") summary.append("") for change in results["changes"]: line_num = change.get("line", "?") action = change.get("action", "?") orig_line_num = change.get("original_line_number", "") line_info = f"Line {line_num}" if relative_line_numbers and orig_line_num != "": line_info = f"Line {line_num} (relative: {orig_line_num})" if action == "replace": summary.append(f"{line_info}: Replaced") summary.append(f" - {change.get('before', '').strip()}") summary.append(f" + {change.get('after', '').strip()}") elif action == "insert_before": summary.append(f"{line_info}: Inserted before") summary.append(f" + {change.get('content', '').strip()}") elif action == "insert_after": summary.append(f"{line_info}: Inserted after") summary.append(f" + {change.get('content', '').strip()}") elif action == "delete": summary.append(f"{line_info}: Deleted") summary.append(f" - {change.get('content', '').strip()}") if "error" in change: summary.append(f" ! Error: {change['error']}") summary.append("") return "\n".join(summary) except Exception as e: return f"Error editing file: {str(e)}"
• mcp_filesystem/operations.py:692-960 (helper)

  The core helper function implementing the file line editing logic. Performs path validation, reads the file, applies edits (replace, insert, delete) with support for verification, relative numbering, offset/limit, dry-run, sorts edits to avoid index shifts, and returns detailed results.
  async def edit_file_at_line( self, path: Union[str, Path], line_edits: List[Dict[str, Any]], offset: int = 0, limit: Optional[int] = None, relative_line_numbers: bool = False, abort_on_verification_failure: bool = False, encoding: str = "utf-8", dry_run: bool = False, ) -> Dict[str, Any]: """Edit specific lines in a text file. Args: path: Path to the file line_edits: List of edits to apply. Each edit is a dict with: - line_number: Line number to edit (0-based if relative_line_numbers=True, otherwise 1-based) - action: "replace", "insert_before", "insert_after", "delete" - content: New content for replace/insert operations (optional for delete) - expected_content: (Optional) Expected content of the line being edited for verification offset: Line offset (0-based) to start considering lines limit: Maximum number of lines to consider relative_line_numbers: Whether line numbers in edits are relative to offset abort_on_verification_failure: Whether to abort all edits if any verification fails encoding: Text encoding (default: utf-8) dry_run: If True, returns what would be changed without modifying the file Returns: Dict with edit results including verification information Raises: ValueError: If path is outside allowed directories FileNotFoundError: If file does not exist """ abs_path, allowed = await self.validator.validate_path(path) if not allowed: raise ValueError(f"Path outside allowed directories: {path}") # Parameter validation if offset < 0: raise ValueError("offset must be non-negative") if limit is not None and limit < 0: raise ValueError("limit must be non-negative") try: # Read the entire file content = await anyio.to_thread.run_sync( partial(abs_path.read_text, encoding=encoding) ) lines = content.splitlines(keepends=True) # Calculate effective range end_offset = None if limit is not None: end_offset = offset + limit - 1 else: end_offset = len(lines) - 1 # Ensure end_offset is within bounds if end_offset >= len(lines): end_offset = len(lines) - 1 # Track verification failures verification_failures = [] # Validate line_edits and adjust line numbers if needed for i, edit in enumerate(line_edits): if "line_number" not in edit: raise ValueError(f"Edit {i} is missing line_number") if "action" not in edit: raise ValueError(f"Edit {i} is missing action") line_num = edit["line_number"] action = edit["action"] # Handle relative line numbers absolute_line_num = line_num if relative_line_numbers: # If relative, line_num is 0-based and relative to offset if line_num < 0: raise ValueError( f"Relative line number {line_num} must be non-negative" ) absolute_line_num = offset + line_num + 1 else: # Not relative, line_num is 1-based (standard line numbers) absolute_line_num = line_num # Store the adjusted line number for later use edit["_absolute_line_num"] = absolute_line_num # Check if line is within the considered range if ( absolute_line_num < 1 or absolute_line_num > len(lines) + 1 ): # +1 to allow appending at the end raise ValueError( f"Line number {absolute_line_num} is outside file bounds (1-{len(lines)})" ) if offset > 0 and absolute_line_num < offset + 1: raise ValueError( f"Line number {absolute_line_num} is before offset {offset}" ) if limit is not None and absolute_line_num > offset + limit: raise ValueError( f"Line number {absolute_line_num} is beyond limit (offset {offset} + limit {limit})" ) if action not in ["replace", "insert_before", "insert_after", "delete"]: raise ValueError(f"Invalid action '{action}' in edit {i}") if action != "delete" and "content" not in edit: raise ValueError( f"Edit {i} with action '{action}' is missing content" ) # Verify expected content if provided if "expected_content" in edit and absolute_line_num <= len(lines): expected = edit["expected_content"] actual = lines[absolute_line_num - 1].rstrip("\r\n") if expected.rstrip("\r\n") != actual: failure = { "edit_index": i, "line": absolute_line_num, "action": action, "expected": expected, "actual": actual, } verification_failures.append(failure) if abort_on_verification_failure: return { "success": False, "path": str(abs_path), "verification_failures": verification_failures, "message": f"Content verification failed at line {absolute_line_num}", "edits_applied": 0, "changes": [], "dry_run": dry_run, } # If there are verification failures but we're not aborting, # we'll continue with the edits and report the failures # Sort edits by absolute line number in reverse order to avoid line number changes line_edits = sorted( line_edits, key=lambda e: e["_absolute_line_num"], reverse=True ) # Apply edits results = [] for edit in line_edits: # Use the adjusted absolute line number line_num = edit["_absolute_line_num"] action = edit["action"] content_before = lines[line_num - 1] if line_num <= len(lines) else "" if action == "replace": # Ensure proper line endings new_content = edit["content"] if not new_content.endswith("\n") and content_before.endswith("\n"): new_content += "\n" if line_num <= len(lines): lines[line_num - 1] = new_content else: # If replacing beyond the end, append with any necessary newlines while len(lines) < line_num - 1: lines.append("\n") lines.append(new_content) results.append( { "line": line_num, "original_line_number": edit["line_number"], "action": "replace", "before": content_before, "after": new_content, } ) elif action == "insert_before": # Ensure proper line endings new_content = edit["content"] if not new_content.endswith("\n"): new_content += "\n" if line_num <= len(lines): lines.insert(line_num - 1, new_content) else: # If inserting beyond the end, append with any necessary newlines while len(lines) < line_num - 1: lines.append("\n") lines.append(new_content) results.append( { "line": line_num, "original_line_number": edit["line_number"], "action": "insert_before", "content": new_content, } ) elif action == "insert_after": # Ensure proper line endings new_content = edit["content"] if not new_content.endswith("\n"): new_content += "\n" if line_num <= len(lines): lines.insert(line_num, new_content) else: # If inserting beyond the end, append with any necessary newlines while len(lines) < line_num: lines.append("\n") lines.append(new_content) results.append( { "line": line_num, "action": "insert_after", "content": new_content, } ) elif action == "delete": if line_num <= len(lines): deleted_content = lines.pop(line_num - 1) results.append( { "line": line_num, "action": "delete", "content": deleted_content, } ) else: results.append( { "line": line_num, "action": "delete", "error": "Line does not exist", } ) # Write back the file if not a dry run if not dry_run: new_content = "".join(lines) await anyio.to_thread.run_sync( partial(abs_path.write_text, new_content, encoding=encoding) ) return { "path": str(abs_path), "edits_applied": len(results), "dry_run": dry_run, "changes": results, } except FileNotFoundError: raise FileNotFoundError(f"File not found: {path}") except PermissionError: raise ValueError(f"Permission denied: {path}") except UnicodeDecodeError: raise ValueError(f"Cannot decode file as {encoding}")