feat: add pagination for secrets (#179)

Author: mathislucka

src/deepset_mcp/api/secrets/models.py

@@ -10,11 +10,3 @@ class Secret(BaseModel):
 
     name: str
     secret_id: str
-
-
-class SecretList(BaseModel):
-    """Model representing a list of secrets with pagination."""
-
-    data: list[Secret]
-    has_more: bool
-    total: int

src/deepset_mcp/api/secrets/protocols.py

@@ -4,8 +4,8 @@
 
 from typing import Protocol
 
-from deepset_mcp.api.secrets.models import Secret, SecretList
-from deepset_mcp.api.shared_models import NoContentResponse
+from deepset_mcp.api.secrets.models import Secret
+from deepset_mcp.api.shared_models import NoContentResponse, PaginatedResponse
 
 
 class SecretResourceProtocol(Protocol):
@@ -16,7 +16,8 @@ async def list(
         limit: int = 10,
         field: str = "created_at",
         order: str = "DESC",
-    ) -> SecretList:
+        after: str | None = None,
+    ) -> PaginatedResponse[Secret]:
         """List secrets with pagination."""
         ...
 

src/deepset_mcp/api/secrets/resource.py

@@ -6,9 +6,9 @@
 
 from deepset_mcp.api.exceptions import ResourceNotFoundError
 from deepset_mcp.api.protocols import AsyncClientProtocol
-from deepset_mcp.api.secrets.models import Secret, SecretList
+from deepset_mcp.api.secrets.models import Secret
 from deepset_mcp.api.secrets.protocols import SecretResourceProtocol
-from deepset_mcp.api.shared_models import NoContentResponse
+from deepset_mcp.api.shared_models import NoContentResponse, PaginatedResponse
 from deepset_mcp.api.transport import raise_for_status
 
 
@@ -27,34 +27,59 @@ async def list(
         limit: int = 10,
         field: str = "created_at",
         order: str = "DESC",
-    ) -> SecretList:
+        after: str | None = None,
+    ) -> PaginatedResponse[Secret]:
         """List secrets with pagination.
 
+        The returned object can be iterated over to fetch subsequent pages.
+
         :param limit: Maximum number of secrets to return.
         :param field: Field to sort by.
         :param order: Sort order (ASC or DESC).
+        :param after: The cursor to fetch the next page of results.
 
-        :returns: List of secrets with pagination info.
+        :returns: A `PaginatedResponse` object containing the first page of secrets.
         """
-        params = {
+        # 1. Prepare arguments for the initial API call
+        # TODO: Pagination in the deepset API is currently implemented in an unintuitive way.
+        # TODO: The cursor is always time based (created_at) and after signifies secrets older than the current cursor
+        # TODO: while 'before' signals secrets younger than the current cursor.
+        # TODO: This is applied irrespective of any sort (e.g. name) that would conflict with this approach.
+        # TODO: Change this to 'after' once the behaviour is fixed on the deepset API
+        request_params = {
             "limit": str(limit),
             "field": field,
             "order": order,
+            "before": after,
         }
+        request_params = {k: v for k, v in request_params.items() if v is not None}
+
+        # 2. Make the first API call using a private, stateless method
+        page = await self._list_api_call(**request_params)
+
+        # 3. Inject the logic needed for subsequent fetches into the response object
+        page._inject_paginator(
+            fetch_func=self._list_api_call,
+            # Base args for the *next* fetch don't include initial cursors
+            base_args={"limit": str(limit), "field": field, "order": order},
+        )
+        return page
 
+    async def _list_api_call(self, **kwargs: Any) -> PaginatedResponse[Secret]:
+        """A private, stateless method that performs the raw API call."""
         resp = await self._client.request(
             endpoint="v2/secrets",
             method="GET",
             response_type=dict[str, Any],
-            params=params,
+            params=kwargs,
         )
 
         raise_for_status(resp)
 
         if resp.json is None:
             raise ResourceNotFoundError("Failed to retrieve secrets.")
 
-        return SecretList(**resp.json)
+        return PaginatedResponse[Secret].create_with_cursor_field(resp.json, "secret_id")
 
     async def create(self, name: str, secret: str) -> NoContentResponse:
         """Create a new secret.

src/deepset_mcp/tools/secrets.py

@@ -23,39 +23,48 @@ class EnvironmentSecretList(BaseModel):
     data: list[EnvironmentSecret]
     has_more: bool
     total: int
+    next_cursor: str | None = None
 
 
-async def list_secrets(*, client: AsyncClientProtocol, limit: int = 10) -> EnvironmentSecretList | str:
+async def list_secrets(
+    *, client: AsyncClientProtocol, limit: int = 10, after: str | None = None
+) -> EnvironmentSecretList | str:
     """Lists all secrets available in the user's deepset organization.
 
     Use this tool to retrieve a list of secrets with their names and IDs.
     This is useful for getting an overview of all secrets before retrieving specific ones.
 
     :param client: The deepset API client
     :param limit: Maximum number of secrets to return (default: 10)
+    :param after: The cursor to fetch the next page of results
 
     :returns: List of secrets or error message
     """
     try:
-        secrets_list = await client.secrets().list(limit=limit)
-        integrations_list = await client.integrations().list()
+        secrets_list = await client.secrets().list(limit=limit, after=after)
 
         env_secrets = [EnvironmentSecret(name=secret.name, id=secret.secret_id) for secret in secrets_list.data]
-        for integration in integrations_list.integrations:
-            env_vars = TOKEN_DOMAIN_MAPPING.get(integration.provider_domain, [])
-            for env_var in env_vars:
-                env_secrets.append(
-                    EnvironmentSecret(
-                        name=env_var,
-                        id=str(integration.model_registry_token_id),
-                        invalid=integration.invalid,
+
+        # Only fetch integrations if no cursor is provided (first page)
+        # This optimizes performance by skipping the integrations call for subsequent pages
+        if after is None:
+            integrations_list = await client.integrations().list()
+            for integration in integrations_list.integrations:
+                env_vars = TOKEN_DOMAIN_MAPPING.get(integration.provider_domain, [])
+                for env_var in env_vars:
+                    env_secrets.append(
+                        EnvironmentSecret(
+                            name=env_var,
+                            id=str(integration.model_registry_token_id),
+                            invalid=integration.invalid,
+                        )
                     )
-                )
 
         return EnvironmentSecretList(
             data=env_secrets,
             has_more=secrets_list.has_more,
             total=len(env_secrets),
+            next_cursor=secrets_list.next_cursor,
         )
     except ResourceNotFoundError as e:
         return f"Error: {str(e)}"

test/integration/test_integration_secret_resource.py

@@ -6,9 +6,9 @@
 
 from deepset_mcp.api.client import AsyncDeepsetClient
 from deepset_mcp.api.exceptions import ResourceNotFoundError, UnexpectedAPIError
-from deepset_mcp.api.secrets.models import Secret, SecretList
+from deepset_mcp.api.secrets.models import Secret
 from deepset_mcp.api.secrets.resource import SecretResource
-from deepset_mcp.api.shared_models import NoContentResponse
+from deepset_mcp.api.shared_models import NoContentResponse, PaginatedResponse
 
 pytestmark = pytest.mark.integration
 
@@ -47,7 +47,7 @@ async def test_create_and_get_secret(
         assert create_result.success is True
 
         # List secrets to find our created secret
-        secrets: SecretList = await secret_resource.list()
+        secrets: PaginatedResponse[Secret] = await secret_resource.list()
 
         # Find our secret in the list
         created_secret = None
@@ -96,7 +96,7 @@ async def test_list_secrets(secret_resource: SecretResource) -> None:
                     break
 
         # Test listing all secrets
-        all_secrets: SecretList = await secret_resource.list()
+        all_secrets: PaginatedResponse[Secret] = await secret_resource.list()
         assert len(all_secrets.data) >= 3
 
         # Verify our created secrets are in the list
@@ -105,12 +105,12 @@ async def test_list_secrets(secret_resource: SecretResource) -> None:
             assert name in secret_names
 
         # Test listing with limit
-        limited_secrets: SecretList = await secret_resource.list(limit=2)
+        limited_secrets: PaginatedResponse[Secret] = await secret_resource.list(limit=2)
         assert len(limited_secrets.data) <= 2
 
         # Test with different sort order
-        asc_secrets: SecretList = await secret_resource.list(order="ASC")
-        desc_secrets: SecretList = await secret_resource.list(order="DESC")
+        asc_secrets: PaginatedResponse[Secret] = await secret_resource.list(order="ASC")
+        desc_secrets: PaginatedResponse[Secret] = await secret_resource.list(order="DESC")
 
         # At least verify we got results (can't easily test order without knowing full dataset)
         assert len(asc_secrets.data) >= 0
@@ -139,7 +139,7 @@ async def test_delete_secret(
         await secret_resource.create(name=test_secret_name, secret=test_secret_value)
 
         # Find the created secret
-        secrets: SecretList = await secret_resource.list()
+        secrets: PaginatedResponse[Secret] = await secret_resource.list()
         secret_to_delete = None
         for secret in secrets.data:
             if secret.name == test_secret_name:
@@ -196,3 +196,59 @@ async def test_delete_nonexistent_secret(secret_resource: SecretResource) -> Non
         # This is acceptable behavior for DELETE operations
         # API might return 404 or other error codes for nonexistent resources
         pass
+
+
+@pytest.mark.asyncio
+async def test_pagination_iteration(
+    secret_resource: SecretResource,
+) -> None:
+    """Test iterating over multiple pages of secrets using the async iterator."""
+    # Create several test secrets
+    test_secrets = [
+        ("test-pagination-secret-1", "value-1"),
+        ("test-pagination-secret-2", "value-2"),
+        ("test-pagination-secret-3", "value-3"),
+        ("test-pagination-secret-4", "value-4"),
+        ("test-pagination-secret-5", "value-5"),
+    ]
+
+    created_secret_ids = []
+
+    try:
+        # Create test secrets
+        for name, value in test_secrets:
+            await secret_resource.create(name=name, secret=value)
+            # Find the created secret ID for cleanup
+            secrets = await secret_resource.list()
+            for secret in secrets.data:
+                if secret.name == name and secret.secret_id not in created_secret_ids:
+                    created_secret_ids.append(secret.secret_id)
+                    break
+
+        # Get the first page with a small limit to ensure pagination
+        paginator = await secret_resource.list(limit=2)
+
+        # Collect all secrets by iterating through pages
+        all_secrets = []
+        async for secret in paginator:
+            all_secrets.append(secret)
+
+        # Verify we got at least our created secrets (might have more from other tests)
+        assert len(all_secrets) >= 5
+
+        # Verify all secrets are Secret instances
+        for secret in all_secrets:
+            assert isinstance(secret, Secret)
+
+        # Verify our created secrets are in the results
+        retrieved_names = [s.name for s in all_secrets]
+        for name, _ in test_secrets:
+            assert name in retrieved_names
+
+    finally:
+        # Clean up created secrets
+        for secret_id in created_secret_ids:
+            try:
+                await secret_resource.delete(secret_id)
+            except Exception as e:
+                print(f"Failed to delete test secret {secret_id}: {e}")