From 5190e8c84979c954a98edfe5b5be692f01451e96 Mon Sep 17 00:00:00 2001
From: Stefan Lohmaier <stefan@slohmaier.de>
Date: Fri, 12 Jun 2026 08:08:08 +0200
Subject: [PATCH] Add write_file, create_folder, delete_file, move_file to
 Files server

Full CRUD for oCIS files:
- write_file: text or base64 binary content
- create_folder: MKCOL
- delete_file: DELETE (files and folders)
- move_file: MOVE (rename or relocate)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 files/server.py | 75 +++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 75 insertions(+)

diff --git a/files/server.py b/files/server.py
index ceb2b3b..997e41c 100644
--- a/files/server.py
+++ b/files/server.py
@@ -162,6 +162,81 @@ def search_files(
     return "\n".join(lines) if lines else "Keine Dateien gefunden"
 
 
+@mcp.tool()
+def write_file(
+    path: Annotated[str, Field(description="Full file path, e.g. '/Documents/notes.txt', '/todo.md'. Creates parent dirs automatically.")],
+    content: Annotated[str, Field(description="File content as text (for text files) or base64-encoded data (for binary, set is_base64=true)")],
+    is_base64: Annotated[bool, Field(description="True if content is base64-encoded binary data")] = False,
+) -> str:
+    """Write or overwrite a file. Use text content for txt/md/json etc. Use base64 for binary files (images, documents)."""
+    user = get_current_user()
+    if not user: return "Error: not authenticated"
+    if is_base64:
+        data = base64.b64decode(content)
+    else:
+        data = content.encode("utf-8")
+    ct = _guess_mime(path, None)
+    r = httpx.put(_dav(user, path), content=data, auth=_auth(user),
+                  headers={"Content-Type": ct}, timeout=60)
+    if r.status_code in (200, 201, 204):
+        return f"Datei geschrieben: {path} ({len(data):,} bytes)"
+    if r.status_code == 409:
+        return f"Fehler: Uebergeordnetes Verzeichnis existiert nicht. Erstelle es zuerst mit create_folder."
+    return f"Fehler: HTTP {r.status_code}"
+
+
+@mcp.tool()
+def create_folder(
+    path: Annotated[str, Field(description="Folder path to create, e.g. '/Documents/Projekte', '/Fotos/2026'")],
+) -> str:
+    """Create a new folder/directory."""
+    user = get_current_user()
+    if not user: return "Error: not authenticated"
+    r = httpx.request("MKCOL", _dav(user, path), auth=_auth(user), timeout=15)
+    if r.status_code in (200, 201):
+        return f"Ordner erstellt: {path}"
+    if r.status_code == 405:
+        return f"Ordner existiert bereits: {path}"
+    if r.status_code == 409:
+        return f"Fehler: Uebergeordnetes Verzeichnis existiert nicht."
+    return f"Fehler: HTTP {r.status_code}"
+
+
+@mcp.tool()
+def delete_file(
+    path: Annotated[str, Field(description="Path to file or folder to delete, e.g. '/Documents/old.txt', '/temp/'")],
+) -> str:
+    """Delete a file or folder (including all contents). This cannot be undone."""
+    user = get_current_user()
+    if not user: return "Error: not authenticated"
+    r = httpx.request("DELETE", _dav(user, path), auth=_auth(user), timeout=30)
+    if r.status_code in (200, 204):
+        return f"Geloescht: {path}"
+    if r.status_code == 404:
+        return f"Nicht gefunden: {path}"
+    return f"Fehler: HTTP {r.status_code}"
+
+
+@mcp.tool()
+def move_file(
+    source: Annotated[str, Field(description="Current path, e.g. '/Documents/old-name.txt'")],
+    destination: Annotated[str, Field(description="New path, e.g. '/Documents/new-name.txt' or '/Archive/file.txt'")],
+) -> str:
+    """Move or rename a file or folder."""
+    user = get_current_user()
+    if not user: return "Error: not authenticated"
+    dest_url = _dav(user, destination)
+    r = httpx.request("MOVE", _dav(user, source), auth=_auth(user),
+                      headers={"Destination": dest_url}, timeout=30)
+    if r.status_code in (200, 201, 204):
+        return f"Verschoben: {source} -> {destination}"
+    if r.status_code == 404:
+        return f"Quelle nicht gefunden: {source}"
+    if r.status_code == 409:
+        return f"Fehler: Zielverzeichnis existiert nicht."
+    return f"Fehler: HTTP {r.status_code}"
+
+
 def create_app():
     from contextlib import asynccontextmanager
     mcp_app = mcp.streamable_http_app()