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

fix: address PR review feedback (round 2)

Browse Source 
Bug fixes:
- Catch OCSError/HTTPStatusError in all server tools, convert to McpError
- Guard update_collective against empty body (raise ValueError)
- Use restore_page response data in status message

ADR-017 annotation fix:
- Distinguish "remove" (reversible association) from "delete" (permanent):
  remove_tag and deck_remove_label_from_card no longer set destructiveHint
- Update annotation test to exclude "remove" from destructive keywords

Data model improvements:
- Add trashTimestamp field to PageInfo
- Create ListTrashedPagesResponse with is_trash context flag
- Add collective_id to ListTagsResponse

Test robustness:
- Read NC credentials from environment variables (not hardcoded)
- Filter landing page by parentId == 0 instead of assuming pages[0]

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-03-25 09:08:13 +01:00
co-authored by Claude Opus 4.6
parent 3393cd9756
commit 44a27bd9e9
6 changed files with 125 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/client/collectives.py
+7 -1
View File
@@ -71,10 +71,16 @@ class CollectivesClient(BaseNextcloudClient):
async def update_collective(
self, collective_id: int, emoji: str | None = None
) -> dict[str, Any]:
"""Update a collective (emoji)."""
"""Update a collective (emoji).
Raises:
ValueError: If no fields are provided to update.
"""
json_data: dict[str, Any] = {}
if emoji is not None:
json_data["emoji"] = emoji
if not json_data:
raise ValueError("At least one field must be provided to update")
response = await self._make_request(
"PUT",
f"{API_BASE}/collectives/{collective_id}",
nextcloud_mcp_server/models/collectives.py
+12
View File
@@ -42,6 +42,9 @@ class PageInfo(BaseModel):
default_factory=list, description="Ordered subpage IDs"
)
isFullWidth: bool | None = Field(default=None, description="Full-width page layout")
trashTimestamp: int | None = Field(
default=None, description="Timestamp when the page was trashed"
)
class CollectiveTag(BaseModel):
@@ -120,11 +123,20 @@ class SearchPagesResponse(BaseResponse):
collective_id: int = Field(description="Collective ID")
class ListTrashedPagesResponse(ListPagesResponse):
"""Response for listing trashed pages in a collective."""
is_trash: bool = Field(
default=True, description="Indicates these are trashed pages"
)
class ListTagsResponse(BaseResponse):
"""Response for listing tags in a collective."""
tags: list[CollectiveTag] = Field(description="List of tags")
total: int = Field(description="Total number of tags")
collective_id: int = Field(description="Collective ID")
class CreateTagResponse(BaseResponse):
nextcloud_mcp_server/server/collectives.py
+90 -28
View File
@@ -4,9 +4,11 @@ import logging
from httpx import HTTPStatusError
from mcp.server.fastmcp import Context, FastMCP
from mcp.types import ToolAnnotations
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData, ToolAnnotations
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.client.collectives import OCSError
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models.collectives import (
Collective,
@@ -19,6 +21,7 @@ from nextcloud_mcp_server.models.collectives import (
ListCollectivesResponse,
ListPagesResponse,
ListTagsResponse,
ListTrashedPagesResponse,
PageInfo,
PageOperationResponse,
SearchPagesResponse,
@@ -28,6 +31,13 @@ from nextcloud_mcp_server.observability.metrics import instrument_tool
logger = logging.getLogger(__name__)
def _handle_collectives_error(e: OCSError | HTTPStatusError) -> McpError:
"""Convert OCS or HTTP errors to McpError."""
if isinstance(e, OCSError):
return McpError(ErrorData(code=e.status_code, message=e.message))
return McpError(ErrorData(code=e.response.status_code, message=str(e)))
def configure_collectives_tools(mcp: FastMCP):
"""Configure Nextcloud Collectives tools for the MCP server."""
@@ -44,7 +54,10 @@ def configure_collectives_tools(mcp: FastMCP):
) -> ListCollectivesResponse:
"""List all Nextcloud Collectives the user has access to"""
client = await get_client(ctx)
raw_collectives = await client.collectives.get_collectives()
try:
raw_collectives = await client.collectives.get_collectives()
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
collectives = [Collective(**c) for c in raw_collectives]
return ListCollectivesResponse(collectives=collectives, total=len(collectives))
@@ -63,7 +76,10 @@ def configure_collectives_tools(mcp: FastMCP):
collective_id: ID of the collective
"""
client = await get_client(ctx)
raw_pages = await client.collectives.get_pages(collective_id)
try:
raw_pages = await client.collectives.get_pages(collective_id)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
pages = [PageInfo(**p) for p in raw_pages]
return ListPagesResponse(
pages=pages, total=len(pages), collective_id=collective_id
@@ -89,7 +105,10 @@ def configure_collectives_tools(mcp: FastMCP):
page_id: ID of the page
"""
client = await get_client(ctx)
raw_page = await client.collectives.get_page(collective_id, page_id)
try:
raw_page = await client.collectives.get_page(collective_id, page_id)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
page = PageInfo(**raw_page)
# Fetch content via WebDAV
@@ -130,7 +149,10 @@ def configure_collectives_tools(mcp: FastMCP):
query: Search query string
"""
client = await get_client(ctx)
raw_pages = await client.collectives.search_pages(collective_id, query)
try:
raw_pages = await client.collectives.search_pages(collective_id, query)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
pages = [PageInfo(**p) for p in raw_pages]
return SearchPagesResponse(
results=pages,
@@ -154,9 +176,12 @@ def configure_collectives_tools(mcp: FastMCP):
collective_id: ID of the collective
"""
client = await get_client(ctx)
raw_tags = await client.collectives.get_tags(collective_id)
try:
raw_tags = await client.collectives.get_tags(collective_id)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
tags = [CollectiveTag(**t) for t in raw_tags]
return ListTagsResponse(tags=tags, total=len(tags))
return ListTagsResponse(tags=tags, total=len(tags), collective_id=collective_id)
@mcp.tool(
title="List Trashed Collective Pages",
@@ -166,16 +191,19 @@ def configure_collectives_tools(mcp: FastMCP):
@instrument_tool
async def collectives_get_trashed_pages(
ctx: Context, collective_id: int
) -> ListPagesResponse:
) -> ListTrashedPagesResponse:
"""List trashed pages in a Nextcloud Collective
Args:
collective_id: ID of the collective
"""
client = await get_client(ctx)
raw_pages = await client.collectives.get_trashed_pages(collective_id)
try:
raw_pages = await client.collectives.get_trashed_pages(collective_id)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
pages = [PageInfo(**p) for p in raw_pages]
return ListPagesResponse(
return ListTrashedPagesResponse(
pages=pages, total=len(pages), collective_id=collective_id
)
@@ -197,7 +225,10 @@ def configure_collectives_tools(mcp: FastMCP):
emoji: Optional emoji for the collective
"""
client = await get_client(ctx)
raw = await client.collectives.create_collective(name, emoji)
try:
raw = await client.collectives.create_collective(name, emoji)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
collective = Collective(**raw)
return CreateCollectiveResponse(
id=collective.id, name=collective.name, emoji=collective.emoji
@@ -219,7 +250,12 @@ def configure_collectives_tools(mcp: FastMCP):
emoji: New emoji for the collective
"""
client = await get_client(ctx)
raw = await client.collectives.update_collective(collective_id, emoji)
try:
raw = await client.collectives.update_collective(collective_id, emoji)
except ValueError as e:
raise McpError(ErrorData(code=400, message=str(e))) from e
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
collective = Collective(**raw)
return CollectiveOperationResponse(
collective_id=collective.id,
@@ -248,7 +284,10 @@ def configure_collectives_tools(mcp: FastMCP):
title: Title of the new page
"""
client = await get_client(ctx)
raw = await client.collectives.create_page(collective_id, parent_id, title)
try:
raw = await client.collectives.create_page(collective_id, parent_id, title)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
page = PageInfo(**raw)
return CreatePageResponse(
id=page.id,
@@ -283,9 +322,12 @@ def configure_collectives_tools(mcp: FastMCP):
copy: If true, copy instead of move
"""
client = await get_client(ctx)
raw = await client.collectives.move_page(
collective_id, page_id, parent_id, title, index, copy
)
try:
raw = await client.collectives.move_page(
collective_id, page_id, parent_id, title, index, copy
)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
page = PageInfo(**raw)
action = "copied" if copy else "moved"
return PageOperationResponse(
@@ -313,7 +355,10 @@ def configure_collectives_tools(mcp: FastMCP):
page_id: ID of the page to trash
"""
client = await get_client(ctx)
await client.collectives.trash_page(collective_id, page_id)
try:
await client.collectives.trash_page(collective_id, page_id)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
return PageOperationResponse(
page_id=page_id,
collective_id=collective_id,
@@ -337,12 +382,16 @@ def configure_collectives_tools(mcp: FastMCP):
page_id: ID of the page to restore
"""
client = await get_client(ctx)
await client.collectives.restore_page(collective_id, page_id)
try:
raw = await client.collectives.restore_page(collective_id, page_id)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
page = PageInfo(**raw)
return PageOperationResponse(
page_id=page_id,
page_id=page.id,
collective_id=collective_id,
status_code=200,
message="Page restored from trash",
message=f"Page restored from trash (title: {page.title})",
)
@mcp.tool(
@@ -365,7 +414,10 @@ def configure_collectives_tools(mcp: FastMCP):
emoji: Emoji to set, or null to clear
"""
client = await get_client(ctx)
raw = await client.collectives.set_page_emoji(collective_id, page_id, emoji)
try:
raw = await client.collectives.set_page_emoji(collective_id, page_id, emoji)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
page = PageInfo(**raw)
return PageOperationResponse(
page_id=page.id,
@@ -391,7 +443,10 @@ def configure_collectives_tools(mcp: FastMCP):
color: Hex color code (e.g. "FF0000")
"""
client = await get_client(ctx)
raw = await client.collectives.create_tag(collective_id, name, color)
try:
raw = await client.collectives.create_tag(collective_id, name, color)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
tag = CollectiveTag(**raw)
return CreateTagResponse(id=tag.id, name=tag.name, color=tag.color)
@@ -412,7 +467,10 @@ def configure_collectives_tools(mcp: FastMCP):
tag_id: ID of the tag to assign
"""
client = await get_client(ctx)
await client.collectives.assign_tag(collective_id, page_id, tag_id)
try:
await client.collectives.assign_tag(collective_id, page_id, tag_id)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
return PageOperationResponse(
page_id=page_id,
collective_id=collective_id,
@@ -422,16 +480,17 @@ def configure_collectives_tools(mcp: FastMCP):
@mcp.tool(
title="Remove Tag from Collective Page",
annotations=ToolAnnotations(
destructiveHint=True, idempotentHint=True, openWorldHint=True
),
annotations=ToolAnnotations(idempotentHint=True, openWorldHint=True),
)
@require_scopes("collectives:write")
@instrument_tool
async def collectives_remove_tag(
ctx: Context, collective_id: int, page_id: int, tag_id: int
) -> PageOperationResponse:
"""Remove a tag from a page in a Nextcloud Collective
"""Remove a tag from a page in a Nextcloud Collective.
This is a reversible operation — the tag still exists and can be
reassigned with collectives_assign_tag.
Args:
collective_id: ID of the collective
@@ -439,7 +498,10 @@ def configure_collectives_tools(mcp: FastMCP):
tag_id: ID of the tag to remove
"""
client = await get_client(ctx)
await client.collectives.remove_tag(collective_id, page_id, tag_id)
try:
await client.collectives.remove_tag(collective_id, page_id, tag_id)
except (OCSError, HTTPStatusError) as e:
raise _handle_collectives_error(e) from e
return PageOperationResponse(
page_id=page_id,
collective_id=collective_id,
nextcloud_mcp_server/server/deck.py
+1 -3
View File
@@ -641,9 +641,7 @@ def configure_deck_tools(mcp: FastMCP):
@mcp.tool(
title="Remove Label from Deck Card",
annotations=ToolAnnotations(
destructiveHint=True, idempotentHint=True, openWorldHint=True
),
annotations=ToolAnnotations(idempotentHint=True, openWorldHint=True),
)
@require_scopes("deck:write")
@instrument_tool
tests/server/test_annotations.py
+3 -2
View File
@@ -58,8 +58,9 @@ async def test_destructive_tools_have_correct_annotations(nc_mcp_client: ClientS
"""Verify destructive operations are marked correctly."""
tools = await nc_mcp_client.list_tools()
# Known destructive operations
destructive_keywords = ["delete", "remove", "revoke"]
# Known destructive operations (permanently delete data).
# "remove" is excluded — removing associations (labels, tags) is reversible.
destructive_keywords = ["delete", "revoke"]
for tool in tools.tools:
has_destructive_keyword = any(
tests/server/test_collectives_mcp.py
+12 -6
View File
@@ -2,6 +2,7 @@
import json
import logging
import os
import uuid
import httpx
@@ -11,9 +12,10 @@ from mcp import ClientSession
logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration
# Nextcloud credentials for direct API cleanup (matches docker-compose.yml)
_NC_BASE = "http://localhost:8080"
_NC_AUTH = ("admin", "admin")
# Nextcloud credentials from environment (matches .envrc / docker-compose.yml defaults)
_NC_BASE = os.environ.get("NEXTCLOUD_HOST", "http://localhost:8080")
_NC_USER = os.environ.get("NEXTCLOUD_USERNAME", "admin")
_NC_PASS = os.environ.get("NEXTCLOUD_PASSWORD", "admin")
_OCS_HEADERS = {
"OCS-APIRequest": "true",
"Accept": "application/json",
@@ -38,13 +40,15 @@ async def temporary_collective(nc_mcp_client: ClientSession):
collective_id = data["id"]
logger.info(f"Created temporary collective: {name} (ID: {collective_id})")
# Get the landing page ID (auto-created with each collective)
# Get the landing page ID — filter by parentId == 0 (root page)
pages_result = await nc_mcp_client.call_tool(
"collectives_get_pages",
{"collective_id": collective_id},
)
pages_data = json.loads(pages_result.content[0].text)
landing_page_id = pages_data["pages"][0]["id"]
root_pages = [p for p in pages_data["pages"] if p["parentId"] == 0]
assert root_pages, "Expected at least one root page (landing page)"
landing_page_id = root_pages[0]["id"]
yield {
"id": collective_id,
@@ -54,7 +58,9 @@ async def temporary_collective(nc_mcp_client: ClientSession):
# Cleanup: trash and permanently delete the collective via direct OCS API
try:
async with httpx.AsyncClient(base_url=_NC_BASE, auth=_NC_AUTH) as client:
async with httpx.AsyncClient(
base_url=_NC_BASE, auth=(_NC_USER, _NC_PASS)
) as client:
api = "/ocs/v2.php/apps/collectives/api/v1.0"
await client.delete(
f"{api}/collectives/{collective_id}",