Merge pull request #92 from neovasky/master

feat(webdav): add complete file system support

Author: cbcoutinho

.gitignore

@@ -1,2 +1,6 @@
 __pycache__/
 .coverage
+.env
+*.env
+.env.local
+.env.*.local

CHANGELOG.md

@@ -1,3 +1,26 @@
+## [Unreleased]
+
+### Feat
+
+- **webdav**: Add complete file system support with directory browsing, file read/write, and resource management
+- **webdav**: Add `nc_webdav_list_directory` tool for browsing any NextCloud directory
+- **webdav**: Add `nc_webdav_read_file` tool with automatic text/binary content handling
+- **webdav**: Add `nc_webdav_write_file` tool supporting text and base64 binary content
+- **webdav**: Add `nc_webdav_create_directory` tool for creating directories
+- **webdav**: Add `nc_webdav_delete_resource` tool for deleting files and directories
+- **webdav**: Add XML parsing for WebDAV PROPFIND responses with metadata extraction
+
+### Fix
+
+- **types**: Improve type annotations throughout codebase for better IDE support
+- **types**: Fix Context parameter ordering in MCP tools (required before optional)
+- **types**: Add proper type hints for WebDAV client methods
+
+### Refactor
+
+- **webdav**: Extend WebDAV client beyond Notes attachments to general file operations
+- **server**: Enhance error handling and logging for WebDAV operations
+
 ## v0.4.1 (2025-07-10)
 
 ### Fix

README.md

@@ -14,6 +14,7 @@ The server provides integration with multiple Nextcloud apps, enabling LLMs to i
 |-----|----------------|-------------|
 | **Notes** | ✅ Full Support | Create, read, update, delete, and search notes. Handle attachments via WebDAV. |
 | **Tables** | ⚠️ Row Operations | Read table schemas and perform CRUD operations on table rows. Table management not yet supported. |
+| **Files (WebDAV)** | ✅ Full Support | Complete file system access - browse directories, read/write files, create/delete resources. |
 
 ## Available Tools
 
@@ -39,6 +40,16 @@ The server provides integration with multiple Nextcloud apps, enabling LLMs to i
 | `nc_tables_update_row` | Update an existing row in a table |
 | `nc_tables_delete_row` | Delete a row from a table |
 
+### WebDAV File System Tools
+
+| Tool | Description |
+|------|-------------|
+| `nc_webdav_list_directory` | List files and directories in any NextCloud path |
+| `nc_webdav_read_file` | Read file content (text files decoded, binary as base64) |
+| `nc_webdav_write_file` | Create or update files in NextCloud |
+| `nc_webdav_create_directory` | Create new directories |
+| `nc_webdav_delete_resource` | Delete files or directories |
+
 ## Available Resources
 
 | Resource | Description |
@@ -47,6 +58,37 @@ The server provides integration with multiple Nextcloud apps, enabling LLMs to i
 | `notes://settings` | Access Notes app settings |
 | `nc://Notes/{note_id}/attachments/{attachment_filename}` | Access attachments for notes |
 
+### WebDAV File System Access
+
+The server provides complete file system access to your NextCloud instance, enabling you to:
+
+- Browse any directory structure
+- Read and write files of any type
+- Create and delete directories
+- Manage your NextCloud files directly through LLM interactions
+
+**Usage Examples:**
+
+```python
+# List files in root directory
+await nc_webdav_list_directory("")
+
+# Browse a specific folder
+await nc_webdav_list_directory("Documents/Projects")
+
+# Read a text file
+content = await nc_webdav_read_file("Documents/readme.txt")
+
+# Create a new directory
+await nc_webdav_create_directory("NewProject/docs")
+
+# Write content to a file
+await nc_webdav_write_file("NewProject/docs/notes.md", "# My Notes\n\nContent here...")
+
+# Delete a file or directory
+await nc_webdav_delete_resource("old_file.txt")
+```
+
 ### Note Attachments
 
 This server supports adding and retrieving note attachments via WebDAV. Please note the following behavior regarding attachments:

nextcloud_mcp_server/client/webdav.py

@@ -1,9 +1,10 @@
 """WebDAV client for Nextcloud file operations."""
 
 import mimetypes
-from typing import Tuple, Dict, Any, Optional
+from typing import Tuple, Dict, Any, Optional, List
 import logging
 from httpx import HTTPStatusError
+import xml.etree.ElementTree as ET
 
 from .base import BaseNextcloudClient
 
@@ -242,3 +243,192 @@ async def get_note_attachment(
                 f"Unexpected error fetching attachment '{filename}' for note {note_id}: {e}"
             )
             raise e
+
+    async def list_directory(self, path: str = "") -> List[Dict[str, Any]]:
+        """List files and directories in the specified path via WebDAV PROPFIND."""
+        webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
+        if not webdav_path.endswith("/"):
+            webdav_path += "/"
+
+        logger.info(f"Listing directory: {webdav_path}")
+
+        propfind_body = """<?xml version="1.0"?>
+        <d:propfind xmlns:d="DAV:">
+            <d:prop>
+                <d:displayname/>
+                <d:getcontentlength/>
+                <d:getcontenttype/>
+                <d:getlastmodified/>
+                <d:resourcetype/>
+            </d:prop>
+        </d:propfind>"""
+
+        headers = {"Depth": "1", "Content-Type": "text/xml", "OCS-APIRequest": "true"}
+
+        try:
+            response = await self._client.request(
+                "PROPFIND", webdav_path, content=propfind_body, headers=headers
+            )
+            response.raise_for_status()
+
+            # Parse the XML response
+            root = ET.fromstring(response.content)
+            items = []
+
+            # Skip the first response (the directory itself)
+            responses = root.findall(".//{DAV:}response")[1:]
+
+            for response_elem in responses:
+                href = response_elem.find(".//{DAV:}href")
+                if href is None:
+                    continue
+
+                # Extract file/directory name from href
+                href_text = href.text or ""
+                name = href_text.rstrip("/").split("/")[-1]
+                if not name:
+                    continue
+
+                # Get properties
+                propstat = response_elem.find(".//{DAV:}propstat")
+                if propstat is None:
+                    continue
+
+                prop = propstat.find(".//{DAV:}prop")
+                if prop is None:
+                    continue
+
+                # Determine if it's a directory
+                resourcetype = prop.find(".//{DAV:}resourcetype")
+                is_directory = (
+                    resourcetype is not None
+                    and resourcetype.find(".//{DAV:}collection") is not None
+                )
+
+                # Get other properties
+                size_elem = prop.find(".//{DAV:}getcontentlength")
+                size = (
+                    int(size_elem.text)
+                    if size_elem is not None and size_elem.text
+                    else 0
+                )
+
+                content_type_elem = prop.find(".//{DAV:}getcontenttype")
+                content_type = (
+                    content_type_elem.text if content_type_elem is not None else None
+                )
+
+                modified_elem = prop.find(".//{DAV:}getlastmodified")
+                modified = modified_elem.text if modified_elem is not None else None
+
+                items.append(
+                    {
+                        "name": name,
+                        "path": f"{path.rstrip('/')}/{name}" if path else name,
+                        "is_directory": is_directory,
+                        "size": size if not is_directory else None,
+                        "content_type": content_type,
+                        "last_modified": modified,
+                    }
+                )
+
+            logger.info(f"Found {len(items)} items in directory: {webdav_path}")
+            return items
+
+        except HTTPStatusError as e:
+            logger.error(f"HTTP error listing directory '{webdav_path}': {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error listing directory '{webdav_path}': {e}")
+            raise e
+
+    async def read_file(self, path: str) -> Tuple[bytes, str]:
+        """Read a file's content via WebDAV GET."""
+        webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
+
+        logger.info(f"Reading file: {webdav_path}")
+
+        try:
+            response = await self._client.get(webdav_path)
+            response.raise_for_status()
+
+            content = response.content
+            content_type = response.headers.get(
+                "content-type", "application/octet-stream"
+            )
+
+            logger.info(
+                f"Successfully read file '{path}' ({content_type}, {len(content)} bytes)"
+            )
+            return content, content_type
+
+        except HTTPStatusError as e:
+            logger.error(f"HTTP error reading file '{path}': {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error reading file '{path}': {e}")
+            raise e
+
+    async def write_file(
+        self, path: str, content: bytes, content_type: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """Write content to a file via WebDAV PUT."""
+        webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
+
+        logger.info(f"Writing file: {webdav_path}")
+
+        if not content_type:
+            content_type, _ = mimetypes.guess_type(path)
+            if not content_type:
+                content_type = "application/octet-stream"
+
+        headers = {"Content-Type": content_type, "OCS-APIRequest": "true"}
+
+        try:
+            response = await self._client.put(
+                webdav_path, content=content, headers=headers
+            )
+            response.raise_for_status()
+
+            logger.info(
+                f"Successfully wrote file '{path}' (Status: {response.status_code})"
+            )
+            return {"status_code": response.status_code}
+
+        except HTTPStatusError as e:
+            logger.error(f"HTTP error writing file '{path}': {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error writing file '{path}': {e}")
+            raise e
+
+    async def create_directory(self, path: str) -> Dict[str, Any]:
+        """Create a directory via WebDAV MKCOL."""
+        webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
+        if not webdav_path.endswith("/"):
+            webdav_path += "/"
+
+        logger.info(f"Creating directory: {webdav_path}")
+
+        headers = {"OCS-APIRequest": "true"}
+
+        try:
+            response = await self._client.request("MKCOL", webdav_path, headers=headers)
+            response.raise_for_status()
+
+            logger.info(
+                f"Successfully created directory '{path}' (Status: {response.status_code})"
+            )
+            return {"status_code": response.status_code}
+
+        except HTTPStatusError as e:
+            if (
+                e.response.status_code == 405
+            ):  # Method Not Allowed - directory already exists
+                logger.info(f"Directory '{path}' already exists")
+                return {"status_code": 405, "message": "Directory already exists"}
+            logger.error(f"HTTP error creating directory '{path}': {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error creating directory '{path}': {e}")
+            raise e