Add tool to rename model using the LSP server

.changes/unreleased/Enhancement or New Feature-20251031-141654.yaml

@@ -0,0 +1,3 @@
+kind: Enhancement or New Feature
+body: Add ability to call the LSP to rename models
+time: 2025-10-31T14:16:54.720743+01:00

src/dbt_mcp/lsp/lsp_client.py

@@ -7,7 +7,9 @@
 
 import asyncio
 import logging
+from pathlib import Path
 from typing import Any
+from urllib.parse import unquote, urlparse
 
 from dbt_mcp.lsp.lsp_connection import LSPConnection, LspEventName
 
@@ -148,3 +150,247 @@ async def _list_nodes(
                 return {"nodes": result["nodes"]}
 
             return result
+
+    def _uri_to_path(self, uri: str) -> Path:
+        """Convert a file:// URI to a Path object."""
+        parsed = urlparse(uri)
+        # Decode percent-encoded characters and remove the leading slash on Windows
+        path_str = unquote(parsed.path)
+        return Path(path_str)
+
+    def _apply_single_edit(self, lines: list[str], edit: dict[str, Any]) -> None:
+        """Apply a single text edit to a list of lines.
+
+        Args:
+            lines: List of lines (with line endings preserved)
+            edit: LSP TextEdit object with 'range' and 'newText'
+        """
+        start_line_idx = edit["range"]["start"]["line"]
+        start_char = edit["range"]["start"]["character"]
+        end_line_idx = edit["range"]["end"]["line"]
+        end_char = edit["range"]["end"]["character"]
+        new_text = edit["newText"]
+
+        # Handle single line edit
+        if start_line_idx == end_line_idx:
+            if start_line_idx < len(lines):
+                line = lines[start_line_idx]
+                # Strip line ending to work with just the text
+                line_ending = ""
+                if line.endswith("\r\n"):
+                    line_ending = "\r\n"
+                    line = line[:-2]
+                elif line.endswith("\n"):
+                    line_ending = "\n"
+                    line = line[:-1]
+                elif line.endswith("\r"):
+                    line_ending = "\r"
+                    line = line[:-1]
+
+                # Apply the edit
+                lines[start_line_idx] = (
+                    line[:start_char] + new_text + line[end_char:] + line_ending
+                )
+        else:
+            # Multi-line edit
+            start_line = lines[start_line_idx]
+            end_line = lines[end_line_idx]
+
+            # Strip line endings
+            if start_line.endswith("\r\n"):
+                start_line = start_line[:-2]
+            elif start_line.endswith("\n"):
+                start_line = start_line[:-1]
+            elif start_line.endswith("\r"):
+                start_line = start_line[:-1]
+
+            end_line_ending = ""
+            if end_line.endswith("\r\n"):
+                end_line_ending = "\r\n"
+                end_line = end_line[:-2]
+            elif end_line.endswith("\n"):
+                end_line_ending = "\n"
+                end_line = end_line[:-1]
+            elif end_line.endswith("\r"):
+                end_line_ending = "\r"
+                end_line = end_line[:-1]
+
+            start_line_text = start_line[:start_char]
+            end_line_text = end_line[end_char:]
+            lines[start_line_idx] = (
+                start_line_text + new_text + end_line_text + end_line_ending
+            )
+            # Remove the lines in between
+            del lines[start_line_idx + 1 : end_line_idx + 1]
+
+    def _apply_text_edits(self, file_path: Path, edits: list[dict[str, Any]]) -> None:
+        """Apply text edits to a file.
+
+        Args:
+            file_path: Path to the file to edit
+            edits: List of LSP TextEdit objects with 'range' and 'newText'
+        """
+        if not file_path.exists():
+            logger.warning(f"File not found for edits: {file_path}")
+            return
+
+        # Read the file content
+        content = file_path.read_text()
+
+        # Sort edits in reverse order (end to start) to avoid offset issues
+        # We sort by start position descending so we can apply from end to beginning
+        sorted_edits = sorted(
+            edits,
+            key=lambda e: (
+                e["range"]["start"]["line"],
+                e["range"]["start"]["character"],
+            ),
+            reverse=True,
+        )
+
+        # Convert content to list of lines for easier manipulation
+        lines = content.splitlines(keepends=True)
+
+        # Apply each edit
+        for edit in sorted_edits:
+            self._apply_single_edit(lines, edit)
+
+        # Write back to file
+        file_path.write_text("".join(lines))
+        logger.info(f"Applied {len(edits)} edits to {file_path}")
+
+    async def rename_model(
+        self,
+        old_uri: str,
+        new_uri: str,
+        apply_edits: bool = True,
+        timeout: float | None = None,
+    ) -> dict[str, Any]:
+        """Rename a dbt model file and update all references.
+
+        This method:
+        1. Asks the LSP server what edits are needed for the rename
+        2. Applies those edits to update references in other files
+        3. Performs the actual file rename on disk
+        4. Notifies the LSP server that the rename is complete
+
+        Args:
+            old_uri: The current file URI (e.g., "file:///path/to/model.sql")
+            new_uri: The new file URI (e.g., "file:///path/to/new_model.sql")
+            apply_edits: Whether to apply workspace edits and perform the rename (default: True)
+            timeout: Optional timeout for the request
+
+        Returns:
+            Dictionary with:
+            - 'renamed': True if model was renamed
+            - 'old_path': Original file path
+            - 'new_path': New file path
+            - 'files_updated': List of files that had references updated
+            - 'error': Error message if something failed
+        """
+        logger.info(f"Renaming model: {old_uri} -> {new_uri}")
+
+        # Step 1: Ask LSP what edits are needed
+        params = {
+            "files": [
+                {
+                    "oldUri": old_uri,
+                    "newUri": new_uri,
+                }
+            ]
+        }
+
+        try:
+            async with asyncio.timeout(timeout or self.timeout):
+                result = await self.lsp_connection.send_request(
+                    "workspace/willRenameFiles",
+                    params,
+                )
+
+                # Handle None or empty result
+                if result is None:
+                    result = {}
+
+                if "error" in result and result["error"] is not None:
+                    return {"error": result["error"]}
+
+                if not apply_edits:
+                    # Just return the workspace edits without applying them
+                    return result
+
+                # Step 2: Apply workspace edits
+                files_updated = []
+                if result and "changes" in result:
+                    # Handle WorkspaceEdit with 'changes' format
+                    for file_uri, edits in result["changes"].items():
+                        try:
+                            file_path = self._uri_to_path(file_uri)
+                            self._apply_text_edits(file_path, edits)
+                            files_updated.append(str(file_path))
+                        except Exception as e:
+                            logger.error(f"Failed to apply edits to {file_uri}: {e}")
+                            return {
+                                "error": f"Failed to apply edits to {file_uri}: {str(e)}"
+                            }
+                elif result and "documentChanges" in result:
+                    # Handle WorkspaceEdit with 'documentChanges' format
+                    for change in result["documentChanges"]:
+                        if "textDocument" in change and "edits" in change:
+                            file_uri = change["textDocument"]["uri"]
+                            try:
+                                file_path = self._uri_to_path(file_uri)
+                                self._apply_text_edits(file_path, change["edits"])
+                                files_updated.append(str(file_path))
+                            except Exception as e:
+                                logger.error(
+                                    f"Failed to apply edits to {file_uri}: {e}"
+                                )
+                                return {
+                                    "error": f"Failed to apply edits to {file_uri}: {str(e)}"
+                                }
+
+                # Step 3: Perform the actual file rename
+                old_path = self._uri_to_path(old_uri)
+                new_path = self._uri_to_path(new_uri)
+
+                if not old_path.exists():
+                    return {"error": f"Source file does not exist: {old_path}"}
+
+                if new_path.exists():
+                    return {"error": f"Destination file already exists: {new_path}"}
+
+                try:
+                    # Ensure parent directory exists
+                    new_path.parent.mkdir(parents=True, exist_ok=True)
+                    old_path.rename(new_path)
+                    logger.info(f"Renamed file: {old_path} -> {new_path}")
+                except Exception as e:
+                    return {"error": f"Failed to rename file: {str(e)}"}
+
+                # Step 4: Notify LSP of completion (didRenameFiles)
+                try:
+                    self.lsp_connection.send_notification(
+                        "workspace/didRenameFiles",
+                        {
+                            "files": [
+                                {
+                                    "oldUri": old_uri,
+                                    "newUri": new_uri,
+                                }
+                            ]
+                        },
+                    )
+                except Exception as e:
+                    logger.warning(f"Failed to send didRenameFiles notification: {e}")
+
+                return {
+                    "renamed": True,
+                    "old_path": str(old_path),
+                    "new_path": str(new_path),
+                    "files_updated": files_updated,
+                }
+
+        except TimeoutError:
+            return {"error": "Timeout waiting for LSP response"}
+        except Exception as e:
+            return {"error": f"Failed to rename file: {str(e)}"}

src/dbt_mcp/lsp/tools.py

@@ -111,6 +111,16 @@ async def wrapper(*args, **kwargs) -> Any:
                 idempotent_hint=True,
             ),
         ),
+        ToolDefinition(
+            fn=call_with_lsp_client(rename_model),
+            description=get_prompt("lsp/rename_model"),
+            annotations=create_tool_annotations(
+                title="rename_model",
+                read_only_hint=False,
+                destructive_hint=True,  # Modifies files on disk
+                idempotent_hint=False,
+            ),
+        ),
     ]
 
 
@@ -163,6 +173,56 @@ async def get_column_lineage(
         return {"error": error_msg}
 
 
+async def rename_model(
+    lsp_client: LSPClient,
+    old_uri: str = Field(description=get_prompt("lsp/args/old_uri")),
+    new_uri: str = Field(description=get_prompt("lsp/args/new_uri")),
+) -> dict[str, Any]:
+    """Rename a dbt model in the project workspace and update all references.
+
+    Args:
+        lsp_client: The LSP client instance
+        old_uri: The current file URI
+        new_uri: The new file URI
+
+    Returns:
+        Dictionary with either:
+        - 'renamed': True, 'old_path', 'new_path', 'files_updated' on success
+        - 'error' key containing error message on failure
+    """
+    try:
+        response = await lsp_client.rename_model(
+            old_uri=old_uri,
+            new_uri=new_uri,
+            apply_edits=True,
+        )
+
+        # Check for LSP-level errors
+        if "error" in response:
+            logger.error(f"Error renaming model: {response['error']}")
+            return {"error": response["error"]}
+
+        if response.get("renamed"):
+            logger.info(
+                f"Successfully renamed {response['old_path']} to {response['new_path']}"
+            )
+            if response.get("files_updated"):
+                logger.info(
+                    f"Updated {len(response['files_updated'])} files with new references"
+                )
+
+        return response
+
+    except TimeoutError:
+        error_msg = f"Timeout waiting for model rename (old: {old_uri}, new: {new_uri})"
+        logger.error(error_msg)
+        return {"error": error_msg}
+    except Exception as e:
+        error_msg = f"Failed to rename model from {old_uri} to {new_uri}: {str(e)}"
+        logger.error(error_msg)
+        return {"error": error_msg}
+
+
 async def cleanup_lsp_connection() -> None:
     """Clean up the LSP connection when shutting down."""
     global _lsp_connection

src/dbt_mcp/prompts/lsp/args/new_uri.md

@@ -0,0 +1,4 @@
+The new model file URI (e.g., "file:///path/to/project/models/staging/stg_customers_new.sql").
+
+This should be the absolute path to the new model file location using the file:// URI scheme.
+

src/dbt_mcp/prompts/lsp/args/old_uri.md

@@ -0,0 +1,4 @@
+The current model file URI (e.g., "file:///path/to/project/models/staging/stg_customers.sql").
+
+This should be the absolute path to the model file using the file:// URI scheme.
+

src/dbt_mcp/prompts/lsp/rename_model.md

@@ -0,0 +1,19 @@
+Rename a dbt model in the project workspace and automatically update all references.
+
+This tool performs a complete model rename operation with the help of the LSP server:
+
+1. **Queries the LSP server** for all files that reference the model being renamed
+2. **Updates references** in all affected files (e.g., updates model names in `ref()` calls)
+3. **Renames the model file** on disk to the new location
+4. **Notifies the LSP server** that the rename is complete
+
+The tool handles:
+- Renaming model files (.sql files in your models directory)
+- Updating all `ref('model_name')` references throughout the project
+- Updating imports and dependencies in other models
+- Preserving file encoding and line endings
+
+**Important**: This tool makes actual changes to your files. Make sure you have committed any unsaved work before using it.
+
+Use this when you need to safely rename a dbt model while ensuring all references to it are updated correctly throughout your dbt project.
+

src/dbt_mcp/tools/policy.py

@@ -149,4 +149,7 @@ class ToolPolicy:
     ToolName.GET_COLUMN_LINEAGE.value: ToolPolicy(
         name=ToolName.GET_COLUMN_LINEAGE.value, behavior=ToolBehavior.METADATA
     ),
+    ToolName.RENAME_MODEL.value: ToolPolicy(
+        name=ToolName.RENAME_MODEL.value, behavior=ToolBehavior.METADATA
+    ),
 }

src/dbt_mcp/tools/tool_names.py

@@ -60,6 +60,7 @@ class ToolName(Enum):
 
     # dbt LSP tools
     GET_COLUMN_LINEAGE = "get_column_lineage"
+    RENAME_MODEL = "rename_model"
 
     @classmethod
     def get_all_tool_names(cls) -> set[str]:

src/dbt_mcp/tools/toolsets.py

@@ -73,5 +73,6 @@ class Toolset(Enum):
     },
     Toolset.DBT_LSP: {
         ToolName.GET_COLUMN_LINEAGE,
+        ToolName.RENAME_MODEL,
     },
 }