diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 955efac..e4f9b3c 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -21,6 +21,7 @@ from nextcloud_mcp_server.server import ( configure_contacts_tools, configure_deck_tools, configure_notes_tools, + configure_sharing_tools, configure_tables_tools, configure_webdav_tools, ) @@ -375,6 +376,7 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None): "notes": configure_notes_tools, "tables": configure_tables_tools, "webdav": configure_webdav_tools, + "sharing": configure_sharing_tools, "calendar": configure_calendar_tools, "contacts": configure_contacts_tools, "deck": configure_deck_tools, diff --git a/nextcloud_mcp_server/client/__init__.py b/nextcloud_mcp_server/client/__init__.py index be646be..eaba19f 100644 --- a/nextcloud_mcp_server/client/__init__.py +++ b/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() diff --git a/nextcloud_mcp_server/client/sharing.py b/nextcloud_mcp_server/client/sharing.py new file mode 100644 index 0000000..29cdf5c --- /dev/null +++ b/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"] diff --git a/nextcloud_mcp_server/server/__init__.py b/nextcloud_mcp_server/server/__init__.py index 7b3b980..f30b0d2 100644 --- a/nextcloud_mcp_server/server/__init__.py +++ b/nextcloud_mcp_server/server/__init__.py @@ -2,6 +2,7 @@ from .calendar import configure_calendar_tools from .contacts import configure_contacts_tools from .deck import configure_deck_tools from .notes import configure_notes_tools +from .sharing import configure_sharing_tools from .tables import configure_tables_tools from .webdav import configure_webdav_tools @@ -10,6 +11,7 @@ __all__ = [ "configure_contacts_tools", "configure_deck_tools", "configure_notes_tools", + "configure_sharing_tools", "configure_tables_tools", "configure_webdav_tools", ] diff --git a/nextcloud_mcp_server/server/sharing.py b/nextcloud_mcp_server/server/sharing.py new file mode 100644 index 0000000..d1a07a4 --- /dev/null +++ b/nextcloud_mcp_server/server/sharing.py @@ -0,0 +1,133 @@ +"""MCP tools for Nextcloud file/folder sharing operations.""" + +import json + +from nextcloud_mcp_server.context import get_client +from mcp.server.fastmcp import Context, FastMCP + + +def configure_sharing_tools(mcp: FastMCP): + """Configure sharing-related MCP tools. + + Args: + mcp: FastMCP server instance + """ + + @mcp.tool() + async def nc_share_create( + path: str, + share_with: str, + ctx: Context, + share_type: int = 0, + permissions: int = 1, + ) -> str: + """Create a share for a file or folder in Nextcloud. + + Share a file or folder with another user or group. The authenticated user + must own the file/folder being shared. + + Args: + path: Path to file/folder to share (relative to your files, e.g., "/document.txt") + share_with: Username (for user share) or group name (for group share) + share_type: Share type - 0 for user (default), 1 for group, 3 for public link + permissions: Share permissions (default: 1 for read-only): + - 1 = read + - 2 = update + - 4 = create + - 8 = delete + - 16 = share + - 31 = all permissions + Common: 1 (read-only), 3 (read+update), 15 (read+update+create+delete) + + Returns: + JSON string with share information including share ID + """ + client = get_client(ctx) + share_data = await client.sharing.create_share( + path=path, + share_with=share_with, + share_type=share_type, + permissions=permissions, + ) + return json.dumps(share_data, indent=2) + + @mcp.tool() + async def nc_share_delete(share_id: int, ctx: Context) -> str: + """Delete a share by its ID. + + Remove a share that you created. You must be the owner of the share. + + Args: + share_id: The ID of the share to delete + + Returns: + JSON string confirming deletion + """ + client = get_client(ctx) + await client.sharing.delete_share(share_id) + return json.dumps( + {"success": True, "message": f"Share {share_id} deleted"}, indent=2 + ) + + @mcp.tool() + async def nc_share_get(share_id: int, ctx: Context) -> str: + """Get information about a specific share. + + Retrieve details about a share by its ID. You must have access to the share + (either as owner or recipient). + + Args: + share_id: The ID of the share + + Returns: + JSON string with share information + """ + client = get_client(ctx) + share_data = await client.sharing.get_share(share_id) + return json.dumps(share_data, indent=2) + + @mcp.tool() + async def nc_share_list( + ctx: Context, path: str | None = None, shared_with_me: bool = False + ) -> str: + """List shares created by you or shared with you. + + Args: + path: Optional path to filter shares for a specific file/folder + shared_with_me: If True, list shares that others shared with you. + If False (default), list shares you created. + + Returns: + JSON string with list of shares + """ + client = get_client(ctx) + shares = await client.sharing.list_shares( + path=path, shared_with_me=shared_with_me + ) + return json.dumps(shares, indent=2) + + @mcp.tool() + async def nc_share_update(share_id: int, permissions: int, ctx: Context) -> str: + """Update the permissions of an existing share. + + Modify the permissions for a share you created. You must be the owner. + + Args: + share_id: The ID of the share to update + permissions: New permissions value: + - 1 = read + - 2 = update + - 4 = create + - 8 = delete + - 16 = share + - 31 = all permissions + Common: 1 (read-only), 3 (read+update), 15 (read+update+create+delete) + + Returns: + JSON string with updated share information + """ + client = get_client(ctx) + share_data = await client.sharing.update_share( + share_id=share_id, permissions=permissions + ) + return json.dumps(share_data, indent=2)