fix: refuse to silently overwrite existing files in save_data (#539)

Add overwrite flag (default false) + expanduser. Fixes #539.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: Shashankss1205

src/sktime_mcp/server.py

@@ -751,6 +751,14 @@ async def list_tools() -> list[Tool]:
                         "enum": ["csv", "parquet", "json"],
                         "default": "csv",
                     },
+                    "overwrite": {
+                        "type": "boolean",
+                        "description": (
+                            "Replace the file if it already exists. Default false — an "
+                            "existing file is not clobbered unless this is true."
+                        ),
+                        "default": False,
+                    },
                 },
                 "required": ["data_handle", "path"],
             },
@@ -1116,6 +1124,7 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
                 data_handle=arguments["data_handle"],
                 path=arguments["path"],
                 format=arguments.get("format", "csv"),
+                overwrite=arguments.get("overwrite", False),
             )
 
         elif name == "auto_format_on_load":

src/sktime_mcp/tools/save_data.py

@@ -26,6 +26,7 @@ def save_data_tool(
     data_handle: str,
     path: str,
     format: str = "csv",
+    overwrite: bool = False,
 ) -> dict[str, Any]:
     """Persist the data behind a handle to a local file.
 
@@ -42,6 +43,9 @@ def save_data_tool(
         use the `format` argument instead.
     format : str, default="csv"
         Output format. Must be one of: "csv", "parquet", or "json".
+    overwrite : bool, default=False
+        If the target file already exists, the write is refused unless this
+        is True, in which case the response reports ``overwritten: true``.
 
     Returns
     -------
@@ -52,6 +56,7 @@ def save_data_tool(
         - ``"saved_path"`` (str) -- Absolute path to the written file.
         - ``"format"`` (str) -- The format used to write the file.
         - ``"rows"`` (int) -- Number of rows written to the file.
+        - ``"overwritten"`` (bool) -- True if an existing file was replaced.
         - ``"error"`` (str, optional) -- Error message if "success" is False.
     """
     executor = get_executor()
@@ -70,6 +75,20 @@ def save_data_tool(
             "error": f"Unsupported format '{format}'. Choose from: {list(_FORMAT_WRITERS.keys())}",
         }
 
+    # Resolve (expanduser so "~/x" doesn't create a literal "~" dir) and guard
+    # against silently clobbering an existing file (#539).
+    abs_path = Path(path).expanduser().resolve()
+    existed = abs_path.exists()
+    if existed and not overwrite:
+        return {
+            "success": False,
+            "error": (
+                f"File already exists: '{abs_path}'. Pass overwrite=true to replace it, "
+                "or choose a different path."
+            ),
+            "saved_path": str(abs_path),
+        }
+
     data_info = executor._data_handles[data_handle]
     y = data_info["y"]
     X = data_info.get("X")
@@ -88,7 +107,6 @@ def save_data_tool(
             df = pd.concat([df, X], axis=1)
 
         # Ensure target directory exists
-        abs_path = Path(path).resolve()
         abs_path.parent.mkdir(parents=True, exist_ok=True)
 
         # Write
@@ -106,6 +124,7 @@ def save_data_tool(
             "saved_path": str(abs_path),
             "format": fmt,
             "rows": len(df),
+            "overwritten": existed,
         }
 
     except Exception as e:

tests/test_data_management.py

@@ -235,31 +235,47 @@ def test_save_csv(self):
         executor, handle = _make_executor_with_data()
         self._patch(executor)
         try:
-            with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as f:
-                path = f.name
-            result = save_data_tool(handle, path=path, format="csv")
+            with tempfile.TemporaryDirectory() as d:
+                path = str(Path(d) / "out.csv")
+                result = save_data_tool(handle, path=path, format="csv")
+                assert result["success"]
+                assert result["format"] == "csv"
+                assert result["rows"] == 60
+                assert result["overwritten"] is False
+                assert Path(result["saved_path"]).exists()
         finally:
             self._unpatch()
 
-        assert result["success"]
-        assert result["format"] == "csv"
-        assert result["rows"] == 60
-        assert Path(result["saved_path"]).exists()
-        Path(path).unlink()
-
     def test_save_json(self):
         executor, handle = _make_executor_with_data()
         self._patch(executor)
         try:
-            with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
-                path = f.name
-            result = save_data_tool(handle, path=path, format="json")
+            with tempfile.TemporaryDirectory() as d:
+                path = str(Path(d) / "out.json")
+                result = save_data_tool(handle, path=path, format="json")
+                assert result["success"]
+                assert result["format"] == "json"
         finally:
             self._unpatch()
 
-        assert result["success"]
-        assert result["format"] == "json"
-        Path(path).unlink()
+    def test_save_refuses_existing_file(self):
+        executor, handle = _make_executor_with_data()
+        self._patch(executor)
+        try:
+            with tempfile.TemporaryDirectory() as d:
+                path = str(Path(d) / "out.csv")
+                first = save_data_tool(handle, path=path, format="csv")
+                assert first["success"]
+                # second write to the same path is refused without overwrite
+                second = save_data_tool(handle, path=path, format="csv")
+                assert not second["success"]
+                assert "already exists" in second["error"]
+                # ...and allowed with overwrite=True
+                third = save_data_tool(handle, path=path, format="csv", overwrite=True)
+                assert third["success"]
+                assert third["overwritten"] is True
+        finally:
+            self._unpatch()
 
     def test_save_unsupported_format(self):
         executor, handle = _make_executor_with_data()