Add overwrite guards to sheets_write and sheets_create

sheets_write: reads target range before overwriting. If data exists,
returns a warning asking the user to confirm_overwrite=True or use
append=True. Empty ranges proceed without confirmation.

sheets_create: checks Drive for an existing spreadsheet with the same
title before creating. Returns a warning if a duplicate is found.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: RafaelPo

everyrow-mcp/src/everyrow_mcp/sheets_models.py

@@ -99,6 +99,11 @@ class SheetsWriteInput(BaseModel):
         default=False,
         description="If True, append after existing data instead of overwriting.",
     )
+    confirm_overwrite: bool = Field(
+        default=False,
+        description="Must be set to True to overwrite existing data when append=False. "
+        "The tool will check if the range has data and warn you first.",
+    )
 
     @field_validator("spreadsheet_id")
     @classmethod

everyrow-mcp/src/everyrow_mcp/sheets_tools.py

@@ -227,6 +227,23 @@ async def sheets_write(params: SheetsWriteInput) -> list[TextContent]:
                     )
                 ]
             else:
+                # Pre-check: warn if the target range already has data
+                if not params.confirm_overwrite:
+                    existing = await client.read_range(
+                        params.spreadsheet_id, cell_range=params.range
+                    )
+                    if existing:
+                        existing_rows = len(existing)
+                        return [
+                            TextContent(
+                                type="text",
+                                text=f"The range '{params.range}' already contains {existing_rows} rows "
+                                f"(including headers). Writing will overwrite this data. "
+                                f"To proceed, call again with confirm_overwrite=True, "
+                                f"or use append=True to add rows after existing data.",
+                            )
+                        ]
+
                 result = await client.write_range(
                     params.spreadsheet_id, cell_range=params.range, values=values
                 )
@@ -269,6 +286,21 @@ async def sheets_create(params: SheetsCreateInput) -> list[TextContent]:
         token = await get_google_token()
 
         async with GoogleSheetsClient(token) as client:
+            # Duplicate title guard
+            existing = await client.list_spreadsheets(
+                query=params.title, max_results=50
+            )
+            for f in existing:
+                if f.get("name") == params.title:
+                    return [
+                        TextContent(
+                            type="text",
+                            text=f"A spreadsheet named '{params.title}' already exists "
+                            f"(id: {f['id']}). Pick a different title to avoid "
+                            f"creating a duplicate.",
+                        )
+                    ]
+
             metadata = await client.create_spreadsheet(params.title)
             spreadsheet_id = metadata["spreadsheetId"]
             url = metadata.get(

everyrow-mcp/tests/test_sheets_tools.py

@@ -384,7 +384,7 @@ async def test_url_extraction(self, _mock_google_token):
 
 class TestSheetsWriteTool:
     @pytest.mark.asyncio
-    async def test_write_overwrite(self, _mock_google_token):
+    async def test_write_overwrite_confirmed(self, _mock_google_token):
         mock_resp = _mock_response(
             {
                 "updatedRange": "Sheet1!A1:B3",
@@ -399,6 +399,51 @@ async def test_write_overwrite(self, _mock_google_token):
                 SheetsWriteInput(
                     spreadsheet_id="abc123def456ghi789jkl012mno345pqr678stu901v",
                     data=[{"name": "Alice"}, {"name": "Bob"}],
+                    confirm_overwrite=True,
+                )
+            )
+
+        assert "Wrote" in result[0].text
+
+    @pytest.mark.asyncio
+    async def test_write_overwrite_warns_if_existing_data(self, _mock_google_token):
+        """Writing without confirm_overwrite warns when range has data."""
+        read_resp = _mock_response({"values": [["name"], ["Alice"]]})
+
+        with patch.object(
+            httpx.AsyncClient, "get", new_callable=AsyncMock, return_value=read_resp
+        ):
+            result = await sheets_write(
+                SheetsWriteInput(
+                    spreadsheet_id="abc123def456ghi789jkl012mno345pqr678stu901v",
+                    data=[{"name": "Bob"}],
+                )
+            )
+
+        assert "already contains" in result[0].text
+        assert "confirm_overwrite" in result[0].text
+
+    @pytest.mark.asyncio
+    async def test_write_overwrite_proceeds_on_empty_range(self, _mock_google_token):
+        """Writing without confirm_overwrite proceeds when range is empty."""
+        read_resp = _mock_response({})  # empty range
+        write_resp = _mock_response({"updatedRange": "Sheet1!A1:B2", "updatedRows": 2})
+
+        with (
+            patch.object(
+                httpx.AsyncClient, "get", new_callable=AsyncMock, return_value=read_resp
+            ),
+            patch.object(
+                httpx.AsyncClient,
+                "put",
+                new_callable=AsyncMock,
+                return_value=write_resp,
+            ),
+        ):
+            result = await sheets_write(
+                SheetsWriteInput(
+                    spreadsheet_id="abc123def456ghi789jkl012mno345pqr678stu901v",
+                    data=[{"name": "Bob"}],
                 )
             )
 
@@ -432,15 +477,27 @@ async def test_write_append(self, _mock_google_token):
 class TestSheetsCreateTool:
     @pytest.mark.asyncio
     async def test_create_empty(self, _mock_google_token):
-        mock_resp = _mock_response(
+        list_resp = _mock_response({"files": []})  # no duplicates
+        create_resp = _mock_response(
             {
                 "spreadsheetId": "new-id-123",
                 "spreadsheetUrl": "https://docs.google.com/spreadsheets/d/new-id-123",
             }
         )
 
-        with patch.object(
-            httpx.AsyncClient, "post", new_callable=AsyncMock, return_value=mock_resp
+        with (
+            patch.object(
+                httpx.AsyncClient,
+                "get",
+                new_callable=AsyncMock,
+                return_value=list_resp,
+            ),
+            patch.object(
+                httpx.AsyncClient,
+                "post",
+                new_callable=AsyncMock,
+                return_value=create_resp,
+            ),
         ):
             result = await sheets_create(SheetsCreateInput(title="Test"))
 
@@ -451,6 +508,7 @@ async def test_create_empty(self, _mock_google_token):
 
     @pytest.mark.asyncio
     async def test_create_with_data(self, _mock_google_token):
+        list_resp = _mock_response({"files": []})  # no duplicates
         create_resp = _mock_response(
             {
                 "spreadsheetId": "new-id-456",
@@ -460,6 +518,12 @@ async def test_create_with_data(self, _mock_google_token):
         write_resp = _mock_response({"updatedRows": 2})
 
         with (
+            patch.object(
+                httpx.AsyncClient,
+                "get",
+                new_callable=AsyncMock,
+                return_value=list_resp,
+            ),
             patch.object(
                 httpx.AsyncClient,
                 "post",
@@ -480,6 +544,19 @@ async def test_create_with_data(self, _mock_google_token):
         data = json.loads(result[0].text)
         assert data["rows_written"] == 1
 
+    @pytest.mark.asyncio
+    async def test_create_rejects_duplicate_title(self, _mock_google_token):
+        """sheets_create warns when a spreadsheet with the same title exists."""
+        list_resp = _mock_response({"files": [{"id": "existing-id", "name": "Budget"}]})
+
+        with patch.object(
+            httpx.AsyncClient, "get", new_callable=AsyncMock, return_value=list_resp
+        ):
+            result = await sheets_create(SheetsCreateInput(title="Budget"))
+
+        assert "already exists" in result[0].text
+        assert "existing-id" in result[0].text
+
 
 class TestSheetsInfoTool:
     @pytest.mark.asyncio