feat: add secret tools (#103)

Author: mathislucka

src/deepset_mcp/main.py

@@ -43,6 +43,10 @@
     list_pipeline_templates as list_pipeline_templates_tool,
     search_pipeline_templates as search_pipeline_templates_tool,
 )
+from deepset_mcp.tools.secrets import (
+    get_secret as get_secret_tool,
+    list_secrets as list_secrets_tool,
+)
 
 INITIALIZED_MODEL = StaticModel.from_pretrained("minishlab/potion-base-2M")
 
@@ -439,6 +443,34 @@ async def search_pipeline(pipeline_name: str, query: str) -> str:
     return response
 
 
+@mcp.tool()
+async def list_secrets(limit: int = 10) -> str:
+    """Lists all secrets available in the deepset workspace.
+
+    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 limit: Maximum number of secrets to return (default: 10).
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await list_secrets_tool(client, limit)
+    return response
+
+
+@mcp.tool()
+async def get_secret(secret_id: str) -> str:
+    """Retrieves detailed information about a specific secret by its ID.
+
+    Use this tool to get information about a specific secret when you know its ID.
+    The secret value itself is not returned for security reasons, only metadata.
+
+    :param secret_id: The unique identifier of the secret to retrieve.
+    """
+    async with AsyncDeepsetClient() as client:
+        response = await get_secret_tool(client, secret_id)
+    return response
+
+
 # Check if docs search should be enabled
 docs_config = get_docs_config()
 if docs_config:

src/deepset_mcp/tools/secrets.py

@@ -0,0 +1,64 @@
+from deepset_mcp.api.exceptions import ResourceNotFoundError, UnexpectedAPIError
+from deepset_mcp.api.protocols import AsyncClientProtocol
+
+
+async def list_secrets(client: AsyncClientProtocol, limit: int = 10) -> 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)
+
+    :returns: A formatted string containing secret names and IDs
+    """
+    try:
+        response = await client.secrets().list(limit=limit)
+
+        if not response.data:
+            return "No secrets found in this workspace."
+
+        secrets_info = []
+        for secret in response.data:
+            secrets_info.append(f"Name: {secret.name}, ID: {secret.secret_id}")
+
+        result = f"Found {len(response.data)} secret(s):\n" + "\n".join(secrets_info)
+
+        if response.has_more:
+            result += (
+                f"\n\nShowing {len(response.data)} of {response.total} total secrets. Use a higher limit to see more."
+            )
+
+        return result
+
+    except ResourceNotFoundError as e:
+        return f"Error: {str(e)}"
+    except UnexpectedAPIError as e:
+        return f"API Error: {str(e)}"
+    except Exception as e:
+        return f"Unexpected error: {str(e)}"
+
+
+async def get_secret(client: AsyncClientProtocol, secret_id: str) -> str:
+    """Retrieves detailed information about a specific secret by its ID.
+
+    Use this tool to get information about a specific secret when you know its ID.
+    The secret value itself is not returned for security reasons, only metadata.
+
+    :param client: The deepset API client
+    :param secret_id: The unique identifier of the secret to retrieve
+
+    :returns: A formatted string containing secret information
+    """
+    try:
+        response = await client.secrets().get(secret_id)
+
+        return f"Secret Details:\nName: {response.name}\nID: {response.secret_id}"
+
+    except ResourceNotFoundError as e:
+        return f"Error: {str(e)}"
+    except UnexpectedAPIError as e:
+        return f"API Error: {str(e)}"
+    except Exception as e:
+        return f"Unexpected error: {str(e)}"

test/unit/tools/test_secrets.py

@@ -0,0 +1,169 @@
+import pytest
+
+from deepset_mcp.api.exceptions import ResourceNotFoundError, UnexpectedAPIError
+from deepset_mcp.api.protocols import SecretResourceProtocol
+from deepset_mcp.api.secrets.models import Secret, SecretList
+from deepset_mcp.api.shared_models import NoContentResponse
+from deepset_mcp.tools.secrets import get_secret, list_secrets
+from test.unit.conftest import BaseFakeClient
+
+
+class FakeSecretResource(SecretResourceProtocol):
+    def __init__(
+        self,
+        list_response: SecretList | None = None,
+        get_response: Secret | None = None,
+        list_exception: Exception | None = None,
+        get_exception: Exception | None = None,
+    ) -> None:
+        self.list_response = list_response
+        self.get_response = get_response
+        self.list_exception = list_exception
+        self.get_exception = get_exception
+
+    async def list(self, limit: int = 10, field: str = "created_at", order: str = "DESC") -> SecretList:
+        if self.list_exception:
+            raise self.list_exception
+        if self.list_response is None:
+            raise ValueError("No list response configured")
+        return self.list_response
+
+    async def get(self, secret_id: str) -> Secret:
+        if self.get_exception:
+            raise self.get_exception
+        if self.get_response is None:
+            raise ValueError("No get response configured")
+        return self.get_response
+
+    async def create(self, name: str, secret: str) -> NoContentResponse:
+        """Not used in tests, but required by protocol."""
+        return NoContentResponse(message="Created")
+
+    async def delete(self, secret_id: str) -> NoContentResponse:
+        """Not used in tests, but required by protocol."""
+        return NoContentResponse(message="Deleted")
+
+
+class FakeClientWithSecrets(BaseFakeClient):
+    def __init__(self, secret_resource: FakeSecretResource) -> None:
+        super().__init__()
+        self._secret_resource = secret_resource
+
+    def secrets(self) -> FakeSecretResource:
+        return self._secret_resource
+
+
+@pytest.mark.asyncio
+async def test_list_secrets_success() -> None:
+    """Test successful listing of secrets."""
+    secrets_data = [
+        Secret(name="api-key", secret_id="secret-1"),
+        Secret(name="database-password", secret_id="secret-2"),
+    ]
+    secret_list = SecretList(data=secrets_data, has_more=False, total=2)
+    fake_resource = FakeSecretResource(list_response=secret_list)
+    client = FakeClientWithSecrets(fake_resource)
+
+    result = await list_secrets(client, limit=10)
+
+    expected = "Found 2 secret(s):\nName: api-key, ID: secret-1\nName: database-password, ID: secret-2"
+    assert result == expected
+
+
+@pytest.mark.asyncio
+async def test_list_secrets_with_pagination() -> None:
+    """Test listing secrets with pagination info."""
+    secrets_data = [
+        Secret(name="api-key", secret_id="secret-1"),
+    ]
+    secret_list = SecretList(data=secrets_data, has_more=True, total=5)
+    fake_resource = FakeSecretResource(list_response=secret_list)
+    client = FakeClientWithSecrets(fake_resource)
+
+    result = await list_secrets(client, limit=1)
+
+    expected = (
+        "Found 1 secret(s):\nName: api-key, ID: secret-1\n\n"
+        "Showing 1 of 5 total secrets. Use a higher limit to see more."
+    )
+    assert result == expected
+
+
+@pytest.mark.asyncio
+async def test_list_secrets_empty() -> None:
+    """Test listing when no secrets exist."""
+    secret_list = SecretList(data=[], has_more=False, total=0)
+    fake_resource = FakeSecretResource(list_response=secret_list)
+    client = FakeClientWithSecrets(fake_resource)
+
+    result = await list_secrets(client)
+
+    assert result == "No secrets found in this workspace."
+
+
+@pytest.mark.asyncio
+async def test_list_secrets_unexpected_api_error() -> None:
+    """Test handling of UnexpectedAPIError during list."""
+    fake_resource = FakeSecretResource(list_exception=UnexpectedAPIError(500, "Internal server error"))
+    client = FakeClientWithSecrets(fake_resource)
+
+    result = await list_secrets(client)
+
+    assert result == "API Error: Internal server error (Status Code: 500)"
+
+
+@pytest.mark.asyncio
+async def test_list_secrets_generic_exception() -> None:
+    """Test handling of generic exceptions during list."""
+    fake_resource = FakeSecretResource(list_exception=ValueError("Generic error"))
+    client = FakeClientWithSecrets(fake_resource)
+
+    result = await list_secrets(client)
+
+    assert result == "Unexpected error: Generic error"
+
+
+@pytest.mark.asyncio
+async def test_get_secret_success() -> None:
+    """Test successful retrieval of a specific secret."""
+    secret = Secret(name="api-key", secret_id="secret-1")
+    fake_resource = FakeSecretResource(get_response=secret)
+    client = FakeClientWithSecrets(fake_resource)
+
+    result = await get_secret(client, "secret-1")
+
+    expected = "Secret Details:\nName: api-key\nID: secret-1"
+    assert result == expected
+
+
+@pytest.mark.asyncio
+async def test_get_secret_not_found() -> None:
+    """Test handling when secret is not found."""
+    fake_resource = FakeSecretResource(get_exception=ResourceNotFoundError("Secret 'nonexistent' not found."))
+    client = FakeClientWithSecrets(fake_resource)
+
+    result = await get_secret(client, "nonexistent")
+
+    assert result == "Error: Secret 'nonexistent' not found. (Status Code: 404)"
+
+
+@pytest.mark.asyncio
+async def test_get_secret_unexpected_api_error() -> None:
+    """Test handling of UnexpectedAPIError during get."""
+    fake_resource = FakeSecretResource(get_exception=UnexpectedAPIError(500, "Server error"))
+    client = FakeClientWithSecrets(fake_resource)
+
+    result = await get_secret(client, "secret-1")
+
+    assert result == "API Error: Server error (Status Code: 500)"
+
+
+@pytest.mark.asyncio
+async def test_get_secret_generic_exception() -> None:
+    """Test handling of generic exceptions during get."""
+    fake_resource = FakeSecretResource(get_exception=ValueError("Something went wrong"))
+    client = FakeClientWithSecrets(fake_resource)
+
+    result = await get_secret(client, "secret-1")
+
+    assert result == "Unexpected error: Something went wrong"