Add limit/offset pagination to unbounded list tools

Tools updated: list_directory, list_conversations, get_events,
list_shares, list_notifications, list_trash, list_collectives,
get_collective_pages. All now accept limit/offset params and return
{"data": [...], "pagination": {count, offset, limit, has_more}}
matching the existing pattern used by search_files and get_activity.

Default limit is 50 (prevents context overflow on busy instances).
Client-side slicing since most Nextcloud APIs don't support
server-side pagination for these endpoints.

Author: bigcat88

src/nc_mcp_server/tools/calendar.py

@@ -366,6 +366,8 @@ async def get_events(
         calendar_id: str = "personal",
         start: str = "",
         end: str = "",
+        limit: int = 50,
+        offset: int = 0,
     ) -> str:
         """Get events from a calendar, optionally filtered by time range.
 
@@ -379,13 +381,17 @@ async def get_events(
                    Required if end is provided.
             end: Optional range end in ISO 8601 UTC format: "2026-04-30T23:59:59Z".
                  Required if start is provided.
+            limit: Maximum number of events to return (1-500, default 50).
+            offset: Number of events to skip for pagination (default 0).
 
         Returns:
-            JSON list of event objects with: uid, summary, dtstart, dtend, location,
-            description, status, all_day, and optionally rrule and categories.
+            JSON with "data" (list of event objects) and "pagination"
+            (count, offset, limit, has_more).
         """
         if bool(start) != bool(end):
             raise ValueError("Both start and end are required for time-range filtering, or omit both.")
+        limit = max(1, min(500, limit))
+        offset = max(0, offset)
         caldav_start = start.replace("-", "").replace(":", "").replace(".", "") if start else None
         caldav_end = end.replace("-", "").replace(":", "").replace(".", "") if end else None
         if caldav_start:
@@ -405,12 +411,20 @@ async def get_events(
             context=f"Get events from '{calendar_id}'",
         )
         results = _parse_report_xml(response.text or "")
-        events = []
+        all_events = []
         for _href, etag, ical_data in results:
             event = _format_event(ical_data)
             event["etag"] = etag
-            events.append(event)
-        return json.dumps(events)
+            all_events.append(event)
+        page = all_events[offset : offset + limit]
+        has_more = offset + limit < len(all_events)
+
+        return json.dumps(
+            {
+                "data": page,
+                "pagination": {"count": len(page), "offset": offset, "limit": limit, "has_more": has_more},
+            }
+        )
 
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)

src/nc_mcp_server/tools/collectives.py

@@ -42,38 +42,66 @@ def _format_page(p: dict[str, Any]) -> dict[str, Any]:
 def _register_read_tools(mcp: FastMCP) -> None:
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
-    async def list_collectives() -> str:
-        """List all collectives the current user has access to.
+    async def list_collectives(limit: int = 50, offset: int = 0) -> str:
+        """List collectives the current user has access to.
 
         Collectives are shared knowledge bases with wiki-style pages.
-        Each collective has a landing page and may contain nested subpages.
+
+        Args:
+            limit: Maximum number of collectives to return (1-200, default 50).
+            offset: Number of collectives to skip for pagination (default 0).
 
         Returns:
-            JSON list of collectives with id, name, emoji, permissions.
+            JSON with "data" (list of collectives with id, name, emoji, permissions)
+            and "pagination" (count, offset, limit, has_more).
         """
+        limit = max(1, min(200, limit))
+        offset = max(0, offset)
         client = get_client()
         data = await client.ocs_get(f"{API}/collectives")
-        collectives = [_format_collective(c) for c in data["collectives"]]
-        return json.dumps(collectives, default=str)
+        all_collectives = [_format_collective(c) for c in data["collectives"]]
+        page = all_collectives[offset : offset + limit]
+        has_more = offset + limit < len(all_collectives)
+
+        return json.dumps(
+            {
+                "data": page,
+                "pagination": {"count": len(page), "offset": offset, "limit": limit, "has_more": has_more},
+            },
+            default=str,
+        )
 
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
-    async def get_collective_pages(collective_id: int) -> str:
-        """List all pages in a collective.
+    async def get_collective_pages(collective_id: int, limit: int = 50, offset: int = 0) -> str:
+        """List pages in a collective.
 
-        Returns the full page tree including the landing page and all subpages.
-        Each page has a title, emoji, timestamp, size, and file path.
+        Returns the page tree including the landing page and all subpages.
 
         Args:
             collective_id: The numeric collective ID. Use list_collectives to find IDs.
+            limit: Maximum number of pages to return (1-200, default 50).
+            offset: Number of pages to skip for pagination (default 0).
 
         Returns:
-            JSON list of pages with id, title, emoji, timestamp, size, file_name, file_path.
+            JSON with "data" (list of pages with id, title, emoji, timestamp, size)
+            and "pagination" (count, offset, limit, has_more).
         """
+        limit = max(1, min(200, limit))
+        offset = max(0, offset)
         client = get_client()
         data = await client.ocs_get(f"{API}/collectives/{collective_id}/pages")
-        pages = [_format_page(p) for p in data["pages"]]
-        return json.dumps(pages, default=str)
+        all_pages = [_format_page(p) for p in data["pages"]]
+        page = all_pages[offset : offset + limit]
+        has_more = offset + limit < len(all_pages)
+
+        return json.dumps(
+            {
+                "data": page,
+                "pagination": {"count": len(page), "offset": offset, "limit": limit, "has_more": has_more},
+            },
+            default=str,
+        )
 
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)

src/nc_mcp_server/tools/files.py

@@ -61,22 +61,35 @@ def _build_search_xml(user: str, query: str, path: str, limit: int, offset: int,
 def _register_read_tools(mcp: FastMCP) -> None:
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
-    async def list_directory(path: str = "/") -> str:
+    async def list_directory(path: str = "/", limit: int = 50, offset: int = 0) -> str:
         """List files and folders in a Nextcloud directory.
 
         Args:
             path: Directory path relative to user's root (default: "/" for root).
                   Example: "Documents", "Photos/Vacation"
+            limit: Maximum number of entries to return (1-500, default 50).
+            offset: Number of entries to skip for pagination (default 0).
 
         Returns:
-            JSON list of entries, each with: path, is_directory, size, last_modified, content_type.
+            JSON with "data" (list of entries with path, is_directory, size, etc.)
+            and "pagination" (count, offset, limit, has_more).
         """
+        limit = max(1, min(500, limit))
+        offset = max(0, offset)
         client = get_client()
         entries = await client.dav_propfind(path, depth=1)
-        # First entry is the directory itself — skip it
         if entries and entries[0]["path"].rstrip("/") == path.strip("/"):
             entries = entries[1:]
-        return json.dumps(entries, default=str)
+        page = entries[offset : offset + limit]
+        has_more = offset + limit < len(entries)
+
+        return json.dumps(
+            {
+                "data": page,
+                "pagination": {"count": len(page), "offset": offset, "limit": limit, "has_more": has_more},
+            },
+            default=str,
+        )
 
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)

src/nc_mcp_server/tools/notifications.py

@@ -14,21 +14,33 @@ def register(mcp: FastMCP) -> None:
 
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
-    async def list_notifications() -> str:
-        """List all notifications for the current Nextcloud user.
+    async def list_notifications(limit: int = 50, offset: int = 0) -> str:
+        """List notifications for the current Nextcloud user.
 
-        Returns notifications sorted by newest first. Each notification
-        includes: notification_id, app, datetime, subject, message, link,
-        and actions.
+        Returns notifications sorted by newest first.
+
+        Args:
+            limit: Maximum number of notifications to return (1-200, default 50).
+            offset: Number of notifications to skip for pagination (default 0).
 
         Returns:
-            JSON list of notification objects.
+            JSON with "data" (list of notification objects) and "pagination"
+            (count, offset, limit, has_more).
         """
+        limit = max(1, min(200, limit))
+        offset = max(0, offset)
         client = get_client()
-        data = await client.ocs_get(
-            "apps/notifications/api/v2/notifications",
+        data = await client.ocs_get("apps/notifications/api/v2/notifications")
+        page = data[offset : offset + limit]
+        has_more = offset + limit < len(data)
+
+        return json.dumps(
+            {
+                "data": page,
+                "pagination": {"count": len(page), "offset": offset, "limit": limit, "has_more": has_more},
+            },
+            default=str,
         )
-        return json.dumps(data, default=str)
 
     @mcp.tool(annotations=DESTRUCTIVE)
     @require_permission(PermissionLevel.DESTRUCTIVE)

src/nc_mcp_server/tools/shares.py

@@ -45,6 +45,8 @@ async def list_shares(
         path: str = "",
         reshares: bool = False,
         subfiles: bool = False,
+        limit: int = 50,
+        offset: int = 0,
     ) -> str:
         """List file/folder shares from Nextcloud.
 
@@ -55,11 +57,15 @@ async def list_shares(
             path: Optional file/folder path to filter shares (e.g. "/Documents/report.pdf").
             reshares: If true, include shares by other users on the same files.
             subfiles: If true and path is a folder, list shares of files inside it (not the folder itself).
+            limit: Maximum number of shares to return (1-200, default 50).
+            offset: Number of shares to skip for pagination (default 0).
 
         Returns:
-            JSON list of share objects with: id, share_type, path, permissions, share_with, etc.
+            JSON with "data" (list of share objects) and "pagination" (count, offset, limit, has_more).
             share_type values: 0=user, 1=group, 3=public link, 4=email, 6=federated, 10=talk room.
         """
+        limit = max(1, min(200, limit))
+        offset = max(0, offset)
         client = get_client()
         params: dict[str, str] = {}
         if path:
@@ -69,8 +75,17 @@ async def list_shares(
         if subfiles:
             params["subfiles"] = "true"
         data = await client.ocs_get(SHARES_API, params=params)
-        shares = [_format_share(s) for s in data]
-        return json.dumps(shares, default=str)
+        all_shares = [_format_share(s) for s in data]
+        page = all_shares[offset : offset + limit]
+        has_more = offset + limit < len(all_shares)
+
+        return json.dumps(
+            {
+                "data": page,
+                "pagination": {"count": len(page), "offset": offset, "limit": limit, "has_more": has_more},
+            },
+            default=str,
+        )
 
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)

src/nc_mcp_server/tools/talk.py

@@ -128,27 +128,43 @@ def _register_read_tools(mcp: FastMCP) -> None:
 
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
-    async def list_conversations(include_notifications_disabled: bool = False) -> str:
-        """List all Talk conversations the current user is part of.
+    async def list_conversations(
+        include_notifications_disabled: bool = False,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> str:
+        """List Talk conversations the current user is part of.
 
         Returns conversations sorted by last activity (newest first).
-        Each conversation includes: token (unique ID for API calls), type,
-        name, unread counts, and permissions.
 
         Args:
             include_notifications_disabled: If true, also return conversations where
                 notifications are disabled (default: false).
+            limit: Maximum number of conversations to return (1-200, default 50).
+            offset: Number of conversations to skip for pagination (default 0).
 
         Returns:
-            JSON list of conversation objects.
+            JSON with "data" (list of conversation objects) and "pagination"
+            (count, offset, limit, has_more).
         """
+        limit = max(1, min(200, limit))
+        offset = max(0, offset)
         client = get_client()
         params: dict[str, str] = {}
         if not include_notifications_disabled:
             params["noStatusUpdate"] = "0"
         data = await client.ocs_get("apps/spreed/api/v4/room", params=params)
-        conversations = [_format_conversation(room) for room in data]
-        return json.dumps(conversations, default=str)
+        all_convs = [_format_conversation(room) for room in data]
+        page = all_convs[offset : offset + limit]
+        has_more = offset + limit < len(all_convs)
+
+        return json.dumps(
+            {
+                "data": page,
+                "pagination": {"count": len(page), "offset": offset, "limit": limit, "has_more": has_more},
+            },
+            default=str,
+        )
 
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)

src/nc_mcp_server/tools/trashbin.py

@@ -80,22 +80,35 @@ def _parse_trash_xml(xml_text: str, user: str) -> list[dict[str, Any]]:
 def _register_read_tools(mcp: FastMCP) -> None:
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
-    async def list_trash() -> str:
-        """List all items in the Nextcloud trash bin.
+    async def list_trash(limit: int = 50, offset: int = 0) -> str:
+        """List items in the Nextcloud trash bin.
 
         Returns files and folders that were deleted and can be restored.
-        Each item includes its original filename, original path, deletion
-        time, and a trash_path identifier needed for restore/delete operations.
+
+        Args:
+            limit: Maximum number of items to return (1-200, default 50).
+            offset: Number of items to skip for pagination (default 0).
 
         Returns:
-            JSON list of trashed items, each with: trash_path, original_name,
-            original_location, deletion_time (unix), is_directory, size, file_id.
-            Use trash_path with restore_trash_item or delete operations.
+            JSON with "data" (list of trashed items with trash_path, original_name,
+            original_location, deletion_time, is_directory, size, file_id) and
+            "pagination" (count, offset, limit, has_more).
         """
+        limit = max(1, min(200, limit))
+        offset = max(0, offset)
         client = get_client()
         xml_text = await client.trashbin_propfind()
         entries = _parse_trash_xml(xml_text, get_config().user)
-        return json.dumps(entries, default=str)
+        page = entries[offset : offset + limit]
+        has_more = offset + limit < len(entries)
+
+        return json.dumps(
+            {
+                "data": page,
+                "pagination": {"count": len(page), "offset": offset, "limit": limit, "has_more": has_more},
+            },
+            default=str,
+        )
 
 
 def _register_write_tools(mcp: FastMCP) -> None: