feat: add doc search tool (#101)

Author: mathislucka

src/deepset_mcp/main.py

@@ -1,4 +1,5 @@
 import argparse
+import logging
 import os
 from pathlib import Path
 
@@ -11,6 +12,10 @@
     get_latest_custom_component_installation_logs as get_latest_custom_component_installation_logs_tool,
     list_custom_component_installations as list_custom_component_installations_tool,
 )
+from deepset_mcp.tools.doc_search import (
+    get_docs_config,
+    search_docs as search_docs_tool,
+)
 from deepset_mcp.tools.haystack_service import (
     get_component_definition as get_component_definition_tool,
     get_custom_components as get_custom_components_tool,
@@ -434,6 +439,41 @@ async def search_pipeline(pipeline_name: str, query: str) -> str:
     return response
 
 
+# Check if docs search should be enabled
+docs_config = get_docs_config()
+if docs_config:
+    docs_workspace, docs_pipeline_name, docs_api_key = docs_config
+
+    async def search_docs(query: str) -> str:
+        """Search the deepset platform documentation.
+
+        This tool allows you to search through deepset's official documentation to find
+        information about features, API usage, best practices, and troubleshooting guides.
+        Use this when you need to look up specific deepset functionality or help users
+        understand how to use deepset features.
+
+        :param query: The search query to execute against the documentation.
+        :returns: The formatted search results from the documentation.
+        """
+        async with AsyncDeepsetClient(api_key=docs_api_key) as client:
+            response = await search_docs_tool(
+                client=client,
+                workspace=docs_workspace,
+                pipeline_name=docs_pipeline_name,
+                query=query,
+            )
+        return response
+
+    # Add the tool to the server
+    mcp.add_tool(search_docs)
+
+else:
+    logging.warning(
+        "Documentation search tool not enabled. To enable, set the following environment variables: "
+        "DEEPSET_DOCS_WORKSPACE, DEEPSET_DOCS_PIPELINE_NAME, DEEPSET_DOCS_API_KEY"
+    )
+
+
 def main() -> None:
     """Entrypoint for the deepset MCP server."""
     parser = argparse.ArgumentParser(description="Run the Deepset MCP server.")
@@ -447,11 +487,27 @@ def main() -> None:
         "-k",
         help="Deepset API key (env DEEPSET_API_KEY)",
     )
+    parser.add_argument(
+        "--docs-workspace",
+        help="Deepset docs search workspace (env DEEPSET_DOCS_WORKSPACE)",
+    )
+    parser.add_argument(
+        "--docs-pipeline-name",
+        help="Deepset docs pipeline name (env DEEPSET_DOCS_PIPELINE_NAME)",
+    )
+    parser.add_argument(
+        "--docs-api-key",
+        help="Deepset docs pipeline API key (env DEEPSET_DOCS_API_KEY)",
+    )
     args = parser.parse_args()
 
     # prefer flags, fallback to env
     workspace = args.workspace or os.getenv("DEEPSET_WORKSPACE")
     api_key = args.api_key or os.getenv("DEEPSET_API_KEY")
+    docs_workspace = args.docs_workspace or os.getenv("DEEPSET_DOCS_WORKSPACE")
+    docs_pipeline_name = args.docs_pipeline_name or os.getenv("DEEPSET_DOCS_PIPELINE_NAME")
+    docs_api_key = args.docs_api_key or os.getenv("DEEPSET_DOCS_API_KEY")
+
     if not workspace:
         parser.error("Missing workspace: set --workspace or DEEPSET_WORKSPACE")
     if not api_key:
@@ -461,6 +517,14 @@ def main() -> None:
     os.environ["DEEPSET_WORKSPACE"] = workspace
     os.environ["DEEPSET_API_KEY"] = api_key
 
+    # Set docs environment variables if provided
+    if docs_workspace:
+        os.environ["DEEPSET_DOCS_WORKSPACE"] = docs_workspace
+    if docs_pipeline_name:
+        os.environ["DEEPSET_DOCS_PIPELINE_NAME"] = docs_pipeline_name
+    if docs_api_key:
+        os.environ["DEEPSET_DOCS_API_KEY"] = docs_api_key
+
     # run with SSE transport (HTTP+Server-Sent Events)
     mcp.run(transport="stdio")
 

src/deepset_mcp/tools/doc_search.py

@@ -0,0 +1,88 @@
+import os
+
+from deepset_mcp.api.exceptions import BadRequestError, ResourceNotFoundError, UnexpectedAPIError
+from deepset_mcp.api.pipeline.models import DeepsetSearchResponse
+from deepset_mcp.api.protocols import AsyncClientProtocol
+
+
+def doc_search_results_to_llm_readable_string(results: DeepsetSearchResponse) -> str:
+    """Formats results of the doc search pipeline so that they can be read by an LLM.
+
+    :param results: DeepsetSearchResponse object
+    :return: Formatted results.
+    """
+    file_segmented_docs = []
+
+    previous_source_id = None
+    for doc in results.documents:
+        if previous_source_id != doc.meta["source_id"]:
+            file_segmented_docs.append([{"content": doc.content, "file_path": doc.meta.get("original_file_path", "")}])
+            previous_source_id = doc.meta.get("source_id")
+        else:
+            file_segmented_docs[-1].append(
+                {"content": doc.content, "file_path": doc.meta.get("original_file_path", "")}
+            )
+
+    files = []
+    for file_docs in file_segmented_docs:
+        start = file_docs[0]["file_path"]
+        full_doc = " ".join([doc["content"] for doc in file_docs])
+        files.append(start + "\n" + full_doc)
+
+    return "\n----\n".join(files)
+
+
+async def search_docs(
+    client: AsyncClientProtocol,
+    workspace: str,
+    pipeline_name: str,
+    query: str,
+) -> str:
+    """Search deepset documentation using a dedicated docs pipeline.
+
+    Uses the specified pipeline to perform a search with the given query against the deepset
+    documentation. Before executing the search, checks if the pipeline is deployed (status = DEPLOYED).
+    Returns search results in a human-readable format.
+
+    :param client: The async client for API communication.
+    :param workspace: The workspace name for the docs pipeline.
+    :param pipeline_name: Name of the pipeline to use for doc search.
+    :param query: The search query to execute.
+    :returns: A string containing the formatted 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"Documentation pipeline '{pipeline_name}' is not deployed (current status: {pipeline.status})."
+
+        # Execute the search
+        search_response = await client.pipelines(workspace=workspace).search(pipeline_name=pipeline_name, query=query)
+
+        return doc_search_results_to_llm_readable_string(search_response)
+
+    except ResourceNotFoundError:
+        return f"There is no documentation pipeline named '{pipeline_name}' in workspace '{workspace}'."
+    except BadRequestError as e:
+        return f"Failed to search documentation using pipeline '{pipeline_name}': {e}"
+    except UnexpectedAPIError as e:
+        return f"Failed to search documentation using pipeline '{pipeline_name}': {e}"
+    except Exception as e:
+        return f"An unexpected error occurred while searching documentation with pipeline '{pipeline_name}': {str(e)}"
+
+
+def get_docs_config() -> tuple[str, str, str] | None:
+    """Get docs search configuration from environment variables.
+
+    :returns: Tuple of (workspace, pipeline_name, api_key) if all are available, None otherwise.
+    """
+    workspace = os.environ.get("DEEPSET_DOCS_WORKSPACE")
+    pipeline_name = os.environ.get("DEEPSET_DOCS_PIPELINE_NAME")
+    api_key = os.environ.get("DEEPSET_DOCS_API_KEY")
+
+    if workspace and pipeline_name and api_key:
+        return workspace, pipeline_name, api_key
+
+    return None