Add new MCP tool for "search with filters" (#202)

* add new tool for search with filters

* add link to filter docs

* fix: format with ruff

* add imports

Author: tholor

src/deepset_mcp/mcp/tool_registry.py

@@ -37,6 +37,7 @@
     get_pipeline_logs as get_pipeline_logs_tool,
     list_pipelines as list_pipelines_tool,
     search_pipeline as search_pipeline_tool,
+    search_pipeline_with_filters as search_pipeline_with_filters_tool,
     update_pipeline as update_pipeline_tool,
     validate_pipeline as validate_pipeline_tool,
 )
@@ -128,6 +129,10 @@ async def search_docs(query: str) -> str:
         search_pipeline_tool,
         ToolConfig(needs_client=True, needs_workspace=True, memory_type=MemoryType.EXPLORABLE),
     ),
+    "search_pipeline_with_filters": (
+        search_pipeline_with_filters_tool,
+        ToolConfig(needs_client=True, needs_workspace=True, memory_type=MemoryType.EXPLORABLE),
+    ),
     "list_indexes": (
         list_indexes_tool,
         ToolConfig(needs_client=True, needs_workspace=True, memory_type=MemoryType.EXPLORABLE),

src/deepset_mcp/tools/__init__.py

@@ -20,6 +20,7 @@
     get_pipeline_logs,
     list_pipelines,
     search_pipeline,
+    search_pipeline_with_filters,
     update_pipeline,
     validate_pipeline,
 )
@@ -49,6 +50,7 @@
     "get_pipeline_logs",
     "deploy_pipeline",
     "search_pipeline",
+    "search_pipeline_with_filters",
     "create_pipeline",
     "update_pipeline",
     "validate_pipeline",

src/deepset_mcp/tools/pipeline.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import asyncio
+from typing import Any
 
 import yaml
 from pydantic import BaseModel
@@ -367,3 +368,52 @@ async def search_pipeline(
         return f"Failed to search using pipeline '{pipeline_name}': {e}"
     except Exception as e:
         return f"An unexpected error occurred while searching with pipeline '{pipeline_name}': {str(e)}"
+
+
+async def search_pipeline_with_filters(
+    *,
+    client: AsyncClientProtocol,
+    workspace: str,
+    pipeline_name: str,
+    query: str,
+    filters: dict[str, Any] | None = None,
+) -> DeepsetSearchResponse | str:
+    """Searches using a pipeline with filters.
+
+    Uses the specified pipeline to perform a search with the given query and filters.
+    Filters follow the Haystack filter syntax: https://docs.haystack.deepset.ai/docs/metadata-filtering.
+    Before executing the search, checks if the pipeline is deployed (status = DEPLOYED).
+    Returns search results.
+
+    :param client: The async client for API communication.
+    :param workspace: The workspace name.
+    :param pipeline_name: Name of the pipeline to use for search.
+    :param query: The search query to execute.
+    :param filters: The filters to apply to the search.
+
+    :returns: Search results or error message.
+    """
+    try:
+        # First, check if the pipeline exists and get its status
+        pipeline = await client.pipelines(workspace=workspace).get(pipeline_name=pipeline_name)
+
+        # Check if pipeline is deployed
+        if pipeline.status != "DEPLOYED":
+            return (
+                f"Pipeline '{pipeline_name}' is not deployed (current status: {pipeline.status}). "
+                f"Please deploy the pipeline first using the deploy_pipeline tool before attempting to search."
+            )
+
+        # Execute the search
+        return await client.pipelines(workspace=workspace).search(
+            pipeline_name=pipeline_name, query=query, filters=filters if filters is not None else None
+        )
+
+    except ResourceNotFoundError:
+        return f"There is no pipeline named '{pipeline_name}' in workspace '{workspace}'."
+    except BadRequestError as e:
+        return f"Failed to search using pipeline '{pipeline_name}': {e}"
+    except UnexpectedAPIError as e:
+        return f"Failed to search using pipeline '{pipeline_name}': {e}"
+    except Exception as e:
+        return f"An unexpected error occurred while searching with pipeline '{pipeline_name}': {str(e)}"