diff --git a/.gitignore b/.gitignore index c1e64c1..85bf658 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,6 @@ __pycache__/ .coverage +.env +*.env +.env.local +.env.*.local diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f790fe..510a7c8 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 97c1816..5b4a8bf 100644 --- a/README.md +++ b/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: diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index f5649ee..d32061e 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/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 @@ class WebDAVClient(BaseNextcloudClient): 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 = """ + + + + + + + + + """ + + 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 diff --git a/nextcloud_mcp_server/server.py b/nextcloud_mcp_server/server.py index a5edfbd..eef5f12 100644 --- a/nextcloud_mcp_server/server.py +++ b/nextcloud_mcp_server/server.py @@ -38,7 +38,7 @@ logger = logging.getLogger(__name__) @mcp.resource("nc://capabilities") async def nc_get_capabilities(): """Get the Nextcloud Host capabilities""" - ctx = ( + ctx: Context = ( mcp.get_context() ) # https://github.com/modelcontextprotocol/python-sdk/issues/244 client: NextcloudClient = ctx.request_context.lifespan_context.client @@ -48,7 +48,7 @@ async def nc_get_capabilities(): @mcp.resource("notes://settings") async def notes_get_settings(): """Get the Notes App settings""" - ctx = ( + ctx: Context = ( mcp.get_context() ) # https://github.com/modelcontextprotocol/python-sdk/issues/244 client: NextcloudClient = ctx.request_context.lifespan_context.client @@ -133,9 +133,9 @@ async def nc_tables_get_schema(table_id: int, ctx: Context): @mcp.tool() async def nc_tables_read_table( table_id: int, + ctx: Context, limit: int | None = None, offset: int | None = None, - ctx: Context = None, ): """Read rows from a table with optional pagination""" client: NextcloudClient = ctx.request_context.lifespan_context.client @@ -172,7 +172,7 @@ async def nc_tables_delete_row(row_id: int, ctx: Context): @mcp.resource("nc://Notes/{note_id}/attachments/{attachment_filename}") async def nc_notes_get_attachment(note_id: int, attachment_filename: str): """Get a specific attachment from a note""" - ctx = mcp.get_context() + ctx: Context = mcp.get_context() client: NextcloudClient = ctx.request_context.lifespan_context.client # Assuming a method get_note_attachment exists in the client # This method should return the raw content and determine the mime type @@ -191,6 +191,153 @@ async def nc_notes_get_attachment(note_id: int, attachment_filename: str): } +# WebDAV file system tools +@mcp.tool() +async def nc_webdav_list_directory(ctx: Context, path: str = ""): + """List files and directories in the specified NextCloud path. + + Args: + path: Directory path to list (empty string for root directory) + + Returns: + List of items with metadata including name, path, is_directory, size, content_type, last_modified + + Examples: + # List root directory + await nc_webdav_list_directory("") + + # List a specific folder + await nc_webdav_list_directory("Documents/Projects") + """ + client: NextcloudClient = ctx.request_context.lifespan_context.client + return await client.webdav.list_directory(path) + + +@mcp.tool() +async def nc_webdav_read_file(path: str, ctx: Context): + """Read the content of a file from NextCloud. + + Args: + path: Full path to the file to read + + Returns: + Dict with path, content, content_type, size, and encoding (if binary) + Text files are decoded to UTF-8, binary files are base64 encoded + + Examples: + # Read a text file + result = await nc_webdav_read_file("Documents/readme.txt") + print(result['content']) # Decoded text content + + # Read a binary file + result = await nc_webdav_read_file("Images/photo.jpg") + print(result['encoding']) # 'base64' + """ + client: NextcloudClient = ctx.request_context.lifespan_context.client + content, content_type = await client.webdav.read_file(path) + + # For text files, decode content for easier viewing + if content_type and content_type.startswith("text/"): + try: + decoded_content = content.decode("utf-8") + return { + "path": path, + "content": decoded_content, + "content_type": content_type, + "size": len(content), + } + except UnicodeDecodeError: + pass + + # For binary files, return metadata and base64 encoded content + import base64 + + return { + "path": path, + "content": base64.b64encode(content).decode("ascii"), + "content_type": content_type, + "size": len(content), + "encoding": "base64", + } + + +@mcp.tool() +async def nc_webdav_write_file( + path: str, content: str, ctx: Context, content_type: str | None = None +): + """Write content to a file in NextCloud. + + Args: + path: Full path where to write the file + content: File content (text or base64 for binary) + content_type: MIME type (auto-detected if not provided, use 'type;base64' for binary) + + Returns: + Dict with status_code indicating success + + Examples: + # Write a text file + await nc_webdav_write_file("Documents/notes.md", "# My Notes\nContent here...") + + # Write binary data (base64 encoded) + await nc_webdav_write_file("files/data.bin", base64_content, "application/octet-stream;base64") + """ + client: NextcloudClient = ctx.request_context.lifespan_context.client + + # Handle base64 encoded content + if content_type and "base64" in content_type.lower(): + import base64 + + content_bytes = base64.b64decode(content) + content_type = content_type.replace(";base64", "") + else: + content_bytes = content.encode("utf-8") + + return await client.webdav.write_file(path, content_bytes, content_type) + + +@mcp.tool() +async def nc_webdav_create_directory(path: str, ctx: Context): + """Create a directory in NextCloud. + + Args: + path: Full path of the directory to create + + Returns: + Dict with status_code (201 for created, 405 if already exists) + + Examples: + # Create a single directory + await nc_webdav_create_directory("NewProject") + + # Create nested directories (parent must exist) + await nc_webdav_create_directory("Projects/MyApp/docs") + """ + client: NextcloudClient = ctx.request_context.lifespan_context.client + return await client.webdav.create_directory(path) + + +@mcp.tool() +async def nc_webdav_delete_resource(path: str, ctx: Context): + """Delete a file or directory in NextCloud. + + Args: + path: Full path of the file or directory to delete + + Returns: + Dict with status_code indicating result (404 if not found) + + Examples: + # Delete a file + await nc_webdav_delete_resource("old_document.txt") + + # Delete a directory (will delete all contents) + await nc_webdav_delete_resource("temp_folder") + """ + client: NextcloudClient = ctx.request_context.lifespan_context.client + return await client.webdav.delete_resource(path) + + def run(): mcp.run() diff --git a/tests/integration/test_webdav_operations.py b/tests/integration/test_webdav_operations.py new file mode 100644 index 0000000..f95e86f --- /dev/null +++ b/tests/integration/test_webdav_operations.py @@ -0,0 +1,271 @@ +"""Integration tests for WebDAV operations.""" + +import pytest +import logging +import uuid +from httpx import HTTPStatusError + +from nextcloud_mcp_server.client import NextcloudClient + +logger = logging.getLogger(__name__) + +# Mark all tests in this module as integration tests +pytestmark = pytest.mark.integration + + +@pytest.fixture +def test_base_path(): + """Base path for test files/directories.""" + return f"mcp_test_{uuid.uuid4().hex[:8]}" + + +async def test_create_and_delete_directory( + nc_client: NextcloudClient, test_base_path: str +): + """Test creating and deleting directories.""" + test_dir = f"{test_base_path}/test_directory" + + try: + # Create directory + result = await nc_client.webdav.create_directory(test_dir) + assert result["status_code"] == 201 # Created + logger.info(f"Created directory: {test_dir}") + + # Verify directory exists by listing parent + parent_listing = await nc_client.webdav.list_directory(test_base_path) + dir_names = [item["name"] for item in parent_listing] + assert "test_directory" in dir_names + + # Delete directory + delete_result = await nc_client.webdav.delete_resource(test_dir) + assert delete_result["status_code"] in [204, 404] # No Content or Not Found + logger.info(f"Deleted directory: {test_dir}") + + finally: + # Cleanup: ensure directory is deleted + try: + await nc_client.webdav.delete_resource(test_dir) + await nc_client.webdav.delete_resource(test_base_path) + except Exception: + pass + + +async def test_write_read_delete_file(nc_client: NextcloudClient, test_base_path: str): + """Test writing, reading, and deleting files.""" + test_file = f"{test_base_path}/test_file.txt" + test_content = f"Test content {uuid.uuid4().hex}" + + try: + # Create base directory first + await nc_client.webdav.create_directory(test_base_path) + + # Write file + write_result = await nc_client.webdav.write_file( + test_file, test_content.encode("utf-8"), content_type="text/plain" + ) + assert write_result["status_code"] in [200, 201, 204] # Success codes + logger.info(f"Wrote file: {test_file}") + + # Read file back + content, content_type = await nc_client.webdav.read_file(test_file) + assert content.decode("utf-8") == test_content + assert content_type == "text/plain" + logger.info(f"Read file: {test_file}") + + # Verify file appears in directory listing + listing = await nc_client.webdav.list_directory(test_base_path) + file_names = [item["name"] for item in listing] + assert "test_file.txt" in file_names + + # Delete file + delete_result = await nc_client.webdav.delete_resource(test_file) + assert delete_result["status_code"] in [204, 404] # No Content or Not Found + logger.info(f"Deleted file: {test_file}") + + finally: + # Cleanup + try: + await nc_client.webdav.delete_resource(test_file) + await nc_client.webdav.delete_resource(test_base_path) + except Exception: + pass + + +async def test_list_directory_empty_and_populated( + nc_client: NextcloudClient, test_base_path: str +): + """Test listing empty and populated directories.""" + try: + # Create base directory + await nc_client.webdav.create_directory(test_base_path) + + # List empty directory + empty_listing = await nc_client.webdav.list_directory(test_base_path) + assert isinstance(empty_listing, list) + assert len(empty_listing) == 0 + logger.info(f"Empty directory listing: {len(empty_listing)} items") + + # Add some files and directories + await nc_client.webdav.create_directory(f"{test_base_path}/subdir1") + await nc_client.webdav.create_directory(f"{test_base_path}/subdir2") + await nc_client.webdav.write_file( + f"{test_base_path}/file1.txt", b"content1", content_type="text/plain" + ) + await nc_client.webdav.write_file( + f"{test_base_path}/file2.md", + b"# Markdown content", + content_type="text/markdown", + ) + + # List populated directory + populated_listing = await nc_client.webdav.list_directory(test_base_path) + assert len(populated_listing) == 4 # 2 dirs + 2 files + + # Check that we have both files and directories + names = [item["name"] for item in populated_listing] + assert "subdir1" in names + assert "subdir2" in names + assert "file1.txt" in names + assert "file2.md" in names + + # Check metadata is present + for item in populated_listing: + assert "name" in item + assert "path" in item + assert "is_directory" in item + assert "size" in item + assert "content_type" in item + assert "last_modified" in item + + logger.info(f"Populated directory listing: {len(populated_listing)} items") + + finally: + # Cleanup + try: + await nc_client.webdav.delete_resource(f"{test_base_path}/file1.txt") + await nc_client.webdav.delete_resource(f"{test_base_path}/file2.md") + await nc_client.webdav.delete_resource(f"{test_base_path}/subdir1") + await nc_client.webdav.delete_resource(f"{test_base_path}/subdir2") + await nc_client.webdav.delete_resource(test_base_path) + except Exception: + pass + + +async def test_read_nonexistent_file(nc_client: NextcloudClient): + """Test reading a file that doesn't exist.""" + nonexistent_file = f"nonexistent_{uuid.uuid4().hex}.txt" + + with pytest.raises(HTTPStatusError) as exc_info: + await nc_client.webdav.read_file(nonexistent_file) + + assert exc_info.value.response.status_code == 404 + logger.info(f"Correctly got 404 for nonexistent file: {nonexistent_file}") + + +async def test_delete_nonexistent_resource(nc_client: NextcloudClient): + """Test deleting a resource that doesn't exist.""" + nonexistent_resource = f"nonexistent_{uuid.uuid4().hex}" + + result = await nc_client.webdav.delete_resource(nonexistent_resource) + assert result["status_code"] == 404 + logger.info(f"Correctly got 404 for nonexistent resource: {nonexistent_resource}") + + +async def test_create_nested_directories( + nc_client: NextcloudClient, test_base_path: str +): + """Test creating nested directory structures.""" + nested_path = f"{test_base_path}/level1/level2/level3" + + try: + # Create nested directories (should create parent directories automatically) + result = await nc_client.webdav.create_directory(nested_path) + assert result["status_code"] == 201 + + # Verify the structure was created + level1_listing = await nc_client.webdav.list_directory( + f"{test_base_path}/level1" + ) + assert len(level1_listing) == 1 + assert level1_listing[0]["name"] == "level2" + assert level1_listing[0]["is_directory"] is True + + level2_listing = await nc_client.webdav.list_directory( + f"{test_base_path}/level1/level2" + ) + assert len(level2_listing) == 1 + assert level2_listing[0]["name"] == "level3" + assert level2_listing[0]["is_directory"] is True + + logger.info(f"Created nested directory structure: {nested_path}") + + finally: + # Cleanup - delete from deepest to shallowest + try: + await nc_client.webdav.delete_resource(nested_path) + await nc_client.webdav.delete_resource(f"{test_base_path}/level1/level2") + await nc_client.webdav.delete_resource(f"{test_base_path}/level1") + await nc_client.webdav.delete_resource(test_base_path) + except Exception: + pass + + +async def test_overwrite_existing_file(nc_client: NextcloudClient, test_base_path: str): + """Test overwriting an existing file.""" + test_file = f"{test_base_path}/overwrite_test.txt" + original_content = "Original content" + new_content = "New content after overwrite" + + try: + # Create base directory + await nc_client.webdav.create_directory(test_base_path) + + # Write original file + await nc_client.webdav.write_file( + test_file, original_content.encode("utf-8"), content_type="text/plain" + ) + + # Verify original content + content, _ = await nc_client.webdav.read_file(test_file) + assert content.decode("utf-8") == original_content + + # Overwrite with new content + overwrite_result = await nc_client.webdav.write_file( + test_file, new_content.encode("utf-8"), content_type="text/plain" + ) + assert overwrite_result["status_code"] in [200, 204] # OK or No Content + + # Verify new content + content, _ = await nc_client.webdav.read_file(test_file) + assert content.decode("utf-8") == new_content + + logger.info(f"Successfully overwrote file: {test_file}") + + finally: + # Cleanup + try: + await nc_client.webdav.delete_resource(test_file) + await nc_client.webdav.delete_resource(test_base_path) + except Exception: + pass + + +async def test_list_root_directory(nc_client: NextcloudClient): + """Test listing the root directory.""" + root_listing = await nc_client.webdav.list_directory("") + + # Root directory should exist and be listable + assert isinstance(root_listing, list) + # Should have at least some default folders/files + assert len(root_listing) >= 0 + + # Check structure of items + for item in root_listing: + assert "name" in item + assert "path" in item + assert "is_directory" in item + assert "size" in item + assert "content_type" in item + assert "last_modified" in item + + logger.info(f"Root directory contains {len(root_listing)} items")