feat: add pagination to mcp service

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.2.11"
+version = "2.2.12"
 description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

src/uipath/platform/agenthub/_mcp_service.py

@@ -1,14 +1,19 @@
-from typing import List
+from typing import Any, Dict
 
 from ..._config import Config
 from ..._execution_context import ExecutionContext
 from ..._folder_context import FolderContext
 from ..._utils import Endpoint, RequestSpec, header_folder
 from ...tracing import traced
 from ..common._base_service import BaseService
+from ..common.paging import PagedResult
 from ..orchestrator._folder_service import FolderService
 from ..orchestrator.mcp import McpServer
 
+# Pagination limits
+MAX_PAGE_SIZE = 1000  # Maximum items per page (top parameter)
+MAX_SKIP_OFFSET = 10000  # Maximum skip offset for offset-based pagination
+
 
 class McpService(FolderContext, BaseService):
     """Service for managing MCP (Model Context Protocol) servers in UiPath.
@@ -31,82 +36,175 @@ def list(
         self,
         *,
         folder_path: str | None = None,
-    ) -> List[McpServer]:
-        """List all MCP servers.
+        skip: int = 0,
+        top: int = 100,
+    ) -> PagedResult[McpServer]:
+        """List MCP servers with offset-based pagination.
+
+        Returns a single page of results with pagination metadata.
 
         Args:
-            folder_path (Optional[str]): The path of the folder to list servers from.
+            folder_path: The path of the folder to list servers from
+            skip: Number of servers to skip (default 0, max 10000)
+            top: Maximum number of servers to return (default 100, max 1000)
 
         Returns:
-            List[McpServer]: A list of MCP servers with their configuration.
+            PagedResult[McpServer]: Page containing servers and pagination metadata
 
-        Examples:
-            ```python
-            from uipath import UiPath
+        Raises:
+            ValueError: If skip < 0, skip > 10000, top < 1, or top > 1000
 
-            client = UiPath()
-
-            servers = client.mcp.list(folder_path="MyFolder")
-            for server in servers:
-                print(f"{server.name} - {server.slug}")
-            ```
+        Examples:
+            >>> # Get first page
+            >>> result = sdk.mcp.list(top=100)
+            >>> for server in result.items:
+            ...     print(f"{server.name} - {server.slug}")
+            >>>
+            >>> # Check pagination metadata
+            >>> if result.has_more:
+            ...     print(f"More results available. Current: skip={result.skip}, top={result.top}")
+            >>>
+            >>> # Manual pagination to get all servers
+            >>> skip = 0
+            >>> top = 100
+            >>> all_servers = []
+            >>> while True:
+            ...     result = sdk.mcp.list(skip=skip, top=top)
+            ...     all_servers.extend(result.items)
+            ...     if not result.has_more:
+            ...         break
+            ...     skip += top
+            >>>
+            >>> # Helper function for complete iteration
+            >>> def iter_all_servers(sdk, top=100, **filters):
+            ...     skip = 0
+            ...     while True:
+            ...         result = sdk.mcp.list(skip=skip, top=top, **filters)
+            ...         yield from result.items
+            ...         if not result.has_more:
+            ...             break
+            ...         skip += top
+            >>>
+            >>> # Usage
+            >>> for server in iter_all_servers(sdk, folder_path="MyFolder"):
+            ...     process_server(server)
         """
+        if skip < 0:
+            raise ValueError("skip must be >= 0")
+        if skip > MAX_SKIP_OFFSET:
+            raise ValueError(
+                f"skip must be <= {MAX_SKIP_OFFSET} (requested: {skip}). "
+                f"Use pagination with skip and top parameters to retrieve larger datasets."
+            )
+        if top < 1:
+            raise ValueError("top must be >= 1")
+        if top > MAX_PAGE_SIZE:
+            raise ValueError(
+                f"top must be <= {MAX_PAGE_SIZE} (requested: {top}). "
+                f"Use pagination with skip and top parameters to retrieve larger datasets."
+            )
+
         spec = self._list_spec(
             folder_path=folder_path,
+            skip=skip,
+            top=top,
         )
 
         response = self.request(
             spec.method,
             url=spec.endpoint,
             params=spec.params,
             headers=spec.headers,
-        )
+        ).json()
 
-        return [McpServer.model_validate(server) for server in response.json()]
+        servers = [McpServer.model_validate(server) for server in response]
+
+        return PagedResult(
+            items=servers,
+            has_more=len(servers) == top,
+            skip=skip,
+            top=top,
+        )
 
     @traced(name="mcp_list", run_type="uipath")
     async def list_async(
         self,
         *,
         folder_path: str | None = None,
-    ) -> List[McpServer]:
-        """Asynchronously list all MCP servers.
+        skip: int = 0,
+        top: int = 100,
+    ) -> PagedResult[McpServer]:
+        """Async version of list() with offset-based pagination.
+
+        Returns a single page of results with pagination metadata.
 
         Args:
-            folder_path (Optional[str]): The path of the folder to list servers from.
+            folder_path: The path of the folder to list servers from
+            skip: Number of servers to skip (default 0, max 10000)
+            top: Maximum number of servers to return (default 100, max 1000)
 
         Returns:
-            List[McpServer]: A list of MCP servers with their configuration.
+            PagedResult[McpServer]: Page containing servers and pagination metadata
 
-        Examples:
-            ```python
-            import asyncio
+        Raises:
+            ValueError: If skip < 0, skip > 10000, top < 1, or top > 1000
 
-            from uipath import UiPath
-
-            sdk = UiPath()
-
-            async def main():
-                servers = await sdk.mcp.list_async(folder_path="MyFolder")
-                for server in servers:
-                    print(f"{server.name} - {server.slug}")
-
-            asyncio.run(main())
-            ```
+        Examples:
+            >>> # Get first page
+            >>> result = await sdk.mcp.list_async(top=100)
+            >>> for server in result.items:
+            ...     print(f"{server.name} - {server.slug}")
+            >>>
+            >>> # Manual pagination
+            >>> skip = 0
+            >>> top = 100
+            >>> all_servers = []
+            >>> while True:
+            ...     result = await sdk.mcp.list_async(skip=skip, top=top)
+            ...     all_servers.extend(result.items)
+            ...     if not result.has_more:
+            ...         break
+            ...     skip += top
         """
+        if skip < 0:
+            raise ValueError("skip must be >= 0")
+        if skip > MAX_SKIP_OFFSET:
+            raise ValueError(
+                f"skip must be <= {MAX_SKIP_OFFSET} (requested: {skip}). "
+                f"Use pagination with skip and top parameters to retrieve larger datasets."
+            )
+        if top < 1:
+            raise ValueError("top must be >= 1")
+        if top > MAX_PAGE_SIZE:
+            raise ValueError(
+                f"top must be <= {MAX_PAGE_SIZE} (requested: {top}). "
+                f"Use pagination with skip and top parameters to retrieve larger datasets."
+            )
+
         spec = self._list_spec(
             folder_path=folder_path,
+            skip=skip,
+            top=top,
         )
 
-        response = await self.request_async(
-            spec.method,
-            url=spec.endpoint,
-            params=spec.params,
-            headers=spec.headers,
+        response = (
+            await self.request_async(
+                spec.method,
+                url=spec.endpoint,
+                params=spec.params,
+                headers=spec.headers,
+            )
+        ).json()
+
+        servers = [McpServer.model_validate(server) for server in response]
+
+        return PagedResult(
+            items=servers,
+            has_more=len(servers) == top,
+            skip=skip,
+            top=top,
         )
 
-        return [McpServer.model_validate(server) for server in response.json()]
-
     @traced(name="mcp_retrieve", run_type="uipath")
     def retrieve(
         self,
@@ -200,11 +298,17 @@ def _list_spec(
         self,
         *,
         folder_path: str | None,
+        skip: int,
+        top: int,
     ) -> RequestSpec:
         folder_key = self._folders_service.retrieve_folder_key(folder_path)
+
+        params: Dict[str, Any] = {"$skip": skip, "$top": top}
+
         return RequestSpec(
             method="GET",
             endpoint=Endpoint("/agenthub_/api/servers"),
+            params=params,
             headers={
                 **header_folder(folder_key, None),
             },

tests/sdk/services/test_mcp_service.py

@@ -7,6 +7,7 @@
 from uipath._execution_context import ExecutionContext
 from uipath._utils.constants import HEADER_FOLDER_KEY, HEADER_USER_AGENT
 from uipath.platform.agenthub._mcp_service import McpService
+from uipath.platform.common.paging import PagedResult
 from uipath.platform.orchestrator._folder_service import FolderService
 from uipath.platform.orchestrator.mcp import McpServer
 
@@ -87,19 +88,24 @@ def test_list_with_folder_path(
                 json=mock_servers,
             )
 
-            servers = service.list(folder_path="test-folder-path")
+            result = service.list(folder_path="test-folder-path")
 
-            assert len(servers) == 1
-            assert isinstance(servers[0], McpServer)
-            assert servers[0].name == "Test MCP Server"
+            assert isinstance(result, PagedResult)
+            assert len(result.items) == 1
+            assert isinstance(result.items[0], McpServer)
+            assert result.items[0].name == "Test MCP Server"
+            assert result.has_more is False
+            assert result.skip == 0
+            assert result.top == 100
 
             requests = httpx_mock.get_requests()
             assert len(requests) == 2
 
             servers_request = requests[1]
             assert servers_request.method == "GET"
             assert (
-                servers_request.url == f"{base_url}{org}{tenant}/agenthub_/api/servers"
+                servers_request.url
+                == f"{base_url}{org}{tenant}/agenthub_/api/servers?%24skip=0&%24top=100"
             )
             assert HEADER_FOLDER_KEY in servers_request.headers
             assert servers_request.headers[HEADER_FOLDER_KEY] == "resolved-folder-key"
@@ -181,19 +187,24 @@ async def test_list_async(
                 json=mock_servers,
             )
 
-            servers = await service.list_async(folder_path="test-folder-path")
+            result = await service.list_async(folder_path="test-folder-path")
 
-            assert len(servers) == 1
-            assert isinstance(servers[0], McpServer)
-            assert servers[0].name == "Async Test Server"
+            assert isinstance(result, PagedResult)
+            assert len(result.items) == 1
+            assert isinstance(result.items[0], McpServer)
+            assert result.items[0].name == "Async Test Server"
+            assert result.has_more is False
+            assert result.skip == 0
+            assert result.top == 100
 
             requests = httpx_mock.get_requests()
             assert len(requests) == 2
 
             servers_request = requests[1]
             assert servers_request.method == "GET"
             assert (
-                servers_request.url == f"{base_url}{org}{tenant}/agenthub_/api/servers"
+                servers_request.url
+                == f"{base_url}{org}{tenant}/agenthub_/api/servers?%24skip=0&%24top=100"
             )
             assert HEADER_FOLDER_KEY in servers_request.headers
             assert servers_request.headers[HEADER_FOLDER_KEY] == "test-folder-key"
@@ -382,7 +393,10 @@ def test_list_passes_all_kwargs(self, service: McpService) -> None:
                 with patch.object(
                     service, "request", return_value=mock_response
                 ) as mock_request:
-                    service.list(folder_path="test-folder-path")
+                    result = service.list(folder_path="test-folder-path")
+
+                    assert isinstance(result, PagedResult)
+                    assert len(result.items) == 1
 
                     mock_request.assert_called_once()
                     call_kwargs = mock_request.call_args
@@ -398,6 +412,8 @@ def test_list_passes_all_kwargs(self, service: McpService) -> None:
                         call_kwargs.kwargs["headers"][HEADER_FOLDER_KEY]
                         == "test-folder-key"
                     )
+                    assert call_kwargs.kwargs["params"]["$skip"] == 0
+                    assert call_kwargs.kwargs["params"]["$top"] == 100
 
         @pytest.mark.anyio
         async def test_list_async_passes_all_kwargs(self, service: McpService) -> None:
@@ -429,7 +445,10 @@ async def test_list_async_passes_all_kwargs(self, service: McpService) -> None:
                 with patch.object(
                     service, "request_async", return_value=mock_response
                 ) as mock_request:
-                    await service.list_async(folder_path="test-folder-path")
+                    result = await service.list_async(folder_path="test-folder-path")
+
+                    assert isinstance(result, PagedResult)
+                    assert len(result.items) == 1
 
                     mock_request.assert_called_once()
                     call_kwargs = mock_request.call_args
@@ -445,6 +464,8 @@ async def test_list_async_passes_all_kwargs(self, service: McpService) -> None:
                         call_kwargs.kwargs["headers"][HEADER_FOLDER_KEY]
                         == "test-folder-key"
                     )
+                    assert call_kwargs.kwargs["params"]["$skip"] == 0
+                    assert call_kwargs.kwargs["params"]["$top"] == 100
 
         def test_retrieve_passes_all_kwargs(self, service: McpService) -> None:
             """Test that retrieve passes all kwargs to request."""