Merge pull request #92 from neovasky/master

feat(webdav): add complete file system support
This commit is contained in:
Chris Coutinho
2025-07-26 13:28:12 +02:00
committed by GitHub
6 changed files with 682 additions and 5 deletions
+4
View File
@@ -1,2 +1,6 @@
__pycache__/
.coverage
.env
*.env
.env.local
.env.*.local
+23
View File
@@ -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
+42
View File
@@ -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:
+191 -1
View File
@@ -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 = """<?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
+151 -4
View File
@@ -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()
+271
View File
@@ -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")