feat: [ECO-200] - Add Search history tool

README.md

@@ -29,6 +29,37 @@ pipx install uv
 uv sync --locked --all-extras --all-groups
 ```
 
+### Local Development
+
+If you want to test your changes locally, follow these steps:
+
+1. Add a script run-deepset-mcp.sh that uses the binary from the project's virtual env
+
+```bash
+#!/usr/bin/env bash
+# Wrapper to run the local deepset-mcp server for Cursor MCP.
+# Use this as command so it doesn't depend on uv or PATH.
+set -e
+cd "$(dirname "$0")"
+exec .venv/bin/deepset-mcp
+```
+
+2. Use it this way in Cursor:
+
+```bash
+    "deepset": {
+      "command": "/bin/bash",
+      "args": ["/Users/*****/****/deepset-mcp-server/run-deepset-mcp.sh"],
+      "cwd": "/Users/*****/****/deepset-mcp-server",
+      "env": {
+        "DEEPSET_WORKSPACE": "WORKSPACE",
+        "DEEPSET_API_KEY": "API_KEY"
+      }
+    }
+```
+
+Note: If you change the codebase, make sure to restart the MCP server.
+
 ### Code Quality & Testing
 
 Run code quality checks and tests using the Makefile:
@@ -60,6 +91,3 @@ Documentation is built using [MkDocs](https://www.mkdocs.org/) with the Material
 - Content: `docs/` directory
 - Auto-generated API docs via [mkdocstrings](https://mkdocstrings.github.io/)
 - Deployed via GitHub Pages (automated via GitHub Actions on push to main branch)
-
-
-

src/deepset_mcp/api/client.py

@@ -15,6 +15,7 @@
 from deepset_mcp.api.pipeline.resource import PipelineResource
 from deepset_mcp.api.pipeline_template.resource import PipelineTemplateResource
 from deepset_mcp.api.protocols import AsyncClientProtocol
+from deepset_mcp.api.search_history.resource import SearchHistoryResource
 from deepset_mcp.api.secrets.resource import SecretResource
 from deepset_mcp.api.transport import (
     AsyncTransport,
@@ -103,6 +104,14 @@ def pipeline_templates(self, workspace: str) -> PipelineTemplateResource:
         """
         return PipelineTemplateResource(client=self, workspace=workspace)
 
+    def search_history(self, workspace: str) -> SearchHistoryResource:
+        """Resource to interact with search history in the specified workspace.
+
+        :param workspace: Workspace identifier
+        :returns: Search history resource instance
+        """
+        return SearchHistoryResource(client=self, workspace=workspace)
+
     def haystack_service(self) -> HaystackServiceResource:
         """Resource to interact with the Haystack service API.
 

src/deepset_mcp/api/protocols.py

@@ -15,6 +15,7 @@
     from deepset_mcp.api.integrations.protocols import IntegrationResourceProtocol
     from deepset_mcp.api.pipeline.protocols import PipelineResourceProtocol
     from deepset_mcp.api.pipeline_template.protocols import PipelineTemplateResourceProtocol
+    from deepset_mcp.api.search_history.protocols import SearchHistoryResourceProtocol
     from deepset_mcp.api.secrets.protocols import SecretResourceProtocol
     from deepset_mcp.api.user.protocols import UserResourceProtocol
     from deepset_mcp.api.workspace.protocols import WorkspaceResourceProtocol
@@ -126,3 +127,7 @@ def workspaces(self) -> "WorkspaceResourceProtocol":
     def integrations(self) -> "IntegrationResourceProtocol":
         """Access integrations."""
         ...
+
+    def search_history(self, workspace: str) -> "SearchHistoryResourceProtocol":
+        """Access search history in the specified workspace."""
+        ...

src/deepset_mcp/api/search_history/__init__.py

@@ -0,0 +1,10 @@
+# SPDX-FileCopyrightText: 2025-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Search history API module."""
+
+from .models import SearchHistoryEntry
+from .resource import SearchHistoryResource
+
+__all__ = ["SearchHistoryEntry", "SearchHistoryResource"]

src/deepset_mcp/api/search_history/models.py

@@ -0,0 +1,35 @@
+# SPDX-FileCopyrightText: 2025-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Models for search history API."""
+
+from typing import Any
+
+from pydantic import BaseModel, Field, field_validator
+
+
+class SearchHistoryEntry(BaseModel):
+    """A single search history entry from the deepset platform.
+
+    Contains query, answers, prompts, feedback, and other metadata.
+    """
+
+    model_config = {"extra": "allow"}
+
+    query: str | None = Field(default=None, description="The search query that was executed")
+    answer: str | None = Field(default=None, description="The answer returned by the pipeline")
+    created_at: str | None = Field(default=None, description="When the search was performed")
+    pipeline_name: str | None = Field(default=None, description="Name of the pipeline used")
+    feedback: list[dict[str, Any]] | None = Field(default=None, description="User feedback on the search")
+
+    @field_validator("feedback", mode="before")
+    @classmethod
+    def _feedback_to_list(cls, v: Any) -> list[dict[str, Any]] | None:
+        if v is None:
+            return None
+        if isinstance(v, list):
+            return v
+        if isinstance(v, dict):
+            return [v]
+        return None

src/deepset_mcp/api/search_history/protocols.py

@@ -0,0 +1,35 @@
+# SPDX-FileCopyrightText: 2025-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Protocols for search history resources."""
+
+from typing import Protocol
+
+from deepset_mcp.api.search_history.models import SearchHistoryEntry
+from deepset_mcp.api.shared_models import PaginatedResponse
+
+
+class SearchHistoryResourceProtocol(Protocol):
+    """Protocol defining the interface for search history resources."""
+
+    async def list(self, limit: int = 10, after: str | None = None) -> PaginatedResponse[SearchHistoryEntry]:
+        """List search history entries in the workspace.
+
+        :param limit: Maximum number of entries to return per page.
+        :param after: Cursor to fetch the next page of results.
+        :returns: Paginated response of search history entries.
+        """
+        ...
+
+    async def list_pipeline(
+        self, pipeline_name: str, limit: int = 10, after: str | None = None
+    ) -> PaginatedResponse[SearchHistoryEntry]:
+        """List search history entries for a specific pipeline with pagination.
+
+        :param pipeline_name: Name of the pipeline.
+        :param limit: Maximum number of entries to return per page.
+        :param after: Cursor to fetch the next page of results.
+        :returns: Paginated response of search history entries (most recent first).
+        """
+        ...

src/deepset_mcp/api/search_history/resource.py

@@ -0,0 +1,127 @@
+# SPDX-FileCopyrightText: 2025-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Resource implementation for search history API."""
+
+from typing import TYPE_CHECKING
+from urllib.parse import quote
+
+from deepset_mcp.api.search_history.models import SearchHistoryEntry
+from deepset_mcp.api.search_history.protocols import SearchHistoryResourceProtocol
+from deepset_mcp.api.shared_models import PaginatedResponse
+from deepset_mcp.api.transport import raise_for_status
+
+if TYPE_CHECKING:
+    from deepset_mcp.api.protocols import AsyncClientProtocol
+
+
+class SearchHistoryResource(SearchHistoryResourceProtocol):
+    """Manages interactions with the deepset search history API."""
+
+    def __init__(self, client: "AsyncClientProtocol", workspace: str) -> None:
+        """Initialize the search history resource.
+
+        :param client: The async REST client.
+        :param workspace: The workspace to use.
+        """
+        self._client = client
+        self._workspace = workspace
+
+    def _base_path(self) -> str:
+        return f"v1/workspaces/{quote(self._workspace, safe='')}/search_history"
+
+    def _pipeline_path(self, pipeline_name: str) -> str:
+        return (
+            f"v1/workspaces/{quote(self._workspace, safe='')}/pipelines/{quote(pipeline_name, safe='')}/search_history"
+        )
+
+    async def list(self, limit: int = 10, after: str | None = None) -> PaginatedResponse[SearchHistoryEntry]:
+        """List search history entries in the workspace.
+
+        :param limit: Maximum number of entries to return per page.
+        :param after: Cursor to fetch the next page of results.
+        :returns: Paginated response of search history entries.
+        """
+        params: dict[str, str | int] = {"limit": limit}
+        if after is not None:
+            params["after"] = after
+
+        resp = await self._client.request(
+            endpoint=self._base_path(),
+            method="GET",
+            params=params,
+            timeout=70.0,
+        )
+
+        raise_for_status(resp)
+
+        if resp.json is None:
+            return PaginatedResponse(
+                data=[],
+                has_more=False,
+                total=0,
+                next_cursor=None,
+            )
+
+        # API may return paginated shape: { "data": [...], "has_more": bool, "total": int }
+        data = resp.json if isinstance(resp.json, dict) else {"data": resp.json}
+        items = data.get("data", [])
+        if not isinstance(items, list):
+            items = []
+
+        return PaginatedResponse[SearchHistoryEntry].create_with_cursor_field(
+            {
+                "data": items,
+                "has_more": data.get("has_more", False),
+                "total": data.get("total"),
+            },
+            "created_at",
+        )
+
+    async def list_pipeline(
+        self, pipeline_name: str, limit: int = 10, after: str | None = None
+    ) -> PaginatedResponse[SearchHistoryEntry]:
+        """List search history entries for a specific pipeline with pagination.
+
+        Uses the pipeline search history archive endpoint (full history, most recent first).
+
+        :param pipeline_name: Name of the pipeline.
+        :param limit: Maximum number of entries to return per page.
+        :param after: Cursor to fetch the next page of results.
+        :returns: Paginated response of search history entries.
+        """
+        params: dict[str, str | int] = {"limit": limit}
+        if after is not None:
+            params["after"] = after
+
+        resp = await self._client.request(
+            endpoint=f"{self._pipeline_path(pipeline_name)}_archive",
+            method="GET",
+            params=params,
+            timeout=70.0,
+        )
+
+        raise_for_status(resp)
+
+        if resp.json is None:
+            return PaginatedResponse(
+                data=[],
+                has_more=False,
+                total=0,
+                next_cursor=None,
+            )
+
+        data = resp.json if isinstance(resp.json, dict) else {"data": resp.json}
+        items = data.get("data", [])
+        if not isinstance(items, list):
+            items = []
+
+        return PaginatedResponse[SearchHistoryEntry].create_with_cursor_field(
+            {
+                "data": items,
+                "has_more": data.get("has_more", False),
+                "total": data.get("total"),
+            },
+            "created_at",
+        )

src/deepset_mcp/mcp/tool_registry.py

@@ -47,6 +47,10 @@
     list_templates as list_pipeline_templates_tool,
     search_templates as search_pipeline_templates_tool,
 )
+from deepset_mcp.tools.search_history import (
+    list_pipeline_search_history as list_pipeline_search_history_tool,
+    list_search_history as list_search_history_tool,
+)
 from deepset_mcp.tools.secrets import get_secret as get_secret_tool, list_secrets as list_secrets_tool
 from deepset_mcp.tools.workspace import (
     create_workspace as create_workspace_tool,
@@ -184,6 +188,14 @@ async def search_docs(query: str) -> str:
             custom_args={"model": get_initialized_model()},
         ),
     ),
+    "list_search_history": (
+        list_search_history_tool,
+        ToolConfig(needs_client=True, needs_workspace=True, memory_type=MemoryType.EXPLORABLE),
+    ),
+    "list_pipeline_search_history": (
+        list_pipeline_search_history_tool,
+        ToolConfig(needs_client=True, needs_workspace=True, memory_type=MemoryType.EXPLORABLE),
+    ),
     "list_custom_component_installations": (
         list_custom_component_installations_tool,
         ToolConfig(needs_client=True, needs_workspace=True, memory_type=MemoryType.EXPLORABLE),

src/deepset_mcp/tools/__init__.py

@@ -26,6 +26,7 @@
     validate_pipeline,
 )
 from .pipeline_template import get_template, list_templates, search_templates
+from .search_history import list_pipeline_search_history, list_search_history
 from .secrets import get_secret, list_secrets
 from .workspace import get_workspace, list_workspaces
 
@@ -59,6 +60,8 @@
     "list_templates",
     "get_template",
     "search_templates",
+    "list_search_history",
+    "list_pipeline_search_history",
     "get_secret",
     "list_secrets",
     "list_workspaces",