feat: Add VaultSearchTool for consensus-based knowledge retrieval

lib/crewai-tools/pyproject.toml

@@ -156,5 +156,10 @@ exclude-newer = "3 days"
 requires = ["hatchling"]
 build-backend = "hatchling.build"
 
+[dependency-groups]
+dev = [
+    "responses>=0.26.0",
+]
+
 [tool.hatch.version]
 path = "src/crewai_tools/__init__.py"

lib/crewai-tools/src/crewai_tools/tools/vault_search_tool/vault_search_tool.py

@@ -0,0 +1,77 @@
+from typing import Any, Optional, Type
+import requests
+from crewai.tools import BaseTool
+from pydantic import BaseModel, Field
+
+class VaultSearchSchema(BaseModel):
+    """Input schema for VaultSearchTool."""
+    query: str = Field(..., description="The semantic search query to look up in the knowledge vault.")
+
+class VaultSearchTool(BaseTool):
+    """
+    A search tool for retrieving team-verified technical reports from a Knowledge Vault.
+
+    This tool performs semantic search against a centralized repository of peer-validated 
+    content, ensuring that agents prioritize historical team consensus over general 
+    internet search results.
+    """
+    name: str = "Knowledge Vault Search"
+    description: str = (
+        "Useful for retrieving verified technical reports and team consensus. "
+        "Use this to find historical context and peer-validated research before "
+        "conducting new external searches."
+    )
+    args_schema: Type[BaseModel] = VaultSearchSchema
+
+    api_url: str = Field(
+        default="http://localhost:8000", 
+        description="The base URL of the Knowledge Vault API server."
+    )
+
+    def _run(self, query: str) -> str:
+        """
+        Execute a POST request to the vault server and format the top result.
+
+        Args:
+            query (str): The search string provided by the agent.
+
+        Returns:
+            str: A formatted string containing the top search hit, community scores, 
+                 and a content summary, or an error/empty message.
+        """
+        try:
+            # Send search request to the vault server
+            # The server expects a JSON payload with 'query', 'threshold', and 'match_count'
+            response = requests.post(
+                f"{self.api_url.rstrip('/')}/vault/search",
+                json={
+                    "query": query, 
+                    "threshold": 0.7, 
+                    "match_count": 5
+                },
+                timeout=10
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            # Extract results based on the standard server response format: {"hit": bool, "results": list}
+            results = data.get("results", [])
+
+            if not results:
+                return "No verified reports found in the knowledge vault for this query."
+
+            top = results[0]
+
+            # Format the output for the CrewAI Agent to ingest
+            return (
+                f"--- [Top Verified Report Found] ---\n"
+                f"Subject: {top.get('query_text', 'N/A')}\n"
+                f"Community Score: 👍{top.get('upvote_count', 0)} Upvotes | 👎{top.get('downvote_count', 0)} Downvotes\n"
+                f"Content Summary:\n{top.get('response_content', '')[:1500]}\n"
+                f"--- End of Report ---"
+            )
+
+        except requests.exceptions.RequestException as e:
+            return f"Error connecting to Knowledge Vault: {str(e)}"
+        except Exception as e:
+            return f"An unexpected error occurred while searching the vault: {str(e)}"

lib/crewai-tools/tests/tools/test_vault_search_tool.py

@@ -0,0 +1,89 @@
+import pytest
+import responses
+from crewai_tools.tools.vault_search_tool.vault_search_tool import VaultSearchTool
+
+class TestVaultSearchTool:
+    """
+    Unit tests for the VaultSearchTool class.
+
+    These tests verify the tool's ability to handle successful API responses,
+    empty results, server errors, and proper schema definition using 
+    the 'responses' library to mock HTTP traffic.
+    """
+
+    @responses.activate
+    def test_vault_search_successful_hit(self):
+        """
+        Verify the tool correctly parses and formats a successful search result.
+        """
+        # Register a mock rule for a successful POST request
+        responses.add(
+            responses.POST,
+            "http://localhost:8000/vault/search",
+            json={
+                "hit": True,
+                "results": [
+                    {
+                        "query_text": "How to configure VPC?",
+                        "upvote_count": 15,
+                        "downvote_count": 1,
+                        "response_content": "Detailed VPC steps..."
+                    }
+                ]
+            },
+            status=200
+        )
+
+        # Initialize the tool and execute search
+        tool = VaultSearchTool(api_url="http://localhost:8000")
+        result = tool._run(query="VPC config")
+
+        # Assertions to verify the formatted Markdown output
+        assert "--- [Top Verified Report Found] ---" in result
+        assert "Subject: How to configure VPC?" in result
+        assert "👍15 Upvotes" in result
+        assert "Detailed VPC steps..." in result
+
+    @responses.activate
+    def test_vault_search_no_results(self):
+        """
+        Verify the tool's behavior when the vault returns no matching reports.
+        """
+        responses.add(
+            responses.POST,
+            "http://localhost:8000/vault/search",
+            json={"hit": False, "results": []},
+            status=200
+        )
+
+        tool = VaultSearchTool(api_url="http://localhost:8000")
+        result = tool._run(query="Unknown topic")
+
+        assert result == "No verified reports found in the knowledge vault for this query."
+
+    @responses.activate
+    def test_vault_search_api_error(self):
+        """
+        Verify the tool gracefully handles HTTP error codes (e.g., 500 Internal Server Error).
+        """
+        responses.add(
+            responses.POST,
+            "http://localhost:8000/vault/search",
+            status=500
+        )
+
+        tool = VaultSearchTool(api_url="http://localhost:8000")
+        result = tool._run(query="Broken API")
+
+        # The tool should return a descriptive error message instead of crashing
+        assert "Error connecting to Knowledge Vault" in result
+        assert "500" in result
+
+    def test_tool_schema(self):
+        """
+        Verify that the tool inherits correctly from BaseTool and defines its schema properly.
+        """
+        tool = VaultSearchTool()
+        assert tool.name == "Knowledge Vault Search"
+        # Ensure 'query' is a required field in the Pydantic schema
+        assert "query" in tool.args_schema.model_fields