Add pagination to list_users, list_apps, list_tags, list_versions, get_participants (#36)

Author: bigcat88

pyproject.toml

@@ -112,6 +112,7 @@ reportPrivateUsage = false
 executionEnvironments = [
   { root = "tests/test_client_retry.py", reportAttributeAccessIssue = false, reportUnknownMemberType = false },
   { root = "src/nc_mcp_server/tools/calendar.py", reportUnknownMemberType = false, reportUnknownVariableType = false, reportUnknownArgumentType = false },
+  { root = "src/nc_mcp_server/tools/users.py", reportUnknownVariableType = false, reportUnknownArgumentType = false },
 ]
 
 [tool.pytest]

src/nc_mcp_server/tools/app_management.py

@@ -25,27 +25,39 @@ def _format_app(a: dict[str, Any]) -> dict[str, Any]:
 def _register_read_tools(mcp: FastMCP) -> None:
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
-    async def list_apps(app_filter: str = "enabled") -> str:
+    async def list_apps(app_filter: str = "enabled", limit: int = 50, offset: int = 0) -> str:
         """List installed Nextcloud apps. Requires admin privileges.
 
-        Returns the list of app IDs matching the filter criteria.
-
         Args:
             app_filter: Filter apps by status. Options:
                         "enabled" (default) — only enabled apps
                         "disabled" — only disabled apps
                         "all" — all installed apps
+            limit: Maximum number of apps to return (1-500, default 50).
+            offset: Number of apps to skip for pagination (default 0).
 
         Returns:
-            JSON list of app ID strings.
+            JSON with "data" (list of app ID strings) and "pagination"
+            (count, offset, limit, has_more).
         """
         valid = {"enabled", "disabled", "all"}
         if app_filter not in valid:
             raise ValueError(f"Invalid filter '{app_filter}'. Must be one of: {', '.join(sorted(valid))}")
+        limit = max(1, min(500, limit))
+        offset = max(0, offset)
         client = get_client()
         params = {} if app_filter == "all" else {"filter": app_filter}
         data = await client.ocs_get(APPS_API, params=params)
-        return json.dumps(data["apps"])
+        all_apps = data["apps"]
+        page = all_apps[offset : offset + limit]
+        has_more = offset + limit < len(all_apps)
+        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/system_tags.py

@@ -59,16 +59,22 @@ def _parse_tags_xml(xml_text: str) -> list[dict[str, Any]]:
 def _register_read_tools(mcp: FastMCP) -> None:
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
-    async def list_tags() -> str:
-        """List all system tags available in this Nextcloud instance.
+    async def list_tags(limit: int = 50, offset: int = 0) -> str:
+        """List system tags available in this Nextcloud instance.
 
         System tags are shared labels that can be assigned to files for
-        organization and filtering. Tags have visibility and assignability
-        settings controlled by admins.
+        organization and filtering.
+
+        Args:
+            limit: Maximum number of tags to return (1-500, default 50).
+            offset: Number of tags to skip for pagination (default 0).
 
         Returns:
-            JSON list of tags, each with: id, name, user_visible, user_assignable.
+            JSON with "data" (list of tags with id, name, user_visible,
+            user_assignable) and "pagination" (count, offset, limit, has_more).
         """
+        limit = max(1, min(500, limit))
+        offset = max(0, offset)
         client = get_client()
         response = await client.dav_request(
             "PROPFIND",
@@ -77,8 +83,16 @@ async def list_tags() -> str:
             headers={"Content-Type": "application/xml; charset=utf-8"},
             context="List system tags",
         )
-        tags = _parse_tags_xml(response.text or "")
-        return json.dumps(tags)
+        all_tags = _parse_tags_xml(response.text or "")
+        page = all_tags[offset : offset + limit]
+        has_more = offset + limit < len(all_tags)
+        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

@@ -237,21 +237,33 @@ async def get_messages(
 
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
-    async def get_participants(token: str) -> str:
+    async def get_participants(token: str, limit: int = 50, offset: int = 0) -> str:
         """List participants in a Talk conversation.
 
         Args:
             token: The conversation token. Use list_conversations to find tokens.
+            limit: Maximum number of participants to return (1-200, default 50).
+            offset: Number of participants to skip for pagination (default 0).
 
         Returns:
-            JSON list of participant objects, each with: attendee_id,
-            actor_id, display_name, participant_type (owner/moderator/user/guest),
-            in_call status.
+            JSON with "data" (list of participant objects with attendee_id,
+            actor_id, display_name, participant_type, in_call) 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"apps/spreed/api/v4/room/{token}/participants")
-        participants = [_format_participant(p) for p in data]
-        return json.dumps(participants, default=str)
+        all_participants = [_format_participant(p) for p in data]
+        page = all_participants[offset : offset + limit]
+        has_more = offset + limit < len(all_participants)
+        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/users.py

@@ -25,20 +25,32 @@ async def get_current_user() -> str:
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
     async def list_users(search: str = "", limit: int = 25, offset: int = 0) -> str:
-        """List Nextcloud users.
+        """List Nextcloud users. Uses server-side pagination.
 
         Args:
             search: Optional search string to filter users by name/email.
-            limit: Maximum number of users to return (default 25).
-            offset: Offset for pagination.
+            limit: Maximum number of users to return (1-200, default 25).
+            offset: Number of users to skip for pagination (default 0).
 
         Returns:
-            JSON list of user IDs matching the search.
+            JSON with "data" (list of user ID strings) and "pagination"
+            (count, offset, limit, has_more).
         """
+        limit = max(1, min(200, limit))
+        offset = max(0, offset)
         client = get_client()
         params = {"search": search, "limit": str(limit), "offset": str(offset)}
         data = await client.ocs_get("cloud/users", params=params)
-        return json.dumps(data, default=str)
+        raw = data["users"] if isinstance(data, dict) and "users" in data else data
+        users: list[str] = list(raw) if not isinstance(raw, list) else raw
+        has_more = len(users) == limit
+        return json.dumps(
+            {
+                "data": users,
+                "pagination": {"count": len(users), "offset": offset, "limit": limit, "has_more": has_more},
+            },
+            default=str,
+        )
 
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)

src/nc_mcp_server/tools/versions.py

@@ -60,24 +60,36 @@ def _parse_versions_xml(xml_text: str, user: str, file_id: int) -> list[dict[str
 def _register_read_tools(mcp: FastMCP) -> None:
     @mcp.tool(annotations=READONLY)
     @require_permission(PermissionLevel.READ)
-    async def list_versions(file_id: int) -> str:
-        """List all versions of a file by its file ID.
+    async def list_versions(file_id: int, limit: int = 50, offset: int = 0) -> str:
+        """List versions of a file by its file ID.
 
-        Returns the version history including the current version.
-        Use the file_id from list_directory or search_files results.
+        Returns the version history. Use the file_id from list_directory
+        or search_files results.
 
         Args:
             file_id: The numeric Nextcloud file ID.
+            limit: Maximum number of versions to return (1-200, default 50).
+            offset: Number of versions to skip for pagination (default 0).
 
         Returns:
-            JSON list of versions, each with: version_id (unix timestamp),
-            last_modified, size, content_type, author, and optionally label.
-            Use version_id with restore_version to revert the file.
+            JSON with "data" (list of versions with version_id, last_modified,
+            size, content_type, author, label) and "pagination"
+            (count, offset, limit, has_more).
         """
+        limit = max(1, min(200, limit))
+        offset = max(0, offset)
         client = get_client()
         xml_text = await client.versions_propfind(file_id)
-        entries = _parse_versions_xml(xml_text, get_config().user, file_id)
-        return json.dumps(entries, default=str)
+        all_versions = _parse_versions_xml(xml_text, get_config().user, file_id)
+        page = all_versions[offset : offset + limit]
+        has_more = offset + limit < len(all_versions)
+        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:

tests/integration/test_app_management.py

@@ -15,28 +15,28 @@
 class TestListApps:
     @pytest.mark.asyncio
     async def test_list_enabled_returns_list(self, nc_mcp: McpTestHelper) -> None:
-        result = await nc_mcp.call("list_apps")
-        apps: list[str] = json.loads(result)
+        result = await nc_mcp.call("list_apps", limit=500)
+        apps: list[str] = json.loads(result)["data"]
         assert isinstance(apps, list)
         assert len(apps) > 0
 
     @pytest.mark.asyncio
     async def test_list_enabled_contains_core_apps(self, nc_mcp: McpTestHelper) -> None:
-        result = await nc_mcp.call("list_apps", app_filter="enabled")
-        apps = json.loads(result)
+        result = await nc_mcp.call("list_apps", app_filter="enabled", limit=500)
+        apps = json.loads(result)["data"]
         assert "files" in apps
         assert "dav" in apps
 
     @pytest.mark.asyncio
     async def test_list_all(self, nc_mcp: McpTestHelper) -> None:
-        result = await nc_mcp.call("list_apps", app_filter="all")
-        apps = json.loads(result)
+        result = await nc_mcp.call("list_apps", app_filter="all", limit=500)
+        apps = json.loads(result)["data"]
         assert len(apps) > 0
 
     @pytest.mark.asyncio
     async def test_list_disabled(self, nc_mcp: McpTestHelper) -> None:
-        result = await nc_mcp.call("list_apps", app_filter="disabled")
-        apps = json.loads(result)
+        result = await nc_mcp.call("list_apps", app_filter="disabled", limit=500)
+        apps = json.loads(result)["data"]
         assert isinstance(apps, list)
 
     @pytest.mark.asyncio
@@ -73,10 +73,10 @@ class TestEnableDisableApp:
     async def test_disable_and_enable(self, nc_mcp: McpTestHelper) -> None:
         await nc_mcp.call("disable_app", app_id=SAFE_APP)
         try:
-            disabled = json.loads(await nc_mcp.call("list_apps", app_filter="disabled"))
+            disabled = json.loads(await nc_mcp.call("list_apps", app_filter="disabled", limit=500))["data"]
             assert SAFE_APP in disabled
             await nc_mcp.call("enable_app", app_id=SAFE_APP)
-            enabled = json.loads(await nc_mcp.call("list_apps", app_filter="enabled"))
+            enabled = json.loads(await nc_mcp.call("list_apps", app_filter="enabled", limit=500))["data"]
             assert SAFE_APP in enabled
         finally:
             await nc_mcp.call("enable_app", app_id=SAFE_APP)
@@ -98,8 +98,8 @@ async def test_disable_returns_confirmation(self, nc_mcp: McpTestHelper) -> None
 class TestAppManagementPermissions:
     @pytest.mark.asyncio
     async def test_read_only_allows_list(self, nc_mcp_read_only: McpTestHelper) -> None:
-        result = await nc_mcp_read_only.call("list_apps")
-        assert isinstance(json.loads(result), list)
+        result = await nc_mcp_read_only.call("list_apps", limit=500)
+        assert isinstance(json.loads(result)["data"], list)
 
     @pytest.mark.asyncio
     async def test_read_only_blocks_enable(self, nc_mcp_read_only: McpTestHelper) -> None:

tests/integration/test_system_tags.py

@@ -29,16 +29,16 @@ async def _get_test_file_id(nc_mcp: McpTestHelper) -> int:
 class TestListTags:
     @pytest.mark.asyncio
     async def test_returns_list(self, nc_mcp: McpTestHelper) -> None:
-        result = await nc_mcp.call("list_tags")
-        data = json.loads(result)
+        result = await nc_mcp.call("list_tags", limit=200)
+        data = json.loads(result)["data"]
         assert isinstance(data, list)
 
     @pytest.mark.asyncio
     async def test_created_tag_appears_in_list(self, nc_mcp: McpTestHelper) -> None:
         tag_id = await _create_tag(nc_mcp, "mcp-test-visible")
         try:
-            result = await nc_mcp.call("list_tags")
-            tags = json.loads(result)
+            result = await nc_mcp.call("list_tags", limit=200)
+            tags = json.loads(result)["data"]
             ids = [t["id"] for t in tags]
             assert tag_id in ids
             tag = next(t for t in tags if t["id"] == tag_id)
@@ -53,8 +53,8 @@ async def test_created_tag_appears_in_list(self, nc_mcp: McpTestHelper) -> None:
     async def test_tag_has_required_fields(self, nc_mcp: McpTestHelper) -> None:
         tag_id = await _create_tag(nc_mcp, "mcp-test-fields")
         try:
-            result = await nc_mcp.call("list_tags")
-            tags = json.loads(result)
+            result = await nc_mcp.call("list_tags", limit=200)
+            tags = json.loads(result)["data"]
             tag = next(t for t in tags if t["id"] == tag_id)
             for field in ["id", "name", "user_visible", "user_assignable"]:
                 assert field in tag, f"Missing field: {field}"
@@ -100,7 +100,7 @@ async def test_create_non_assignable_tag(self, nc_mcp: McpTestHelper) -> None:
         result = await nc_mcp.call("create_tag", name="mcp-test-noa", user_assignable=False)
         tag_id = int(json.loads(result)["id"])
         try:
-            tags = json.loads(await nc_mcp.call("list_tags"))
+            tags = json.loads(await nc_mcp.call("list_tags", limit=200))["data"]
             tag = next(t for t in tags if t["id"] == tag_id)
             assert tag["user_assignable"] is False
         finally:
@@ -218,7 +218,7 @@ async def test_delete_tag(self, nc_mcp: McpTestHelper) -> None:
         tag_id = await _create_tag(nc_mcp, "mcp-test-delete")
         result = await nc_mcp.call("delete_tag", tag_id=tag_id)
         assert "deleted" in result.lower()
-        tags = json.loads(await nc_mcp.call("list_tags"))
+        tags = json.loads(await nc_mcp.call("list_tags", limit=200))["data"]
         assert not any(t["id"] == tag_id for t in tags)
 
     @pytest.mark.asyncio
@@ -239,8 +239,8 @@ async def test_delete_removes_from_files(self, nc_mcp: McpTestHelper) -> None:
 class TestSystemTagPermissions:
     @pytest.mark.asyncio
     async def test_read_only_allows_list(self, nc_mcp_read_only: McpTestHelper) -> None:
-        result = await nc_mcp_read_only.call("list_tags")
-        assert isinstance(json.loads(result), list)
+        result = await nc_mcp_read_only.call("list_tags", limit=200)
+        assert isinstance(json.loads(result)["data"], list)
 
     @pytest.mark.asyncio
     async def test_read_only_blocks_create(self, nc_mcp_read_only: McpTestHelper) -> None:

tests/integration/test_talk.py

@@ -268,8 +268,8 @@ class TestGetParticipants:
     async def test_returns_participants_list(self, nc_mcp: McpTestHelper) -> None:
         room = await _create_room(nc_mcp, "test-participants")
         try:
-            result = await nc_mcp.call("get_participants", token=str(room["token"]))
-            data: list[Any] = json.loads(result)
+            result = await nc_mcp.call("get_participants", token=str(room["token"]), limit=200)
+            data: list[Any] = json.loads(result)["data"]
             assert isinstance(data, list)
             assert len(data) >= 1
         finally:
@@ -279,8 +279,8 @@ async def test_returns_participants_list(self, nc_mcp: McpTestHelper) -> None:
     async def test_creator_is_owner(self, nc_mcp: McpTestHelper) -> None:
         room = await _create_room(nc_mcp, "test-owner")
         try:
-            result = await nc_mcp.call("get_participants", token=str(room["token"]))
-            data = json.loads(result)
+            result = await nc_mcp.call("get_participants", token=str(room["token"]), limit=200)
+            data = json.loads(result)["data"]
             admin = next(p for p in data if p["actor_id"] == "admin")
             assert admin["participant_type"] == "owner"
         finally:
@@ -290,8 +290,8 @@ async def test_creator_is_owner(self, nc_mcp: McpTestHelper) -> None:
     async def test_participant_has_required_fields(self, nc_mcp: McpTestHelper) -> None:
         room = await _create_room(nc_mcp, "test-part-fields")
         try:
-            result = await nc_mcp.call("get_participants", token=str(room["token"]))
-            data = json.loads(result)
+            result = await nc_mcp.call("get_participants", token=str(room["token"]), limit=200)
+            data = json.loads(result)["data"]
             p = data[0]
             required = ["attendee_id", "actor_type", "actor_id", "display_name", "participant_type", "in_call"]
             for field in required:
@@ -437,8 +437,8 @@ async def test_creator_is_participant(self, nc_mcp: McpTestHelper) -> None:
         result = await nc_mcp.call("create_conversation", room_type=2, name="test-creator-part")
         data = json.loads(result)
         try:
-            participants = json.loads(await nc_mcp.call("get_participants", token=str(data["token"])))
-            actor_ids = [p["actor_id"] for p in participants]
+            raw = await nc_mcp.call("get_participants", token=str(data["token"]), limit=200)
+            actor_ids = [p["actor_id"] for p in json.loads(raw)["data"]]
             assert "admin" in actor_ids
         finally:
             await _delete_room(nc_mcp, str(data["token"]))