brandon/nextcloud-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(notes): Add append to note functionality

Browse Source
This commit is contained in:
Chris Coutinho
2025-05-24 21:28:10 +02:00
parent dd7eab05db
commit 892e0b4c01
4 changed files with 184 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1
View File
@@ -12,6 +12,7 @@ Currently, the server primarily interacts with the Nextcloud Notes API, providin
* `nc_notes_create_note`: Create a new note.
* `nc_notes_update_note`: Update an existing note by ID.
* `nc_notes_append_content`: Append content to an existing note with a clear separator.
* `nc_notes_delete_note`: Delete a note by ID.
* `nc_notes_search_notes`: Search notes by title or content.
* `nc_get_note`: Get a specific note by ID.
nextcloud_mcp_server/client.py
+45 -17
View File
@@ -159,9 +159,37 @@ class NextcloudClient:
except Exception as e:
logger.error(f"Error cleaning up old attachment directory for note {note_id}: {e}")
# Continue with update even if cleanup failed
return updated_note
def notes_append_content(self, *, note_id: int, content: str):
"""Append content to an existing note with a standard separator"""
logger.info(f"Appending content to note {note_id}")
# Get current note
current_note = self.notes_get_note(note_id=note_id)
# Use fixed separator for consistency
separator = "\n---\n"
# Combine content
existing_content = current_note.get("content", "")
if existing_content:
new_content = existing_content + separator + content
else:
new_content = content # No separator needed for empty notes
logger.info(f"Combining existing content ({len(existing_content)} chars) with new content ({len(content)} chars)")
# Update with combined content
return self.notes_update_note(
note_id=note_id,
etag=current_note["etag"],
content=new_content,
title=None, # Keep existing title
category=None # Keep existing category
)
def notes_search_notes(self, *, query: str):
"""
Search notes using token-based matching with relevance ranking.
@@ -263,7 +291,7 @@ class NextcloudClient:
# Construct path to old attachment directory
old_category_path_part = f"{old_category}/" if old_category else ""
old_attachment_dir_path = f"Notes/{old_category_path_part}.attachments.{note_id}/"
logger.info(f"Cleaning up old attachment directory: {old_attachment_dir_path}")
try:
delete_result = self.delete_webdav_resource(path=old_attachment_dir_path)
@@ -299,7 +327,7 @@ class NextcloudClient:
logger.info(f"Resource '{webdav_path}' doesn't exist, no deletion needed.")
return {"status_code": 404}
# For other errors, continue with deletion attempt
# Proceed with deletion
response = self._client.delete(webdav_path, headers=headers)
response.raise_for_status() # Raises for 4xx/5xx status codes
@@ -308,7 +336,7 @@ class NextcloudClient:
return {"status_code": response.status_code}
except HTTPStatusError as e:
logger.error(
logger.warning(
"HTTP error deleting WebDAV resource '%s': %s",
webdav_path,
e,
@@ -321,7 +349,7 @@ class NextcloudClient:
logger.info("Resource '%s' not found, no deletion needed.", webdav_path)
return {"status_code": 404} # Indicate resource was not found
except Exception as e:
logger.error(
logger.warning(
"Unexpected error deleting WebDAV resource '%s': %s",
webdav_path,
e,
@@ -334,20 +362,20 @@ class NextcloudClient:
try:
note_details = self.notes_get_note(note_id=note_id)
category = note_details.get("category", "")
# Check for other potential categories (if any note was moved between categories)
# We can't reliably detect this without a dedicated tracking mechanism, but we can
# implement a basic check for common category names and empty category
potential_categories = []
if category:
potential_categories.append(category) # Current category first
# Add empty category (uncategorized notes)
if category != "":
potential_categories.append("")
# We could add logic here to check for other common categories if needed
logger.info(f"Note {note_id} has category: '{category}', will check attachment directories in: {potential_categories}")
except HTTPStatusError as e:
# If note doesn't exist (404), we can't delete attachments anyway.
@@ -380,7 +408,7 @@ class NextcloudClient:
except Exception as e:
# Log the error but don't re-raise, as API note deletion itself was successful
# Also, we want to try other potential categories even if one fails
logger.error(f"Failed during WebDAV deletion for category '{cat}' attachment directory: {e}")
logger.warning(f"Failed during WebDAV deletion for category '{cat}' attachment directory: {e}")
return json_response
@@ -409,7 +437,7 @@ class NextcloudClient:
logger.info(f"Uploading attachment for note {note_id} (category: '{category or ''}') to WebDAV path: {attachment_path}")
# Log current auth settings to diagnose the issue
logger.info("WebDAV auth settings - Username: %s, Auth Type: %s",
logger.info("WebDAV auth settings - Username: %s, Auth Type: %s",
self.username, type(self._client.auth).__name__)
if not mime_type:
@@ -423,7 +451,7 @@ class NextcloudClient:
# by checking the Notes directory
notes_dir_path = f"{webdav_base}/Notes"
logger.info("Testing WebDAV access to Notes directory: %s", notes_dir_path)
# Log details of the auth being used by the client for this specific request
if self._client.auth:
auth_header = self._client.auth.auth_flow(self._client.build_request("GET", notes_dir_path)).__next__().headers.get("Authorization")
@@ -433,9 +461,9 @@ class NextcloudClient:
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
logger.info("Headers for PROPFIND (Notes dir): %s", propfind_headers)
notes_dir_response = self._client.request("PROPFIND", notes_dir_path,
notes_dir_response = self._client.request("PROPFIND", notes_dir_path,
headers=propfind_headers)
if notes_dir_response.status_code == 401:
logger.error("WebDAV authentication failed for Notes directory. Please verify WebDAV permissions.")
raise HTTPStatusError(
@@ -447,9 +475,9 @@ class NextcloudClient:
logger.error("Error accessing WebDAV Notes directory: %s", notes_dir_response.status_code)
notes_dir_response.raise_for_status()
else:
logger.info("Successfully accessed WebDAV Notes directory (Status: %s)",
logger.info("Successfully accessed WebDAV Notes directory (Status: %s)",
notes_dir_response.status_code)
# Ensure the parent directory exists using MKCOL
# parent_dir_path is now determined by the helper method
logger.info("Ensuring attachments directory exists: %s", parent_dir_path)
@@ -465,7 +493,7 @@ class NextcloudClient:
)
mkcol_response.raise_for_status()
else:
logger.info("Created/verified directory: %s (Status: %s)",
logger.info("Created/verified directory: %s (Status: %s)",
parent_dir_path, mkcol_response.status_code)
# Proceed with the PUT request
nextcloud_mcp_server/server.py
+8
View File
@@ -98,6 +98,14 @@ def nc_notes_update_note(
)
@mcp.tool()
def nc_notes_append_content(note_id: int, content: str, ctx: Context):
"""Append content to an existing note with a clear separator"""
logger.info("Appending content to note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
return client.notes_append_content(note_id=note_id, content=content)
@mcp.tool()
def nc_notes_search_notes(query: str, ctx: Context):
"""Search notes by title or content, returning only id, title, and category."""
tests/integration/test_notes_api.py
+130 -4
View File
@@ -19,10 +19,10 @@ def test_notes_api_create_and_read(nc_client: NextcloudClient, temporary_note: d
"""
created_note_data = temporary_note # Get data from fixture
note_id = created_note_data["id"]
logger.info(f"Reading note created by fixture, ID: {note_id}")
read_note = nc_client.notes_get_note(note_id=note_id)
assert read_note["id"] == note_id
assert read_note["title"] == created_note_data["title"]
assert read_note["content"] == created_note_data["content"]
@@ -40,7 +40,7 @@ def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict):
update_title = f"Updated Title {uuid.uuid4().hex[:8]}"
update_content = f"Updated Content {uuid.uuid4().hex[:8]}"
logger.info(f"Attempting to update note ID: {note_id} with etag: {original_etag}")
updated_note = nc_client.notes_update_note(
note_id=note_id,
@@ -50,7 +50,7 @@ def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict):
# category=original_category # Explicitly pass category if required by update
)
logger.info(f"Note updated: {updated_note}")
assert updated_note["id"] == note_id
assert updated_note["title"] == update_title
assert updated_note["content"] == update_content
@@ -111,4 +111,130 @@ def test_notes_api_delete_nonexistent(nc_client: NextcloudClient):
assert excinfo.value.response.status_code == 404
logger.info(f"Deleting non-existent note ID: {non_existent_id} correctly failed with 404.")
def test_notes_api_append_content_to_existing_note(nc_client: NextcloudClient, temporary_note: dict):
"""
Tests appending content to an existing note using the new append functionality.
"""
created_note_data = temporary_note
note_id = created_note_data["id"]
original_content = created_note_data["content"]
append_text = f"Appended content {uuid.uuid4().hex[:8]}"
logger.info(f"Appending content to note ID: {note_id}")
updated_note = nc_client.notes_append_content(
note_id=note_id,
content=append_text
)
logger.info(f"Note after append: {updated_note}")
# Verify the note was updated
assert updated_note["id"] == note_id
assert "etag" in updated_note
assert updated_note["etag"] != created_note_data["etag"] # Etag must change
# Verify content has the separator and appended text
expected_content = original_content + "\n---\n" + append_text
assert updated_note["content"] == expected_content
# Verify by reading the note again
time.sleep(1) # Allow potential propagation delay
read_note = nc_client.notes_get_note(note_id=note_id)
assert read_note["content"] == expected_content
logger.info(f"Successfully appended content to note ID: {note_id}")
def test_notes_api_append_content_to_empty_note(nc_client: NextcloudClient):
"""
Tests appending content to an empty note (no separator should be added).
"""
# Create an empty note
test_title = f"Empty Note {uuid.uuid4().hex[:8]}"
test_category = "Test"
logger.info(f"Creating empty note for append test")
empty_note = nc_client.notes_create_note(
title=test_title,
content="", # Empty content
category=test_category
)
note_id = empty_note["id"]
try:
append_text = f"First content {uuid.uuid4().hex[:8]}"
logger.info(f"Appending content to empty note ID: {note_id}")
updated_note = nc_client.notes_append_content(
note_id=note_id,
content=append_text
)
# For empty notes, content should just be the appended text (no separator)
assert updated_note["content"] == append_text
# Verify by reading the note again
time.sleep(1)
read_note = nc_client.notes_get_note(note_id=note_id)
assert read_note["content"] == append_text
logger.info(f"Successfully appended content to empty note ID: {note_id}")
finally:
# Clean up the test note
try:
nc_client.notes_delete_note(note_id=note_id)
logger.info(f"Cleaned up test note ID: {note_id}")
except Exception as e:
logger.warning(f"Failed to clean up test note ID: {note_id}: {e}")
def test_notes_api_append_content_multiple_times(nc_client: NextcloudClient, temporary_note: dict):
"""
Tests appending content multiple times to verify separator behavior.
"""
created_note_data = temporary_note
note_id = created_note_data["id"]
original_content = created_note_data["content"]
first_append = f"First append {uuid.uuid4().hex[:8]}"
second_append = f"Second append {uuid.uuid4().hex[:8]}"
logger.info(f"Performing multiple appends to note ID: {note_id}")
# First append
updated_note = nc_client.notes_append_content(
note_id=note_id,
content=first_append
)
expected_content_after_first = original_content + "\n---\n" + first_append
assert updated_note["content"] == expected_content_after_first
# Second append
updated_note = nc_client.notes_append_content(
note_id=note_id,
content=second_append
)
expected_content_after_second = expected_content_after_first + "\n---\n" + second_append
assert updated_note["content"] == expected_content_after_second
# Verify by reading the note again
time.sleep(1)
read_note = nc_client.notes_get_note(note_id=note_id)
assert read_note["content"] == expected_content_after_second
logger.info(f"Successfully performed multiple appends to note ID: {note_id}")
def test_notes_api_append_content_nonexistent_note(nc_client: NextcloudClient):
"""
Tests that appending to a non-existent note fails with 404.
"""
non_existent_id = 999999999
logger.info(f"Attempting to append to non-existent note ID: {non_existent_id}")
with pytest.raises(HTTPStatusError) as excinfo:
nc_client.notes_append_content(
note_id=non_existent_id,
content="This should fail"
)
assert excinfo.value.response.status_code == 404
logger.info(f"Appending to non-existent note ID: {non_existent_id} correctly failed with 404.")
# --- Attachment tests moved to test_attachments.py ---