chore: add query filter to search history tool (#208)

Author: YassineGabsi

src/deepset_mcp/api/search_history/protocols.py

@@ -13,23 +13,27 @@
 class SearchHistoryResourceProtocol(Protocol):
     """Protocol defining the interface for search history resources."""
 
-    async def list(self, limit: int = 10, after: str | None = None) -> PaginatedResponse[SearchHistoryEntry]:
+    async def list(
+        self, limit: int = 10, after: str | None = None, query_filter: 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.
+        :param query_filter: OData filter expression to narrow results.
         :returns: Paginated response of search history entries.
         """
         ...
 
     async def list_pipeline(
-        self, pipeline_name: str, limit: int = 10, after: str | None = None
+        self, pipeline_name: str, limit: int = 10, after: str | None = None, query_filter: 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.
+        :param query_filter: OData filter expression to narrow results.
         :returns: Paginated response of search history entries (most recent first).
         """
         ...

src/deepset_mcp/api/search_history/resource.py

@@ -36,16 +36,21 @@ def _pipeline_path(self, pipeline_name: str) -> str:
             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]:
+    async def list(
+        self, limit: int = 10, after: str | None = None, query_filter: 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.
+        :param query_filter: OData filter expression to narrow results.
         :returns: Paginated response of search history entries.
         """
         params: dict[str, str | int] = {"limit": limit}
         if after is not None:
             params["after"] = after
+        if query_filter is not None:
+            params["filter"] = query_filter
 
         resp = await self._client.request(
             endpoint=self._base_path(),
@@ -80,7 +85,7 @@ async def list(self, limit: int = 10, after: str | None = None) -> PaginatedResp
         )
 
     async def list_pipeline(
-        self, pipeline_name: str, limit: int = 10, after: str | None = None
+        self, pipeline_name: str, limit: int = 10, after: str | None = None, query_filter: str | None = None
     ) -> PaginatedResponse[SearchHistoryEntry]:
         """List search history entries for a specific pipeline with pagination.
 
@@ -89,11 +94,14 @@ async def list_pipeline(
         :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.
+        :param query_filter: OData filter expression to narrow results.
         :returns: Paginated response of search history entries.
         """
         params: dict[str, str | int] = {"limit": limit}
         if after is not None:
             params["after"] = after
+        if query_filter is not None:
+            params["filter"] = query_filter
 
         resp = await self._client.request(
             endpoint=f"{self._pipeline_path(pipeline_name)}_archive",

src/deepset_mcp/tools/search_history.py

@@ -11,7 +11,12 @@
 
 
 async def list_search_history(
-    *, client: AsyncClientProtocol, workspace: str, limit: int = 10, after: str | None = None
+    *,
+    client: AsyncClientProtocol,
+    workspace: str,
+    limit: int = 10,
+    after: str | None = None,
+    query_filter: str | None = None,
 ) -> PaginatedResponse[SearchHistoryEntry] | str:
     """Retrieves search history for the configured deepset workspace.
 
@@ -24,10 +29,18 @@ async def list_search_history(
     :param limit: Maximum number of entries to return per page.
     :param after: The cursor to fetch the next page of results.
         If there are more results to fetch, the cursor will appear as `next_cursor` on the response.
+    :param query_filter: An OData filter expression to narrow down results.
+        Supported fields: query, client_source_path, pipeline_version_id, answer, api_key,
+        created_at, created_by, tags/tag_id, feedbacks, feedbacks/score, feedbacks/comment,
+        feedbacks/bookmarked, session_id, search_session_id, feedbacks/result_id,
+        request/filters, request/params, duration.
+        Example: "created_at ge 2024-01-01T00:00:00Z" or "query eq 'my search'".
     :returns: Paginated list of search history entries or error message.
     """
     try:
-        return await client.search_history(workspace=workspace).list(limit=limit, after=after)
+        return await client.search_history(workspace=workspace).list(
+            limit=limit, after=after, query_filter=query_filter
+        )
     except ResourceNotFoundError:
         return f"There is no workspace named '{workspace}'. Did you mean to configure it?"
     except (BadRequestError, UnexpectedAPIError) as e:
@@ -41,6 +54,7 @@ async def list_pipeline_search_history(
     pipeline_name: str,
     limit: int = 10,
     after: str | None = None,
+    query_filter: str | None = None,
 ) -> PaginatedResponse[SearchHistoryEntry] | str:
     """Retrieves search history for a specific pipeline with pagination.
 
@@ -54,11 +68,17 @@ async def list_pipeline_search_history(
     :param limit: Maximum number of entries to return per page.
     :param after: The cursor to fetch the next page of results.
         If there are more results to fetch, the cursor will appear as `next_cursor` on the response.
+    :param query_filter: An OData filter expression to narrow down results.
+        Supported fields: query, client_source_path, pipeline_version_id, answer, api_key,
+        created_at, created_by, tags/tag_id, feedbacks, feedbacks/score, feedbacks/comment,
+        feedbacks/bookmarked, session_id, search_session_id, feedbacks/result_id,
+        request/filters, request/params, duration.
+        Example: "created_at ge 2024-01-01T00:00:00Z" or "query eq 'my search'".
     :returns: Paginated list of search history entries or error message.
     """
     try:
         return await client.search_history(workspace=workspace).list_pipeline(
-            pipeline_name=pipeline_name, limit=limit, after=after
+            pipeline_name=pipeline_name, limit=limit, after=after, query_filter=query_filter
         )
     except ResourceNotFoundError:
         return (

test/unit/tools/test_search_history.py

@@ -27,15 +27,17 @@ def __init__(
         self._list_exception = list_exception
         self._list_pipeline_exception = list_pipeline_exception
 
-    async def list(self, limit: int = 10, after: str | None = None) -> PaginatedResponse[SearchHistoryEntry]:
+    async def list(
+        self, limit: int = 10, after: str | None = None, query_filter: str | None = None
+    ) -> PaginatedResponse[SearchHistoryEntry]:
         if self._list_exception:
             raise self._list_exception
         if self._list_response is not None:
             return self._list_response
         return PaginatedResponse(data=[], has_more=False, total=0)
 
     async def list_pipeline(
-        self, pipeline_name: str, limit: int = 10, after: str | None = None
+        self, pipeline_name: str, limit: int = 10, after: str | None = None, query_filter: str | None = None
     ) -> PaginatedResponse[SearchHistoryEntry]:
         if self._list_pipeline_exception:
             raise self._list_pipeline_exception