fix: add trash/delete collective tools and address review feedback (round 4)
Add collectives_trash_collective and collectives_delete_collective MCP tools with proper destructiveHint annotations. Refactor integration test fixture to use MCP tools for cleanup instead of direct httpx/OCS calls. Optimize _get_ocs_headers() to class-level constant. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
f3caad122d
commit
95edd9ba8e
@@ -24,13 +24,15 @@ class CollectivesClient(BaseNextcloudClient):
|
||||
|
||||
app_name = "collectives"
|
||||
|
||||
_OCS_HEADERS: dict[str, str] = {
|
||||
"OCS-APIRequest": "true",
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
|
||||
def _get_ocs_headers(self) -> dict[str, str]:
|
||||
"""Get standard headers required for OCS API calls."""
|
||||
return {
|
||||
"OCS-APIRequest": "true",
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
return self._OCS_HEADERS
|
||||
|
||||
def _unwrap_ocs(self, response_json: dict[str, Any]) -> Any:
|
||||
"""Unwrap OCS envelope, validating the status before returning data."""
|
||||
@@ -90,6 +92,26 @@ class CollectivesClient(BaseNextcloudClient):
|
||||
data = self._unwrap_ocs(response.json())
|
||||
return data["collective"]
|
||||
|
||||
async def trash_collective(self, collective_id: int) -> None:
|
||||
"""Move a collective to trash (soft delete)."""
|
||||
await self._make_request(
|
||||
"DELETE",
|
||||
f"{API_BASE}/collectives/{collective_id}",
|
||||
headers=self._get_ocs_headers(),
|
||||
)
|
||||
|
||||
async def delete_collective(self, collective_id: int) -> None:
|
||||
"""Permanently delete a collective (must be trashed first).
|
||||
|
||||
This is irreversible. The collective must be in the trash before
|
||||
calling this method.
|
||||
"""
|
||||
await self._make_request(
|
||||
"DELETE",
|
||||
f"{API_BASE}/collectives/trash/{collective_id}",
|
||||
headers=self._get_ocs_headers(),
|
||||
)
|
||||
|
||||
# Pages
|
||||
|
||||
async def get_pages(self, collective_id: int) -> list[dict[str, Any]]:
|
||||
|
||||
@@ -265,6 +265,66 @@ def configure_collectives_tools(mcp: FastMCP):
|
||||
message=f"Collective updated (emoji: {collective.emoji})",
|
||||
)
|
||||
|
||||
@mcp.tool(
|
||||
title="Trash Collective",
|
||||
annotations=ToolAnnotations(
|
||||
destructiveHint=True, idempotentHint=False, openWorldHint=True
|
||||
),
|
||||
)
|
||||
@require_scopes("collectives:write")
|
||||
@instrument_tool
|
||||
async def collectives_trash_collective(
|
||||
ctx: Context, collective_id: int
|
||||
) -> CollectiveOperationResponse:
|
||||
"""Move a Nextcloud Collective to trash (soft delete).
|
||||
|
||||
The collective can be restored or permanently deleted afterwards.
|
||||
|
||||
Args:
|
||||
collective_id: ID of the collective to trash
|
||||
"""
|
||||
client = await get_client(ctx)
|
||||
try:
|
||||
await client.collectives.trash_collective(collective_id)
|
||||
except (OCSError, HTTPStatusError) as e:
|
||||
raise _handle_collectives_error(e) from e
|
||||
return CollectiveOperationResponse(
|
||||
collective_id=collective_id,
|
||||
status_code=200,
|
||||
message="Collective moved to trash",
|
||||
)
|
||||
|
||||
@mcp.tool(
|
||||
title="Delete Collective",
|
||||
annotations=ToolAnnotations(
|
||||
destructiveHint=True, idempotentHint=True, openWorldHint=True
|
||||
),
|
||||
)
|
||||
@require_scopes("collectives:write")
|
||||
@instrument_tool
|
||||
async def collectives_delete_collective(
|
||||
ctx: Context, collective_id: int
|
||||
) -> CollectiveOperationResponse:
|
||||
"""Permanently delete a Nextcloud Collective.
|
||||
|
||||
WARNING: This is irreversible. The collective must be in the trash
|
||||
first (use collectives_trash_collective). All pages and content
|
||||
will be permanently destroyed.
|
||||
|
||||
Args:
|
||||
collective_id: ID of the trashed collective to permanently delete
|
||||
"""
|
||||
client = await get_client(ctx)
|
||||
try:
|
||||
await client.collectives.delete_collective(collective_id)
|
||||
except (OCSError, HTTPStatusError) as e:
|
||||
raise _handle_collectives_error(e) from e
|
||||
return CollectiveOperationResponse(
|
||||
collective_id=collective_id,
|
||||
status_code=200,
|
||||
message="Collective permanently deleted",
|
||||
)
|
||||
|
||||
@mcp.tool(
|
||||
title="Create Collective Page",
|
||||
annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True),
|
||||
|
||||
Reference in New Issue
Block a user