feat: add sharing API client and server tools

nextcloud_mcp_server/client/__init__.py

@@ -16,6 +16,7 @@ from .calendar import CalendarClient
 from .contacts import ContactsClient
 from .deck import DeckClient
 from .notes import NotesClient
+from .sharing import SharingClient
 from .tables import TablesClient
 from .webdav import WebDAVClient
 from .users import UsersClient
@@ -73,6 +74,7 @@ class NextcloudClient:
         self.contacts = ContactsClient(self._client, username)
         self.deck = DeckClient(self._client, username)
         self.users = UsersClient(self._client, username)
+        self.sharing = SharingClient(self._client, username)
 
         # Initialize controllers
         self._notes_search = NotesSearchController()

nextcloud_mcp_server/client/sharing.py

@@ -0,0 +1,194 @@
+"""Nextcloud OCS Sharing API client for file/folder sharing operations."""
+
+import logging
+from typing import Any
+
+from .base import BaseNextcloudClient, retry_on_429
+
+logger = logging.getLogger(__name__)
+
+
+class SharingClient(BaseNextcloudClient):
+    """Client for Nextcloud OCS Sharing API operations."""
+
+    @retry_on_429
+    async def create_share(
+        self,
+        path: str,
+        share_with: str,
+        share_type: int = 0,
+        permissions: int = 1,
+    ) -> dict[str, Any]:
+        """Create a share for a file or folder.
+
+        Args:
+            path: Path to file/folder to share (relative to user's files)
+            share_with: Username (for user share) or group name (for group share)
+            share_type: Share type (0=user, 1=group, 3=public link)
+            permissions: Share permissions:
+                - 1 = read
+                - 2 = update
+                - 4 = create
+                - 8 = delete
+                - 16 = share
+                - 31 = all permissions
+                Common combinations: 1 (read-only), 3 (read+update), 15 (read+update+create+delete)
+
+        Returns:
+            Share data including share ID
+
+        Raises:
+            HTTPStatusError: If the request fails
+        """
+        response = await self._client.post(
+            "/ocs/v2.php/apps/files_sharing/api/v1/shares",
+            headers={"OCS-APIRequest": "true", "Accept": "application/json"},
+            data={
+                "path": path,
+                "shareType": share_type,
+                "shareWith": share_with,
+                "permissions": permissions,
+            },
+        )
+        response.raise_for_status()
+        data = response.json()
+
+        if data["ocs"]["meta"]["statuscode"] not in (100, 200):
+            raise RuntimeError(
+                f"OCS API error: {data['ocs']['meta'].get('message', 'Unknown error')}"
+            )
+
+        share_data = data["ocs"]["data"]
+        logger.info(
+            f"Created share {share_data['id']}: {path} -> {share_with} "
+            f"(type={share_type}, permissions={permissions})"
+        )
+        return share_data
+
+    @retry_on_429
+    async def delete_share(self, share_id: int) -> None:
+        """Delete a share by its ID.
+
+        Args:
+            share_id: The share ID to delete
+
+        Raises:
+            HTTPStatusError: If the request fails
+        """
+        response = await self._client.delete(
+            f"/ocs/v2.php/apps/files_sharing/api/v1/shares/{share_id}",
+            headers={"OCS-APIRequest": "true", "Accept": "application/json"},
+        )
+        response.raise_for_status()
+        data = response.json()
+
+        if data["ocs"]["meta"]["statuscode"] not in (100, 200):
+            raise RuntimeError(
+                f"OCS API error: {data['ocs']['meta'].get('message', 'Unknown error')}"
+            )
+
+        logger.info(f"Deleted share {share_id}")
+
+    @retry_on_429
+    async def get_share(self, share_id: int) -> dict[str, Any]:
+        """Get information about a specific share.
+
+        Args:
+            share_id: The share ID
+
+        Returns:
+            Share data
+
+        Raises:
+            HTTPStatusError: If the request fails
+        """
+        response = await self._client.get(
+            f"/ocs/v2.php/apps/files_sharing/api/v1/shares/{share_id}",
+            headers={"OCS-APIRequest": "true", "Accept": "application/json"},
+        )
+        response.raise_for_status()
+        data = response.json()
+
+        if data["ocs"]["meta"]["statuscode"] not in (100, 200):
+            raise RuntimeError(
+                f"OCS API error: {data['ocs']['meta'].get('message', 'Unknown error')}"
+            )
+
+        return data["ocs"]["data"]
+
+    @retry_on_429
+    async def list_shares(
+        self, path: str | None = None, shared_with_me: bool = False
+    ) -> list[dict[str, Any]]:
+        """List shares.
+
+        Args:
+            path: Optional path to filter shares for a specific file/folder
+            shared_with_me: If True, list shares shared with the current user
+
+        Returns:
+            List of share data
+
+        Raises:
+            HTTPStatusError: If the request fails
+        """
+        params = {}
+        if path:
+            params["path"] = path
+        if shared_with_me:
+            params["shared_with_me"] = "true"
+
+        response = await self._client.get(
+            "/ocs/v2.php/apps/files_sharing/api/v1/shares",
+            params=params,
+            headers={"OCS-APIRequest": "true", "Accept": "application/json"},
+        )
+        response.raise_for_status()
+        data = response.json()
+
+        if data["ocs"]["meta"]["statuscode"] not in (100, 200):
+            raise RuntimeError(
+                f"OCS API error: {data['ocs']['meta'].get('message', 'Unknown error')}"
+            )
+
+        # Handle both single share and list of shares
+        shares_data = data["ocs"]["data"]
+        if isinstance(shares_data, dict):
+            return [shares_data]
+        return shares_data if shares_data else []
+
+    @retry_on_429
+    async def update_share(
+        self, share_id: int, permissions: int | None = None
+    ) -> dict[str, Any]:
+        """Update a share's permissions.
+
+        Args:
+            share_id: The share ID to update
+            permissions: New permissions value (see create_share for values)
+
+        Returns:
+            Updated share data
+
+        Raises:
+            HTTPStatusError: If the request fails
+        """
+        data = {}
+        if permissions is not None:
+            data["permissions"] = permissions
+
+        response = await self._client.put(
+            f"/ocs/v2.php/apps/files_sharing/api/v1/shares/{share_id}",
+            headers={"OCS-APIRequest": "true", "Accept": "application/json"},
+            data=data,
+        )
+        response.raise_for_status()
+        result = response.json()
+
+        if result["ocs"]["meta"]["statuscode"] not in (100, 200):
+            raise RuntimeError(
+                f"OCS API error: {result['ocs']['meta'].get('message', 'Unknown error')}"
+            )
+
+        logger.info(f"Updated share {share_id}")
+        return result["ocs"]["data"]