Add DataToolService for dataset access and management

Introduces DataToolService to provide access to specific CSV datasets via new tools: data_provider and show_tables. Updates the factory and service registration to include the new service and adds the DATA domain to the Domain enum. This enables controlled access to allowed data files for the MCP server.

Author: Fr4nc3

src/backend/v3/mcp_server/core/factory.py

@@ -18,6 +18,7 @@ class Domain(Enum):
     TECH_SUPPORT = "tech_support"
     RETAIL = "retail"
     GENERAL = "general"
+    DATA = "data"
 
 
 class MCPToolBase(ABC):

src/backend/v3/mcp_server/mcp_server.py

@@ -11,7 +11,7 @@
 from typing import Optional
 
 from core.factory import MCPToolFactory
-from services import HRService, TechSupportService, GeneralService
+from services import HRService, TechSupportService, GeneralService, DataToolService
 from config.settings import config
 
 # Setup logging
@@ -26,6 +26,9 @@
 factory.register_service(TechSupportService())
 factory.register_service(GeneralService())
 
+# Register DataToolService with the dataset path
+factory.register_service(DataToolService(dataset_path="data/datasets"))
+
 
 def create_fastmcp_server():
     """Create and configure FastMCP server."""

src/backend/v3/mcp_server/services/__init__.py

@@ -5,5 +5,6 @@
 from .hr_service import HRService
 from .tech_support_service import TechSupportService
 from .general_service import GeneralService
+from .data_tool_service import DataToolService
 
-__all__ = ["HRService", "TechSupportService", "GeneralService"]
+__all__ = ["HRService", "TechSupportService", "GeneralService", "DataToolService"]

src/backend/v3/mcp_server/services/data_tool_service.py

@@ -0,0 +1,90 @@
+import os
+import logging
+from typing import List
+from ..core.factory import MCPToolBase, Domain
+
+ALLOWED_FILES = [
+    "competitor_Pricing_Analysis.csv",
+    "customer_Churn_Analysis.csv",
+    "customer_feedback_surveys.csv",
+    "customer_profile.csv",
+    "delivery_performance_metrics.csv",
+    "email_Marketing_Engagement.csv",
+    "loyalty_Program_Overview.csv",
+    "product_return_rates.csv",
+    "product_table.csv",
+    "purchase_history.csv",
+    "social_media_sentiment_analysis.csv",
+    "store_visit_history.csv",
+    "subscription_benefits_utilization.csv",
+    "unauthorized_Access_Attempts.csv",
+    "warehouse_Incident_Reports.csv",
+    "website_activity_log.csv",
+]
+
+
+class DataToolService(MCPToolBase):
+    def __init__(self, dataset_path: str):
+        super().__init__(Domain.DATA)
+        self.dataset_path = dataset_path
+        self.allowed_files = set(ALLOWED_FILES)
+
+    def _find_file(self, filename: str) -> str:
+        """
+        Searches recursively within the dataset_path for an exact filename match (case-sensitive).
+        Returns the full path if found, else None.
+        """
+        logger = logging.getLogger("find_file")
+        for root, _, files in os.walk(self.dataset_path):
+            if filename in files:
+                full_path = os.path.join(root, filename)
+                logger.info("Found file: %s", full_path)
+                return full_path
+        logger.warning(
+            "File '%s' not found in '%s' directory.", filename, self.dataset_path
+        )
+        return None
+
+    def register_tools(self, mcp):
+        @mcp.tool()
+        def data_provider(tablename: str) -> str:
+            """A tool that provides data from database based on given table name as parameter."""
+            logger = logging.getLogger("file_provider")
+            logger.info("Table '%s' requested.", tablename)
+            tablename = tablename.strip()
+            filename = (
+                f"{tablename}.csv"
+                if not tablename.lower().endswith(".csv")
+                else tablename
+            )
+            if filename not in self.allowed_files:
+                logger.error("File '%s' is not allowed.", filename)
+                return f"File '{filename}' is not allowed."
+            file_path = self._find_file(filename)
+            if not file_path:
+                logger.error("File '%s' not found.", filename)
+                return f"File '{filename}' not found."
+            try:
+                with open(file_path, "r", encoding="utf-8") as file:
+                    data = file.read()
+                return data
+            except IOError as e:
+                logger.error("Error reading file '%s': %s", filename, e)
+                return None
+
+        @mcp.tool()
+        def show_tables() -> List[str]:
+            """Returns a list of allowed table names (without .csv extension) that exist in the dataset path."""
+            logger = logging.getLogger("show_tables")
+            found_tables = []
+            for filename in self.allowed_files:
+                file_path = self._find_file(filename)
+                if file_path:
+                    table_name = filename[:-4]  # Remove .csv
+                    found_tables.append(table_name)
+                    logger.info("Found table: %s", table_name)
+            if not found_tables:
+                logger.warning(
+                    "No allowed CSV tables found in '%s' directory.", self.dataset_path
+                )
+            return found_tables