Update the backend

backend/README.md

@@ -5,10 +5,7 @@ This folder contains the production-ready FastAPI stack plus the companion MCP s
 ## Directory Overview
 
 - `api/` – FastAPI application (routes, services, storage helpers, MCP clients)
-- `mcp_servers/` – Stand-alone MCP servers:
-  - `rag_server.py` / `main.py` – pgvector-backed retrieval over tenant documents
-  - `web_server.py` – DuckDuckGo-powered search with English bias
-  - `admin_server.py` – Governance utilities (regex rules, violation logging, tenant registry)
+- `mcp_server/` – Unified MCP server exposing rag/web/admin tools via namespaces
 - `workers/` – Celery workers and schedulers for async ingestion + analytics maintenance
 
 ## Prerequisites
@@ -32,17 +29,14 @@ cp env.example .env   # update MCP URLs + LLM settings
    uvicorn backend.api.main:app --port 8000 --reload
    ```
 
-2. **MCP servers** (each in its own shell):
+2. **Unified MCP server (rag/web/admin)**
    ```bash
-   # RAG / knowledge base
-   python -m backend.mcp_servers.main        # or python -m backend.mcp_servers.rag_server
-
-   # Web search
-   python -m backend.mcp_servers.web_server
-
-   # Admin / governance
-   python -m backend.mcp_servers.admin_server
+   python backend/mcp_server/server.py
    ```
+   This single endpoint exposes the following namespaced tools:
+   - `rag.search`, `rag.ingest`, `rag.delete`
+   - `web.search`
+   - `admin.getRules`, `admin.addRule`, `admin.deleteRule`, `admin.logViolation`
 
 3. **Optional workers** (if running Celery-based ingestion/analytics jobs):
    ```bash
@@ -76,7 +70,7 @@ Use the helper scripts in the repo root when validating backend changes:
 - `python check_rag_database.py` – Talks directly to the pgvector database to list tenant IDs, preview stored chunks, and run safeguarded searches via `search_vectors()`. Helpful when troubleshooting suspected cross-tenant leakage.
 - `python test_manual.py` – Legacy manual smoke test harness (analytics store, admin rules, API surface).
 
-> **Troubleshooting tip:** If the isolation script reports a failure, first run `check_rag_database.py` to confirm documents are tagged with the correct `tenant_id`, then restart the RAG MCP server so it reloads the updated SQL filtering logic.
+> **Troubleshooting tip:** If the isolation script reports a failure, first run `check_rag_database.py` to confirm documents are tagged with the correct `tenant_id`, then restart the unified MCP server so it reloads the updated SQL filtering logic.
 
 ## Environment Variables (excerpt)
 
@@ -89,3 +83,20 @@ Defined in `env.example`:
 
 Update these before starting the servers to ensure the agent can reach every MCP endpoint and LLM runtime.
 
+## Unified MCP tool instructions
+
+Agents that speak the Model Context Protocol should connect to the `integrachat` server id defined in `backend/mcp_server/server.py` and call the namespaced tools directly:
+
+| Namespace | Tool | Purpose |
+| --- | --- | --- |
+| `rag` | `search` | Retrieve tenant-scoped document chunks |
+| `rag` | `ingest` | Chunk + store new knowledge |
+| `rag` | `delete` | Remove one/all stored documents |
+| `web` | `search` | DuckDuckGo English-biased search |
+| `admin` | `getRules` | Fetch tenant governance rules (list or detailed) |
+| `admin` | `addRule` | Insert or update a rule |
+| `admin` | `deleteRule` | Remove a rule by text |
+| `admin` | `logViolation` | Persist a red-flag event into analytics |
+
+Always send `tenant_id`, and optionally `user_id`, in the payload so the shared middleware can enforce isolation and log analytics.
+

backend/api/mcp_clients/rag_client.py

@@ -44,6 +44,7 @@ class RAGClient:
     async def ingest(self, content: str, tenant_id: str):
         """
         Sends content to the RAG server for ingestion.
+        Returns the unwrapped data from the MCP server response.
         """
 
         try:
@@ -60,6 +61,17 @@ class RAGClient:
                 return {"error": f"HTTP {response.status_code}"}
 
             data = response.json()
+            
+            # MCP server wraps response in a 'data' field
+            # Extract the actual result data
+            if isinstance(data, dict) and "data" in data:
+                result = data["data"]
+                # Map chunks_ingested to chunks_stored for consistency
+                if "chunks_ingested" in result:
+                    result["chunks_stored"] = result.pop("chunks_ingested")
+                return result
+            
+            # If not wrapped, return as-is (backward compatibility)
             return data
 
         except Exception as e:
@@ -69,6 +81,7 @@ class RAGClient:
     async def list_documents(self, tenant_id: str, limit: int = 1000, offset: int = 0):
         """
         List all documents for a tenant.
+        Returns the unwrapped data from the MCP server response.
         """
 
         try:
@@ -86,6 +99,13 @@ class RAGClient:
                 return {"documents": [], "total": 0, "limit": limit, "offset": offset}
 
             data = response.json()
+            
+            # MCP server wraps response in a 'data' field
+            # Extract the actual result data
+            if isinstance(data, dict) and "data" in data:
+                return data["data"]
+            
+            # If not wrapped, return as-is (backward compatibility)
             return data
 
         except Exception as e:
@@ -95,6 +115,7 @@ class RAGClient:
     async def delete_document(self, tenant_id: str, document_id: int):
         """
         Delete a specific document by ID for a tenant.
+        Returns the unwrapped data from the MCP server response.
         """
         try:
             async with httpx.AsyncClient(timeout=30.0) as client:
@@ -115,6 +136,18 @@ class RAGClient:
                 return {"error": f"HTTP {response.status_code}: {error_text}"}
 
             data = response.json()
+            
+            # Check if MCP server returned an error response
+            if isinstance(data, dict) and data.get("status") == "error":
+                error_msg = data.get("message", "Unknown error")
+                return {"error": error_msg}
+            
+            # MCP server wraps response in a 'data' field
+            # Extract the actual result data
+            if isinstance(data, dict) and "data" in data:
+                return data["data"]
+            
+            # If not wrapped, return as-is (backward compatibility)
             return data
 
         except httpx.ConnectError as e:
@@ -127,6 +160,7 @@ class RAGClient:
     async def delete_all_documents(self, tenant_id: str):
         """
         Delete all documents for a tenant.
+        Returns the unwrapped data from the MCP server response.
         """
         try:
             async with httpx.AsyncClient(timeout=30.0) as client:
@@ -145,6 +179,18 @@ class RAGClient:
                 return {"error": f"HTTP {response.status_code}: {error_text}"}
 
             data = response.json()
+            
+            # Check if MCP server returned an error response
+            if isinstance(data, dict) and data.get("status") == "error":
+                error_msg = data.get("message", "Unknown error")
+                return {"error": error_msg}
+            
+            # MCP server wraps response in a 'data' field
+            # Extract the actual result data
+            if isinstance(data, dict) and "data" in data:
+                return data["data"]
+            
+            # If not wrapped, return as-is (backward compatibility)
             return data
 
         except httpx.ConnectError as e:

backend/api/routes/rag.py

@@ -264,7 +264,12 @@ async def rag_delete_all(
     try:
         result = await rag_client.delete_all_documents(x_tenant_id)
         if "error" in result:
-            raise HTTPException(status_code=500, detail=result["error"])
+            error_msg = result["error"]
+            # Check if it's a connection error (503) or other error
+            if "Cannot connect" in error_msg:
+                raise HTTPException(status_code=503, detail=error_msg)
+            else:
+                raise HTTPException(status_code=500, detail=error_msg)
         return result
     except HTTPException:
         raise

backend/mcp_server/__init__.py

@@ -0,0 +1,6 @@
+"""
+Unified MCP server package for IntegraChat.
+"""
+
+__all__ = ["common", "rag", "web", "admin"]
+

backend/mcp_server/admin/__init__.py

@@ -0,0 +1,6 @@
+"""
+Admin governance tooling namespace.
+"""
+
+__all__ = ["rules", "violations"]
+

backend/mcp_server/admin/rules.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from typing import Mapping
+
+from backend.api.storage.rules_store import RulesStore
+from backend.mcp_server.common.tenant import TenantContext
+from backend.mcp_server.common.utils import ToolValidationError, tool_handler
+
+_rules_store = RulesStore()
+
+
+@tool_handler("admin.getRules")
+async def admin_get_rules(context: TenantContext, payload: Mapping[str, object]) -> dict[str, object]:
+    """
+    Return the active admin rules for the tenant.
+    """
+
+    detailed = bool(payload.get("detailed", False))
+    rules = (
+        _rules_store.get_rules_detailed(context.tenant_id)
+        if detailed
+        else _rules_store.get_rules(context.tenant_id)
+    )
+    return {"tenant_id": context.tenant_id, "rules": rules, "detailed": detailed}
+
+
+@tool_handler("admin.addRule")
+async def admin_add_rule(context: TenantContext, payload: Mapping[str, object]) -> dict[str, object]:
+    """
+    Add a new governance rule.
+    """
+
+    rule_text = payload.get("rule")
+    if not isinstance(rule_text, str) or not rule_text.strip():
+        raise ToolValidationError("rule must be a non-empty string")
+
+    pattern = payload.get("pattern")
+    if pattern is not None and not isinstance(pattern, str):
+        raise ToolValidationError("pattern must be a string if provided")
+
+    severity = payload.get("severity", "medium")
+    if not isinstance(severity, str):
+        raise ToolValidationError("severity must be a string")
+
+    description = payload.get("description")
+    if description is not None and not isinstance(description, str):
+        raise ToolValidationError("description must be a string if provided")
+
+    enabled = bool(payload.get("enabled", True))
+
+    success = _rules_store.add_rule(
+        tenant_id=context.tenant_id,
+        rule=rule_text.strip(),
+        pattern=pattern.strip() if isinstance(pattern, str) and pattern.strip() else None,
+        severity=severity.strip(),
+        description=description.strip() if isinstance(description, str) and description.strip() else None,
+        enabled=enabled,
+    )
+
+    if not success:
+        raise ToolValidationError("rule already exists or could not be saved")
+
+    return {
+        "tenant_id": context.tenant_id,
+        "rule": rule_text.strip(),
+        "enabled": enabled,
+    }
+
+
+@tool_handler("admin.deleteRule")
+async def admin_delete_rule(context: TenantContext, payload: Mapping[str, object]) -> dict[str, object]:
+    """
+    Delete an existing rule by its text value.
+    """
+
+    rule_text = payload.get("rule")
+    if not isinstance(rule_text, str) or not rule_text.strip():
+        raise ToolValidationError("rule must be provided for deletion")
+
+    deleted = _rules_store.delete_rule(context.tenant_id, rule_text.strip())
+    return {
+        "tenant_id": context.tenant_id,
+        "rule": rule_text.strip(),
+        "deleted": deleted,
+    }
+

backend/mcp_server/admin/violations.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from typing import Mapping, Optional
+
+from backend.mcp_server.common.logging import log_redflag_violation
+from backend.mcp_server.common.tenant import TenantContext
+from backend.mcp_server.common.utils import ToolValidationError, tool_handler
+
+
+@tool_handler("admin.logViolation")
+async def log_violation(context: TenantContext, payload: Mapping[str, object]) -> dict[str, object]:
+    """
+    Persist a red-flag violation for analytics and auditing.
+    """
+
+    rule_id = payload.get("rule_id") or payload.get("ruleId")
+    if not isinstance(rule_id, str) or not rule_id.strip():
+        raise ToolValidationError("rule_id must be provided")
+
+    rule_pattern = payload.get("rule_pattern") or payload.get("rulePattern") or rule_id
+    if not isinstance(rule_pattern, str):
+        raise ToolValidationError("rule_pattern must be a string")
+
+    severity = payload.get("severity", "medium")
+    if not isinstance(severity, str):
+        raise ToolValidationError("severity must be a string")
+
+    matched_text = payload.get("matched_text") or payload.get("matchedText")
+    if not isinstance(matched_text, str) or not matched_text.strip():
+        raise ToolValidationError("matched_text is required")
+
+    confidence = payload.get("confidence")
+    if confidence is not None:
+        try:
+            confidence_value: Optional[float] = float(confidence)
+        except (TypeError, ValueError):
+            raise ToolValidationError("confidence must be numeric")
+    else:
+        confidence_value = None
+
+    message_preview = payload.get("message_preview") or payload.get("messagePreview")
+    if message_preview is not None and not isinstance(message_preview, str):
+        raise ToolValidationError("message_preview must be a string if provided")
+
+    log_redflag_violation(
+        tenant_id=context.tenant_id,
+        rule_id=rule_id.strip(),
+        rule_pattern=rule_pattern.strip(),
+        severity=severity.strip(),
+        matched_text=matched_text.strip(),
+        confidence=confidence_value,
+        message_preview=message_preview.strip() if isinstance(message_preview, str) else None,
+        user_id=context.user_id,
+    )
+
+    return {
+        "tenant_id": context.tenant_id,
+        "rule_id": rule_id.strip(),
+        "severity": severity.strip(),
+        "logged": True,
+    }
+

backend/mcp_server/common/__init__.py

@@ -0,0 +1,8 @@
+"""
+Shared utilities for the unified MCP server.
+"""
+
+from .tenant import TenantContext, TenantValidationError, build_tenant_context
+
+__all__ = ["TenantContext", "TenantValidationError", "build_tenant_context"]
+

backend/{mcp_servers → mcp_server/common}/database.py

@@ -1,17 +1,20 @@
 """
-Supabase database connection and utilities for MCP servers.
+Supabase/PostgreSQL database utilities shared by all MCP tools.
 
-This module provides both:
+This module provides:
 1. Direct PostgreSQL connections (via psycopg2) for pgvector operations
-2. Supabase client for REST API operations
+2. A Supabase client for REST-style administrative needs
 """
 
+from __future__ import annotations
+
 import os
 from typing import Optional, List, Dict, Any
+
 import psycopg2
 import psycopg2.extras
-from supabase import create_client, Client
 from dotenv import load_dotenv
+from supabase import Client, create_client
 
 # Load environment variables
 load_dotenv()
@@ -20,9 +23,9 @@ load_dotenv()
 # Environment variables
 # -----------------------------------
 
-DATABASE_URL = os.getenv("POSTGRESQL_URL")   # Direct PostgreSQL connection
+DATABASE_URL = os.getenv("POSTGRESQL_URL")  # Direct PostgreSQL connection
 SUPABASE_URL = os.getenv("SUPABASE_URL")
-SUPABASE_KEY = os.getenv("SUPABASE_SERVICE_KEY")   # MUST be service role key
+SUPABASE_KEY = os.getenv("SUPABASE_SERVICE_KEY")  # MUST be service role key
 
 # Global Supabase client instance
 _supabase_client: Optional[Client] = None
@@ -64,7 +67,8 @@ def initialize_database():
         print("✅ pgvector extension enabled")
 
         # Create documents table
-        cur.execute("""
+        cur.execute(
+            """
             CREATE TABLE IF NOT EXISTS documents (
                 id BIGSERIAL PRIMARY KEY,
                 tenant_id TEXT NOT NULL,
@@ -72,23 +76,28 @@ def initialize_database():
                 embedding vector(384) NOT NULL,
                 created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
             );
-        """)
+        """
+        )
         print("✅ documents table created")
 
         # Create index for vector similarity search
-        cur.execute("""
+        cur.execute(
+            """
             CREATE INDEX IF NOT EXISTS documents_embedding_idx 
             ON documents 
             USING ivfflat (embedding vector_cosine_ops)
             WITH (lists = 100);
-        """)
+        """
+        )
         print("✅ vector index created")
 
         # Create index for tenant_id for faster filtering
-        cur.execute("""
+        cur.execute(
+            """
             CREATE INDEX IF NOT EXISTS documents_tenant_id_idx 
             ON documents (tenant_id);
-        """)
+        """
+        )
         print("✅ tenant_id index created")
 
         conn.commit()
@@ -112,6 +121,9 @@ def insert_document_chunks(tenant_id: str, text: str, embedding: list):
     Insert document chunk + embedding.
     """
     try:
+        # Normalize tenant_id to ensure consistency
+        tenant_id = tenant_id.strip()
+        
         conn = get_connection()
         cur = conn.cursor()
 
@@ -120,7 +132,7 @@ def insert_document_chunks(tenant_id: str, text: str, embedding: list):
             INSERT INTO documents (tenant_id, chunk_text, embedding)
             VALUES (%s, %s, %s);
             """,
-            (tenant_id, text, embedding)
+            (tenant_id, text, embedding),
         )
 
         conn.commit()
@@ -142,7 +154,7 @@ def search_vectors(tenant_id: str, vector: list, limit: int = 5) -> List[Dict[st
         if not tenant_id or not tenant_id.strip():
             print("DB SEARCH ERROR: tenant_id is empty")
             return []
-        
+
         tenant_id = tenant_id.strip()
         conn = get_connection()
         cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
@@ -159,7 +171,7 @@ def search_vectors(tenant_id: str, vector: list, limit: int = 5) -> List[Dict[st
             ORDER BY embedding <=> %s::vector(384)
             LIMIT %s;
             """,
-            (vector, tenant_id, vector, limit)
+            (vector, tenant_id, vector, limit),
         )
 
         rows = cur.fetchall()
@@ -169,60 +181,88 @@ def search_vectors(tenant_id: str, vector: list, limit: int = 5) -> List[Dict[st
         for row in rows:
             row_tenant_id = row.get("tenant_id", "")
             if row_tenant_id != tenant_id:
-                print(f"WARNING: Found document with tenant_id '{row_tenant_id}' when searching for '{tenant_id}' - skipping")
+                print(
+                    f"WARNING: Found document with tenant_id '{row_tenant_id}' when searching for '{tenant_id}' - skipping"
+                )
                 continue
-            
+
             results.append(
                 {
                     "text": row["chunk_text"],
                     "similarity": float(row.get("similarity", 0.0)),
                 }
             )
-        
+
         cur.close()
         conn.close()
-        
+
         return results
 
     except Exception as e:
         print(f"DB SEARCH ERROR (tenant_id={tenant_id}): {e}")
         import traceback
+
         traceback.print_exc()
         return []
 
 
-def list_all_documents(tenant_id: str, limit: int = 1000, offset: int = 0) -> Dict[str, Any]:
+def list_all_documents(
+    tenant_id: str, limit: int = 1000, offset: int = 0
+) -> Dict[str, Any]:
     """
     List all documents for a tenant with pagination.
+    Handles tenant_id normalization to match documents stored with different formatting.
     """
     try:
+        # Normalize tenant_id to ensure consistency
+        tenant_id_normalized = tenant_id.strip()
+        
         conn = get_connection()
         cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
 
+        # Get all unique tenant_ids that match when normalized
+        cur.execute("SELECT DISTINCT tenant_id FROM documents;")
+        all_tenant_ids = [row[0] for row in cur.fetchall()]
+        
+        # Find tenant_ids that match when normalized
+        matching_tenant_ids = []
+        for stored_tenant_id in all_tenant_ids:
+            if stored_tenant_id and stored_tenant_id.strip() == tenant_id_normalized:
+                matching_tenant_ids.append(stored_tenant_id)
+        
+        if not matching_tenant_ids:
+            # No matching tenant_ids found
+            cur.close()
+            conn.close()
+            return {"documents": [], "total": 0, "limit": limit, "offset": offset}
+        
+        # Build query to match any of the normalized tenant_ids
+        placeholders = ','.join(['%s'] * len(matching_tenant_ids))
         cur.execute(
-            """
+            f"""
             SELECT
                 id,
                 chunk_text,
                 created_at
             FROM documents
-            WHERE tenant_id = %s
+            WHERE tenant_id IN ({placeholders})
             ORDER BY created_at DESC
             LIMIT %s OFFSET %s;
             """,
-            (tenant_id, limit, offset)
+            tuple(matching_tenant_ids) + (limit, offset),
         )
 
         rows = cur.fetchall()
 
-        # Get total count
+        # Get total count for all matching tenant_ids
+        placeholders = ','.join(['%s'] * len(matching_tenant_ids))
         cur.execute(
-            """
+            f"""
             SELECT COUNT(*) as total
             FROM documents
-            WHERE tenant_id = %s;
+            WHERE tenant_id IN ({placeholders});
             """,
-            (tenant_id,)
+            tuple(matching_tenant_ids),
         )
         total_row = cur.fetchone()
         total = total_row["total"] if total_row else 0
@@ -236,10 +276,17 @@ def list_all_documents(tenant_id: str, limit: int = 1000, offset: int = 0) -> Di
                 {
                     "id": row["id"],
                     "text": row["chunk_text"],
-                    "created_at": row["created_at"].isoformat() if row["created_at"] else None,
+                    "created_at": row["created_at"].isoformat()
+                    if row["created_at"]
+                    else None,
                 }
             )
-        return {"documents": results, "total": total, "limit": limit, "offset": offset}
+        return {
+            "documents": results,
+            "total": total,
+            "limit": limit,
+            "offset": offset,
+        }
 
     except Exception as e:
         print("DB LIST ERROR:", e)
@@ -252,19 +299,56 @@ def delete_document(tenant_id: str, document_id: int) -> bool:
     Returns True if document was deleted, False otherwise.
     """
     try:
+        # Normalize tenant_id to ensure consistency
+        tenant_id = tenant_id.strip()
+        
         conn = get_connection()
         cur = conn.cursor()
 
-        # Delete the document (tenant_id check ensures tenant isolation)
+        # First, verify the document exists
         cur.execute(
             """
-            DELETE FROM documents
-            WHERE id = %s AND tenant_id = %s;
+            SELECT id, tenant_id FROM documents
+            WHERE id = %s;
             """,
-            (document_id, tenant_id)
+            (document_id,),
         )
-
-        deleted = cur.rowcount > 0
+        doc_row = cur.fetchone()
+        
+        if doc_row is None:
+            print(f"DB DELETE: Document {document_id} does not exist")
+            cur.close()
+            conn.close()
+            return False
+        
+        doc_tenant_id = doc_row[1] if len(doc_row) > 1 else None
+        # Normalize both tenant_ids for comparison (handle existing data with whitespace)
+        doc_tenant_id_normalized = doc_tenant_id.strip() if doc_tenant_id else None
+        tenant_id_normalized = tenant_id.strip()
+        
+        # Try to delete with normalized comparison - if normalized match, use stored value for actual delete
+        if doc_tenant_id_normalized == tenant_id_normalized:
+            # Tenant IDs match after normalization - proceed with delete using stored tenant_id
+            cur.execute(
+                """
+                DELETE FROM documents
+                WHERE id = %s AND tenant_id = %s;
+                """,
+                (document_id, doc_tenant_id),
+            )
+            deleted = cur.rowcount > 0
+        else:
+            # Tenant IDs don't match - log the mismatch
+            print(f"DB DELETE: Document {document_id} belongs to tenant '{doc_tenant_id}' (normalized: '{doc_tenant_id_normalized}'), not '{tenant_id}' (normalized: '{tenant_id_normalized}')")
+            print(f"DB DELETE: Tenant ID lengths - stored: {len(doc_tenant_id) if doc_tenant_id else 0}, requested: {len(tenant_id)}")
+            print(f"DB DELETE: Tenant ID repr - stored: {repr(doc_tenant_id)}, requested: {repr(tenant_id)}")
+            deleted = False
+        
+        if deleted:
+            print(f"DB DELETE: Successfully deleted document {document_id} for tenant '{tenant_id}'")
+        else:
+            print(f"DB DELETE: Failed to delete document {document_id} for tenant '{tenant_id}' (rowcount: {cur.rowcount})")
+        
         conn.commit()
         cur.close()
         conn.close()
@@ -272,7 +356,9 @@ def delete_document(tenant_id: str, document_id: int) -> bool:
         return deleted
 
     except Exception as e:
-        print("DB DELETE ERROR:", e)
+        print(f"DB DELETE ERROR (document_id={document_id}, tenant_id={tenant_id}): {e}")
+        import traceback
+        traceback.print_exc()
         return False
 
 
@@ -280,20 +366,50 @@ def delete_all_documents(tenant_id: str) -> int:
     """
     Delete all documents for a tenant.
     Returns the number of documents deleted.
+    Handles tenant_id normalization to match documents stored with different formatting.
     """
     try:
+        # Normalize tenant_id
+        tenant_id = tenant_id.strip()
+        
         conn = get_connection()
         cur = conn.cursor()
 
+        # First, get all unique tenant_ids that match when normalized
         cur.execute(
             """
-            DELETE FROM documents
-            WHERE tenant_id = %s;
-            """,
-            (tenant_id,)
+            SELECT DISTINCT tenant_id FROM documents;
+            """
         )
-
-        deleted_count = cur.rowcount
+        all_tenant_ids = [row[0] for row in cur.fetchall()]
+        
+        # Find tenant_ids that match when normalized
+        matching_tenant_ids = []
+        tenant_id_normalized = tenant_id.strip()
+        for stored_tenant_id in all_tenant_ids:
+            if stored_tenant_id and stored_tenant_id.strip() == tenant_id_normalized:
+                matching_tenant_ids.append(stored_tenant_id)
+        
+        if not matching_tenant_ids:
+            print(f"DB DELETE ALL: No documents found for tenant '{tenant_id}' (normalized: '{tenant_id_normalized}')")
+            cur.close()
+            conn.close()
+            return 0
+        
+        # Delete documents matching any of the normalized tenant_ids
+        deleted_count = 0
+        for matching_tenant_id in matching_tenant_ids:
+            cur.execute(
+                """
+                DELETE FROM documents
+                WHERE tenant_id = %s;
+                """,
+                (matching_tenant_id,),
+            )
+            deleted_count += cur.rowcount
+        
+        print(f"DB DELETE ALL: Deleted {deleted_count} document(s) for tenant '{tenant_id}' (matched {len(matching_tenant_ids)} tenant_id variant(s))")
+        
         conn.commit()
         cur.close()
         conn.close()
@@ -301,7 +417,9 @@ def delete_all_documents(tenant_id: str) -> int:
         return deleted_count
 
     except Exception as e:
-        print("DB DELETE ALL ERROR:", e)
+        print(f"DB DELETE ALL ERROR (tenant_id={tenant_id}): {e}")
+        import traceback
+        traceback.print_exc()
         return 0
 
 
@@ -341,3 +459,4 @@ TABLES = {
     "analytics": "analytics_events",
     "tool_usage": "tool_usage_stats",
 }
+

backend/{mcp_servers → mcp_server/common}/embeddings.py

@@ -1,8 +1,13 @@
+"""
+Sentence-transformer embeddings shared across all MCP tools.
+"""
+
 from sentence_transformers import SentenceTransformer
 
 # Load MiniLM model (384-dimensional embeddings)
 model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
 
+
 def embed_text(text: str):
     """
     Generate sentence embedding for use with pgvector.
@@ -15,3 +20,4 @@ def embed_text(text: str):
     """
     vector = model.encode(text)
     return vector.tolist()
+

backend/mcp_server/common/logging.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Dict, Optional
+
+logger = logging.getLogger("integrachat.mcp")
+if not logger.handlers:
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter(
+        "[%(asctime)s] %(levelname)s %(name)s - %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+
+logger.setLevel(os.getenv("LOG_LEVEL", "INFO").upper())
+
+try:
+    from backend.api.storage.analytics_store import AnalyticsStore
+except Exception:  # pragma: no cover - analytics storage is optional during tests
+    AnalyticsStore = None  # type: ignore
+    _analytics_store = None
+else:
+    _analytics_store = AnalyticsStore()
+
+
+def log_tool_usage(
+    tool_name: str,
+    tenant_id: Optional[str],
+    *,
+    success: bool,
+    latency_ms: Optional[int] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+    error_message: Optional[str] = None,
+    user_id: Optional[str] = None,
+):
+    log_data = {
+        "tool": tool_name,
+        "tenant_id": tenant_id,
+        "success": success,
+        "latency_ms": latency_ms,
+        "user_id": user_id,
+        "metadata": metadata or {},
+    }
+    if error_message:
+        log_data["error"] = error_message
+
+    if success:
+        logger.info("tool_completed %s", log_data)
+    else:
+        logger.warning("tool_failed %s", log_data)
+
+    if _analytics_store and tenant_id:
+        try:
+            _analytics_store.log_tool_usage(
+                tenant_id=tenant_id,
+                tool_name=tool_name,
+                latency_ms=latency_ms,
+                success=success,
+                error_message=error_message,
+                metadata=metadata,
+                user_id=user_id,
+            )
+        except Exception as exc:  # pragma: no cover - analytics failures shouldn't crash tools
+            logger.debug("analytics logging failed: %s", exc)
+
+
+def log_rag_search_metrics(
+    tenant_id: str,
+    query: str,
+    hits_count: int,
+    avg_score: Optional[float],
+    top_score: Optional[float],
+    latency_ms: Optional[int] = None,
+):
+    if _analytics_store:
+        try:
+            _analytics_store.log_rag_search(
+                tenant_id=tenant_id,
+                query=query,
+                hits_count=hits_count,
+                avg_score=avg_score,
+                top_score=top_score,
+                latency_ms=latency_ms,
+            )
+        except Exception as exc:  # pragma: no cover
+            logger.debug("rag analytics logging failed: %s", exc)
+
+
+def log_redflag_violation(
+    tenant_id: str,
+    rule_id: str,
+    rule_pattern: str,
+    severity: str,
+    matched_text: str,
+    *,
+    confidence: Optional[float] = None,
+    message_preview: Optional[str] = None,
+    user_id: Optional[str] = None,
+):
+    if _analytics_store:
+        try:
+            _analytics_store.log_redflag_violation(
+                tenant_id=tenant_id,
+                rule_id=rule_id,
+                rule_pattern=rule_pattern,
+                severity=severity,
+                matched_text=matched_text,
+                confidence=confidence,
+                message_preview=message_preview,
+                user_id=user_id,
+            )
+        except Exception as exc:  # pragma: no cover
+            logger.debug("redflag logging failed: %s", exc)
+

backend/mcp_server/common/tenant.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Any, Mapping, Optional
+
+
+class TenantValidationError(ValueError):
+    """Raised when tenant metadata is missing or malformed."""
+
+
+TENANT_ID_PATTERN = re.compile(r"^[A-Za-z0-9_\-.:/]{3,128}$")
+
+
+@dataclass(slots=True)
+class TenantContext:
+    tenant_id: str
+    user_id: Optional[str] = None
+    metadata: Optional[dict[str, Any]] = None
+
+
+def _extract_tenant_id(payload: Mapping[str, Any]) -> str:
+    for key in ("tenant_id", "tenantId", "tenant"):
+        if key in payload:
+            value = payload[key]
+            if isinstance(value, str):
+                return value.strip()
+    raise TenantValidationError("tenant_id is required for every MCP tool call")
+
+
+def _normalize_tenant_id(raw_value: str) -> str:
+    normalized = raw_value.strip()
+    if not normalized:
+        raise TenantValidationError("tenant_id cannot be empty")
+    if not TENANT_ID_PATTERN.match(normalized):
+        raise TenantValidationError(
+            "tenant_id must be 3-128 chars and may only contain letters, numbers, '.', '-', '_', or ':'"
+        )
+    return normalized
+
+
+def build_tenant_context(payload: Mapping[str, Any]) -> TenantContext:
+    tenant_id = _normalize_tenant_id(_extract_tenant_id(payload))
+    user_id: Optional[str] = None
+    metadata: Optional[dict[str, Any]] = None
+
+    for key in ("user_id", "userId"):
+        if key in payload and isinstance(payload[key], str):
+            user_id = payload[key].strip() or None
+            break
+
+    meta_candidate = payload.get("metadata")
+    if isinstance(meta_candidate, dict):
+        metadata = meta_candidate
+
+    return TenantContext(tenant_id=tenant_id, user_id=user_id, metadata=metadata)
+

backend/mcp_server/common/utils.py

@@ -0,0 +1,146 @@
+from __future__ import annotations
+import inspect
+import time
+from typing import Any, Awaitable, Callable, Mapping, Optional
+
+from .logging import log_tool_usage
+from .tenant import TenantContext, TenantValidationError, build_tenant_context
+
+
+class ToolValidationError(ValueError):
+    """Raised when the caller request payload is invalid."""
+
+
+class ToolExecutionError(RuntimeError):
+    """Raised for unexpected runtime failures."""
+
+
+Payload = Mapping[str, Any]
+ToolHandler = Callable[[TenantContext, Payload], Awaitable[dict[str, Any]] | dict[str, Any]]
+
+
+def success_response(
+    tool_name: str,
+    context: TenantContext,
+    data: Any,
+    latency_ms: int,
+    metadata: Optional[dict[str, Any]] = None,
+) -> dict[str, Any]:
+    return {
+        "status": "ok",
+        "tool": tool_name,
+        "tenant_id": context.tenant_id,
+        "latency_ms": latency_ms,
+        "metadata": metadata or {},
+        "data": data,
+    }
+
+
+def error_response(
+    tool_name: str,
+    context: Optional[TenantContext],
+    error: Exception,
+    latency_ms: int,
+    error_type: str = "runtime_error",
+) -> dict[str, Any]:
+    return {
+        "status": "error",
+        "tool": tool_name,
+        "tenant_id": context.tenant_id if context else None,
+        "latency_ms": latency_ms,
+        "error_type": error_type,
+        "message": str(error),
+    }
+
+
+async def maybe_await(result: Any) -> Any:
+    if inspect.isawaitable(result):
+        return await result
+    return result
+
+
+def _truncate(value: Any, max_length: int = 200) -> Any:
+    if isinstance(value, str) and len(value) > max_length:
+        return value[: max_length - 3] + "..."
+    return value
+
+
+def _trim_payload(payload: Payload) -> dict[str, Any]:
+    trimmed: dict[str, Any] = {}
+    for key, value in payload.items():
+        if key in {"content", "query"} and isinstance(value, str):
+            trimmed[key] = _truncate(value)
+        elif isinstance(value, (str, int, float, bool)) or value is None:
+            trimmed[key] = value
+        else:
+            trimmed[key] = "<complex>"
+    return trimmed
+
+
+async def execute_tool(
+    tool_name: str,
+    payload: Payload,
+    handler: ToolHandler,
+) -> dict[str, Any]:
+    start = time.perf_counter()
+    context: Optional[TenantContext] = None
+    try:
+        context = build_tenant_context(payload)
+        result = await maybe_await(handler(context, payload))
+        latency_ms = int((time.perf_counter() - start) * 1000)
+
+        log_tool_usage(
+            tool_name,
+            context.tenant_id,
+            success=True,
+            latency_ms=latency_ms,
+            metadata={"payload": _trim_payload(payload)},
+            user_id=context.user_id,
+        )
+        return success_response(
+            tool_name,
+            context,
+            result,
+            latency_ms,
+        )
+    except (TenantValidationError, ToolValidationError) as exc:
+        latency_ms = int((time.perf_counter() - start) * 1000)
+        log_tool_usage(
+            tool_name,
+            context.tenant_id if context else None,
+            success=False,
+            latency_ms=latency_ms,
+            error_message=str(exc),
+            metadata={"payload": _trim_payload(payload)},
+            user_id=context.user_id if context else None,
+        )
+        return error_response(tool_name, context, exc, latency_ms, "validation_error")
+    except Exception as exc:  # pragma: no cover - safety net
+        latency_ms = int((time.perf_counter() - start) * 1000)
+        log_tool_usage(
+            tool_name,
+            context.tenant_id if context else None,
+            success=False,
+            latency_ms=latency_ms,
+            error_message=str(exc),
+            metadata={"payload": _trim_payload(payload)},
+            user_id=context.user_id if context else None,
+        )
+        return error_response(tool_name, context, exc, latency_ms)
+
+
+def tool_handler(tool_name: str):
+    """
+    Decorator that wires tenant validation, analytics logging, and error handling.
+    """
+
+    def decorator(func: ToolHandler):
+        async def wrapper(payload: Payload) -> dict[str, Any]:
+            return await execute_tool(tool_name, payload, func)
+
+        wrapper.__name__ = func.__name__
+        wrapper.__doc__ = func.__doc__
+        return wrapper
+
+    return decorator
+

backend/mcp_server/rag/__init__.py

@@ -0,0 +1,6 @@
+"""
+Retriever tooling for the unified MCP server.
+"""
+
+__all__ = ["search", "ingest", "delete"]
+

backend/mcp_server/rag/delete.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from typing import Mapping
+
+from backend.mcp_server.common.database import delete_all_documents, delete_document
+from backend.mcp_server.common.tenant import TenantContext
+from backend.mcp_server.common.utils import ToolValidationError, tool_handler
+
+
+@tool_handler("rag.delete")
+async def rag_delete(context: TenantContext, payload: Mapping[str, object]) -> dict[str, object]:
+    """
+    Delete one document by ID or purge all documents for the tenant.
+    """
+
+    document_id = payload.get("document_id")
+    delete_all = bool(payload.get("delete_all", False))
+
+    if delete_all:
+        deleted = delete_all_documents(context.tenant_id)
+        return {
+            "tenant_id": context.tenant_id,
+            "deleted_count": deleted,
+            "mode": "all",
+        }
+
+    if document_id is None:
+        raise ToolValidationError("document_id is required unless delete_all=true")
+
+    try:
+        doc_id_value = int(document_id)
+    except (TypeError, ValueError):
+        raise ToolValidationError("document_id must be an integer")
+
+    if doc_id_value <= 0:
+        raise ToolValidationError("document_id must be positive")
+
+    deleted = delete_document(context.tenant_id, doc_id_value)
+    if not deleted:
+        raise ToolValidationError(f"Document {doc_id_value} not found or access denied")
+    
+    return {
+        "tenant_id": context.tenant_id,
+        "document_id": doc_id_value,
+        "deleted": True,
+        "mode": "single",
+    }
+

backend/mcp_server/rag/ingest.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from typing import Mapping
+
+from backend.api.utils.text_extractor import extract_text
+from backend.mcp_server.common.database import insert_document_chunks
+from backend.mcp_server.common.embeddings import embed_text
+from backend.mcp_server.common.tenant import TenantContext
+from backend.mcp_server.common.utils import ToolValidationError, tool_handler
+
+
+@tool_handler("rag.ingest")
+async def rag_ingest(context: TenantContext, payload: Mapping[str, object]) -> dict[str, object]:
+    """
+    Ingest raw text into the tenant's knowledge base.
+    """
+
+    content = payload.get("content")
+    if not isinstance(content, str) or not content.strip():
+        raise ToolValidationError("content must be a non-empty string")
+
+    max_words = payload.get("chunk_words", 300)
+    try:
+        max_words_value = max(50, min(int(max_words), 800))
+    except (TypeError, ValueError):
+        raise ToolValidationError("chunk_words must be an integer between 50 and 800")
+
+    chunks = extract_text(content, max_words=max_words_value)
+    if not chunks:
+        raise ToolValidationError("no text detected after preprocessing")
+
+    stored = 0
+    for chunk in chunks:
+        vector = embed_text(chunk)
+        insert_document_chunks(context.tenant_id, chunk, vector)
+        stored += 1
+
+    return {
+        "tenant_id": context.tenant_id,
+        "chunks_ingested": stored,
+        "metadata": {"chunk_words": max_words_value},
+    }
+

backend/mcp_server/rag/list.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from typing import Mapping
+
+from backend.mcp_server.common.database import list_all_documents
+from backend.mcp_server.common.tenant import TenantContext
+from backend.mcp_server.common.utils import ToolValidationError, tool_handler
+
+
+@tool_handler("rag.list")
+async def rag_list(context: TenantContext, payload: Mapping[str, object]) -> dict[str, object]:
+    """
+    List stored RAG documents for the tenant with pagination.
+    """
+
+    limit = payload.get("limit", 1000)
+    offset = payload.get("offset", 0)
+
+    try:
+        limit_value = max(1, min(int(limit), 5000))
+    except (TypeError, ValueError):
+        raise ToolValidationError("limit must be an integer between 1 and 5000")
+
+    try:
+        offset_value = max(0, int(offset))
+    except (TypeError, ValueError):
+        raise ToolValidationError("offset must be a non-negative integer")
+
+    return list_all_documents(context.tenant_id, limit=limit_value, offset=offset_value)
+

backend/mcp_server/rag/search.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from statistics import mean
+from typing import Mapping
+
+from backend.mcp_server.common.database import search_vectors
+from backend.mcp_server.common.embeddings import embed_text
+from backend.mcp_server.common.logging import log_rag_search_metrics
+from backend.mcp_server.common.tenant import TenantContext
+from backend.mcp_server.common.utils import ToolValidationError, tool_handler
+
+
+@tool_handler("rag.search")
+async def rag_search(context: TenantContext, payload: Mapping[str, Any]) -> dict[str, Any]:
+    """
+    Perform semantic search across the tenant's knowledge base.
+    """
+
+    query = payload.get("query")
+    if not isinstance(query, str) or not query.strip():
+        raise ToolValidationError("query must be a non-empty string")
+
+    limit = payload.get("limit", 10)
+    try:
+        limit_value = max(1, min(int(limit), 25))
+    except (TypeError, ValueError):
+        raise ToolValidationError("limit must be an integer between 1 and 25")
+
+    threshold = payload.get("threshold", 0.55)
+    try:
+        threshold_value = max(0.0, min(float(threshold), 1.0))
+    except (TypeError, ValueError):
+        raise ToolValidationError("threshold must be a float between 0.0 and 1.0")
+
+    embedding = embed_text(query)
+    raw_results = search_vectors(context.tenant_id, embedding, limit=limit_value)
+    filtered = [
+        {"text": chunk.get("text", ""), "relevance": chunk.get("similarity", 0.0)}
+        for chunk in raw_results
+        if chunk.get("similarity", 0.0) >= threshold_value
+    ][:3]
+
+    hits = len(raw_results)
+    avg_score = mean([item.get("similarity", 0.0) for item in raw_results]) if raw_results else None
+    top_score = raw_results[0].get("similarity") if raw_results else None
+
+    log_rag_search_metrics(
+        tenant_id=context.tenant_id,
+        query=query,
+        hits_count=hits,
+        avg_score=avg_score,
+        top_score=top_score,
+    )
+
+    return {
+        "query": query,
+        "results": filtered,
+        "metadata": {
+            "limit": limit_value,
+            "threshold": threshold_value,
+            "hits_before_filter": hits,
+        },
+    }
+

backend/mcp_server/server.py

@@ -0,0 +1,200 @@
+from __future__ import annotations
+
+import logging
+import os
+from contextlib import asynccontextmanager
+from typing import Awaitable, Callable, Dict, Optional
+
+from fastapi import FastAPI, Query
+import uvicorn
+
+from backend.mcp_server.admin.rules import admin_add_rule, admin_delete_rule, admin_get_rules
+from backend.mcp_server.admin.violations import log_violation as admin_log_violation
+from backend.mcp_server.rag.delete import rag_delete
+from backend.mcp_server.rag.ingest import rag_ingest
+from backend.mcp_server.rag.list import rag_list
+from backend.mcp_server.rag.search import rag_search
+from backend.mcp_server.web.search import web_search
+
+ToolHandler = Callable[[Dict], Awaitable[Dict] | Dict]
+
+logger = logging.getLogger("integrachat.mcp.server")
+if not logger.handlers:
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter(
+        "[%(asctime)s] %(levelname)s %(name)s - %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+logger.setLevel(logging.INFO)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Lifespan context manager for startup and shutdown events."""
+    # Startup
+    try:
+        routes = []
+        for route in app.routes:
+            if hasattr(route, "path") and hasattr(route, "methods"):
+                routes.append(f"{', '.join(route.methods)} {route.path}")
+        logger.info("Registered routes: %s", ", ".join(sorted(routes)))
+    except Exception as e:
+        logger.warning("Could not log routes during startup: %s", e)
+    yield
+    # Shutdown (if needed in the future)
+
+
+app = FastAPI(title="IntegraChat MCP", version="1.0.0", lifespan=lifespan)
+
+
+def _register_tool(tool_name: str, handler: ToolHandler) -> None:
+    """
+    Register the given tool handler under both a namespaced route
+    (/rag/search) and an optional root route (/search) so the server works
+    whether clients point to /rag or directly to the namespace port.
+    """
+    namespace, action = tool_name.split(".", 1)
+    namespaced_path = f"/{namespace}/{action}"
+    root_path = f"/{action}"
+
+    @app.post(namespaced_path)
+    async def namespaced_endpoint(payload: Dict) -> Dict:
+        return await handler(payload)  # type: ignore[arg-type]
+
+    @app.post(root_path)
+    async def root_endpoint(payload: Dict) -> Dict:
+        return await handler(payload)  # type: ignore[arg-type]
+
+
+# Add GET endpoint support for /rag/list (register BEFORE POST to avoid conflicts)
+@app.get("/rag/list")
+async def rag_list_get(
+    tenant_id: str = Query(..., description="Tenant ID"),
+    limit: Optional[int] = Query(1000, description="Maximum number of documents to return"),
+    offset: Optional[int] = Query(0, description="Number of documents to skip")
+) -> Dict:
+    """GET endpoint for listing RAG documents."""
+    logger.info("GET /rag/list called with tenant_id=%s, limit=%s, offset=%s", tenant_id, limit, offset)
+    payload = {
+        "tenant_id": tenant_id,
+        "limit": limit,
+        "offset": offset
+    }
+    result = await rag_list(payload)  # type: ignore[arg-type]
+    return result
+
+@app.get("/list")
+async def rag_list_get_root(
+    tenant_id: str = Query(..., description="Tenant ID"),
+    limit: Optional[int] = Query(1000, description="Maximum number of documents to return"),
+    offset: Optional[int] = Query(0, description="Number of documents to skip")
+) -> Dict:
+    """GET endpoint for listing RAG documents (root path)."""
+    logger.info("GET /list called with tenant_id=%s, limit=%s, offset=%s", tenant_id, limit, offset)
+    payload = {
+        "tenant_id": tenant_id,
+        "limit": limit,
+        "offset": offset
+    }
+    result = await rag_list(payload)  # type: ignore[arg-type]
+    return result
+
+# Add DELETE endpoint support for /rag/delete/{document_id}
+@app.delete("/rag/delete/{document_id}")
+async def rag_delete_document(
+    document_id: int,
+    tenant_id: str = Query(..., description="Tenant ID")
+) -> Dict:
+    """DELETE endpoint for deleting a specific document."""
+    try:
+        logger.info("DELETE /rag/delete/%s called with tenant_id=%s", document_id, tenant_id)
+        payload = {
+            "tenant_id": tenant_id,
+            "document_id": document_id
+        }
+        result = await rag_delete(payload)  # type: ignore[arg-type]
+        logger.info("DELETE /rag/delete/%s result: %s", document_id, result)
+        return result
+    except Exception as e:
+        logger.error("Error in DELETE /rag/delete/%s: %s", document_id, e, exc_info=True)
+        raise
+
+@app.delete("/delete/{document_id}")
+async def rag_delete_document_root(
+    document_id: int,
+    tenant_id: str = Query(..., description="Tenant ID")
+) -> Dict:
+    """DELETE endpoint for deleting a specific document (root path)."""
+    logger.info("DELETE /delete/%s called with tenant_id=%s", document_id, tenant_id)
+    payload = {
+        "tenant_id": tenant_id,
+        "document_id": document_id
+    }
+    result = await rag_delete(payload)  # type: ignore[arg-type]
+    return result
+
+# Add DELETE endpoint support for /rag/delete-all
+@app.delete("/rag/delete-all")
+async def rag_delete_all(
+    tenant_id: str = Query(..., description="Tenant ID")
+) -> Dict:
+    """DELETE endpoint for deleting all documents."""
+    try:
+        logger.info("DELETE /rag/delete-all called with tenant_id=%s", tenant_id)
+        payload = {
+            "tenant_id": tenant_id,
+            "delete_all": True
+        }
+        result = await rag_delete(payload)  # type: ignore[arg-type]
+        return result
+    except Exception as e:
+        logger.error("Error in DELETE /rag/delete-all: %s", e, exc_info=True)
+        raise
+
+@app.delete("/delete-all")
+async def rag_delete_all_root(
+    tenant_id: str = Query(..., description="Tenant ID")
+) -> Dict:
+    """DELETE endpoint for deleting all documents (root path)."""
+    try:
+        logger.info("DELETE /delete-all called with tenant_id=%s", tenant_id)
+        payload = {
+            "tenant_id": tenant_id,
+            "delete_all": True
+        }
+        result = await rag_delete(payload)  # type: ignore[arg-type]
+        return result
+    except Exception as e:
+        logger.error("Error in DELETE /delete-all: %s", e, exc_info=True)
+        raise
+
+_register_tool("rag.search", rag_search)
+_register_tool("rag.ingest", rag_ingest)
+_register_tool("rag.delete", rag_delete)
+_register_tool("rag.list", rag_list)
+
+_register_tool("web.search", web_search)
+
+_register_tool("admin.getRules", admin_get_rules)
+_register_tool("admin.addRule", admin_add_rule)
+_register_tool("admin.deleteRule", admin_delete_rule)
+_register_tool("admin.logViolation", admin_log_violation)
+
+
+@app.get("/health")
+async def health() -> Dict[str, str]:
+    return {"status": "ok", "service": "mcp"}
+
+
+def main():
+    host = os.getenv("MCP_HOST", "0.0.0.0")
+    port = int(os.getenv("MCP_PORT", "8001"))
+    logger.info("Starting IntegraChat MCP HTTP server on %s:%s", host, port)
+    uvicorn.run("backend.mcp_server.server:app", host=host, port=port)
+
+
+if __name__ == "__main__":
+    main()
+

backend/mcp_server/web/__init__.py

@@ -0,0 +1,6 @@
+"""
+Web search tooling namespace.
+"""
+
+__all__ = ["search"]
+

backend/mcp_server/web/search.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from typing import Mapping
+
+from duckduckgo_search import DDGS
+
+from backend.mcp_server.common.tenant import TenantContext
+from backend.mcp_server.common.utils import ToolExecutionError, ToolValidationError, tool_handler
+
+
+@tool_handler("web.search")
+async def web_search(context: TenantContext, payload: Mapping[str, object]) -> dict[str, object]:
+    """
+    Perform a DuckDuckGo web search with an English-results bias.
+    """
+
+    query = payload.get("query")
+    if not isinstance(query, str) or not query.strip():
+        raise ToolValidationError("query must be a non-empty string")
+
+    max_results = payload.get("max_results", 5)
+    try:
+        max_results_value = max(1, min(int(max_results), 10))
+    except (TypeError, ValueError):
+        raise ToolValidationError("max_results must be an integer between 1 and 10")
+
+    region = str(payload.get("region", "us-en"))
+
+    try:
+        ddg = DDGS()
+        query_string = query
+        if "lang:en" not in query_string.lower():
+            query_string = f"{query_string} lang:en"
+
+        try:
+            results = ddg.text(query_string, max_results=max_results_value, region=region)
+        except TypeError:
+            results = ddg.text(query_string, max_results=max_results_value)
+
+        formatted = [
+            {
+                "title": item.get("title"),
+                "snippet": item.get("body"),
+                "url": item.get("href"),
+            }
+            for item in results
+        ]
+
+        return {
+            "query": query,
+            "results": formatted,
+            "metadata": {"max_results": max_results_value, "region": region},
+        }
+    except Exception as exc:
+        raise ToolExecutionError(f"web search failed: {exc}") from exc
+

backend/mcp_servers/admin_server.py

@@ -1,51 +0,0 @@
-# =============================================================
-# File: backend/mcp_servers/admin_server.py
-# =============================================================
-
-from fastapi import FastAPI
-from fastapi.middleware.cors import CORSMiddleware
-import logging
-import sys
-import os
-
-# Fix Python module paths
-current_dir = os.path.dirname(__file__)
-sys.path.insert(0, current_dir)
-
-from models.admin import EvalRequest, AlertPayload
-
-
-admin_app = FastAPI(title="Admin MCP Server")
-
-# Enable CORS
-admin_app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-log = logging.getLogger("admin_mcp")
-logging.basicConfig(level=logging.INFO)
-
-
-@admin_app.post("/eval")
-async def eval_query(req: EvalRequest):
-    danger = ["delete all data", "export users", "password", "token"]
-    q = req.query.lower()
-    for d in danger:
-        if d in q:
-            return {"action": "block", "reason": d}
-    return {"action": "allow"}
-
-
-@admin_app.post("/alert")
-async def alert(payload: AlertPayload):
-    log.warning(f"Alert received for tenant {payload.tenant_id}: {payload.violations}")
-    return {"status": "ok"}
-
-
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(admin_app, host="0.0.0.0", port=8003)

backend/mcp_servers/main.py

@@ -1,243 +0,0 @@
-from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel
-from fastapi.middleware.cors import CORSMiddleware
-from dotenv import load_dotenv
-import sys
-import os
-
-# --------------------------------------------------------
-# Fix Python module paths
-# --------------------------------------------------------
-
-current_dir = os.path.dirname(__file__)
-parent_dir = os.path.dirname(current_dir)
-
-sys.path.insert(0, current_dir)          # For embeddings + database
-sys.path.insert(0, os.path.join(parent_dir, "api"))  # For utils
-
-
-# --------------------------------------------------------
-# Imports AFTER adjusting paths
-# --------------------------------------------------------
-
-from embeddings import embed_text
-from database import insert_document_chunks, search_vectors, list_all_documents, initialize_database, delete_document, delete_all_documents
-from utils.text_extractor import extract_text
-
-
-# --------------------------------------------------------
-# Load environment variables
-# --------------------------------------------------------
-
-load_dotenv()
-
-
-# --------------------------------------------------------
-# FastAPI App
-# --------------------------------------------------------
-
-app = FastAPI(
-    title="RAG MCP Server",
-    description="Provides semantic search + ingestion for tenant knowledge bases",
-    version="1.0.0"
-)
-
-
-# --------------------------------------------------------
-# Enable CORS
-# --------------------------------------------------------
-
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-
-# --------------------------------------------------------
-# Startup Event - Initialize Database
-# --------------------------------------------------------
-
-@app.on_event("startup")
-async def startup_event():
-    """Initialize database schema on server startup."""
-    try:
-        print("Initializing database schema...")
-        initialize_database()
-    except Exception as e:
-        print(f"Warning: Database initialization failed: {e}")
-        print("Server will continue, but database operations may fail.")
-
-
-# --------------------------------------------------------
-# Request Models
-# --------------------------------------------------------
-
-class IngestPayload(BaseModel):
-    tenant_id: str
-    content: str
-
-
-class SearchPayload(BaseModel):
-    query: str
-    tenant_id: str
-
-
-# --------------------------------------------------------
-# Health Check
-# --------------------------------------------------------
-
-@app.get("/")
-def root():
-    return {"status": "RAG MCP SERVER RUNNING"}
-
-
-# --------------------------------------------------------
-# Ingest Route
-# --------------------------------------------------------
-
-@app.post("/ingest")
-def ingest(payload: IngestPayload):
-    """
-    Ingest raw text:
-    - Chunk text
-    - Embed chunks
-    - Store in Postgres
-    """
-    try:
-        chunks = extract_text(payload.content)
-
-        if not chunks:
-            raise HTTPException(400, "No text found to ingest.")
-
-        inserted = 0
-
-        for chunk in chunks:
-            embedding = embed_text(chunk)
-            insert_document_chunks(payload.tenant_id, chunk, embedding)
-            inserted += 1
-
-        return {
-            "status": "ok",
-            "tenant_id": payload.tenant_id,
-            "chunks_stored": inserted
-        }
-
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-# --------------------------------------------------------
-# Search Route
-# --------------------------------------------------------
-
-@app.post("/search")
-def search(payload: SearchPayload):
-    """
-    Semantic search using pgvector + MiniLM embeddings.
-    Results are filtered by tenant_id in the database query.
-    """
-    try:
-        # Validate tenant_id is provided
-        if not payload.tenant_id or not payload.tenant_id.strip():
-            raise HTTPException(status_code=400, detail="tenant_id is required")
-        
-        query_embedding = embed_text(payload.query)
-        # search_vectors filters by tenant_id in the SQL query
-        results = search_vectors(payload.tenant_id.strip(), query_embedding, limit=10)
-
-        # Log for debugging (remove in production)
-        print(f"[RAG Search] tenant_id={payload.tenant_id}, query={payload.query[:50]}, results_count={len(results)}")
-
-        return {
-            "tenant_id": payload.tenant_id,
-            "query": payload.query,
-            "results": results
-        }
-
-    except HTTPException:
-        raise
-    except Exception as e:
-        print(f"[RAG Search Error] tenant_id={payload.tenant_id}, error={str(e)}")
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-# --------------------------------------------------------
-# List All Documents Route
-# --------------------------------------------------------
-
-@app.get("/list")
-def list_documents(tenant_id: str, limit: int = 1000, offset: int = 0):
-    """
-    List all documents for a tenant with pagination.
-    """
-    try:
-        result = list_all_documents(tenant_id, limit=limit, offset=offset)
-        return result
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-# --------------------------------------------------------
-# Delete Document Route
-# --------------------------------------------------------
-
-@app.delete("/delete/{document_id}")
-def delete_doc(document_id: int, tenant_id: str):
-    """
-    Delete a specific document by ID for a tenant.
-    """
-    try:
-        deleted = delete_document(tenant_id, document_id)
-        if not deleted:
-            raise HTTPException(status_code=404, detail="Document not found or access denied")
-        
-        return {
-            "status": "ok",
-            "tenant_id": tenant_id,
-            "document_id": document_id,
-            "message": "Document deleted successfully"
-        }
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-@app.delete("/delete-all")
-def delete_all_docs(tenant_id: str):
-    """
-    Delete all documents for a tenant.
-    """
-    try:
-        deleted_count = delete_all_documents(tenant_id)
-        return {
-            "status": "ok",
-            "tenant_id": tenant_id,
-            "deleted_count": deleted_count,
-            "message": f"Deleted {deleted_count} document(s)"
-        }
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-# --------------------------------------------------------
-# Allow "python main.py" to start server
-# --------------------------------------------------------
-
-if __name__ == "__main__":
-    import uvicorn
-
-    print("Starting RAG MCP Server on http://0.0.0.0:8001")
-    print("API Documentation: http://localhost:8001/docs")
-    print("Note: Reload mode disabled when running directly")
-
-    # Run the app directly (reload doesn't work with app object)
-    uvicorn.run(
-        app,  # Pass the app object directly
-        host="0.0.0.0",
-        port=8001,
-        reload=False  # Reload requires module path, not app object
-    )

backend/mcp_servers/models/__init__.py

@@ -1,18 +0,0 @@
-"""
-MCP Server Models Package
-
-This package contains all Pydantic models used across MCP servers.
-"""
-
-from .admin import EvalRequest, AlertPayload
-from .rag import IngestRequest, SearchRequest
-from .web import WebSearchRequest
-
-__all__ = [
-    "EvalRequest",
-    "AlertPayload",
-    "IngestRequest",
-    "SearchRequest",
-    "WebSearchRequest",
-]
-

backend/mcp_servers/models/admin.py

@@ -1,14 +0,0 @@
-from pydantic import BaseModel
-from typing import Optional
-
-
-class EvalRequest(BaseModel):
-    tenant_id: str
-    query: str
-
-
-class AlertPayload(BaseModel):
-    tenant_id: str
-    violations: list
-    source: Optional[dict] = None
-

backend/mcp_servers/models/rag.py

@@ -1,12 +0,0 @@
-from pydantic import BaseModel
-
-
-class IngestRequest(BaseModel):
-    tenant_id: str
-    content: str
-
-
-class SearchRequest(BaseModel):
-    tenant_id: str
-    query: str
-

backend/mcp_servers/models/web.py

@@ -1,7 +0,0 @@
-from pydantic import BaseModel
-
-
-class WebSearchRequest(BaseModel):
-    tenant_id: str
-    query: str
-

backend/mcp_servers/placeholder.txt

@@ -1,4 +0,0 @@
-This directory contains the MCP server implementations.
-For the Hugging Face Space submission, only placeholder files are included.
-The full MCP server code exists separately.
-

backend/mcp_servers/rag_server.py

@@ -1,84 +0,0 @@
-# =============================================================
-# File: backend/mcp_servers/rag_server.py
-# =============================================================
-
-from fastapi import FastAPI
-from fastapi.middleware.cors import CORSMiddleware
-import sys
-import os
-
-# Fix Python module paths
-current_dir = os.path.dirname(__file__)
-sys.path.insert(0, current_dir)
-
-from typing import Any, Dict, List
-
-from embeddings import embed_text
-from database import insert_document_chunks, search_vectors
-from models.rag import IngestRequest, SearchRequest
-
-
-rag_app = FastAPI(title="RAG MCP Server")
-
-# Enable CORS
-rag_app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-
-# Wrapper functions to match expected interface
-def db_insert(tenant_id: str, content: str, vector: list):
-    """Wrapper for insert_document_chunks to match expected interface."""
-    return insert_document_chunks(tenant_id, content, vector)
-
-
-def db_search(tenant_id: str, vector: list, limit: int = 5):
-    """Wrapper for search_vectors to match expected interface."""
-    results = search_vectors(tenant_id, vector, limit)
-    # search_vectors returns list of dicts with "text" and "similarity"
-    # Preserve the structure and use similarity as relevance
-    return [
-        {
-            "text": result.get("text", ""),
-            "relevance": result.get("similarity", 0.0)
-        }
-        for result in results
-    ]
-
-
-@rag_app.post("/ingest")
-async def ingest(req: IngestRequest):
-    vector = embed_text(req.content)
-    db_insert(req.tenant_id, req.content, vector)
-    return {"status": "ok"}
-
-
-@rag_app.post("/search")
-async def search(req: SearchRequest):
-    """
-    Search documents for a specific tenant.
-    Results are already filtered by tenant_id in the database query.
-    """
-    vector = embed_text(req.query)
-    # db_search already filters by tenant_id and returns results sorted by similarity
-    results = db_search(req.tenant_id, vector, limit=10)  # Get more results for filtering
-    # Filter by relevance threshold and limit to top 3
-    filtered = [chunk for chunk in results if chunk.get("relevance", 0.0) >= 0.55][:3]
-    return {
-        "results": filtered,
-        "metadata": {
-            "total_retrieved": len(results),
-            "returned": len(filtered),
-            "threshold": 0.55
-        }
-    }
-
-
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(rag_app, host="0.0.0.0", port=8001)
-

backend/mcp_servers/web_server.py

@@ -1,71 +0,0 @@
-# =============================================================
-# File: backend/mcp_servers/web_server.py
-# =============================================================
-
-from fastapi import FastAPI
-from fastapi.middleware.cors import CORSMiddleware
-from duckduckgo_search import DDGS
-import sys
-import os
-
-# Fix Python module paths
-current_dir = os.path.dirname(__file__)
-sys.path.insert(0, current_dir)
-
-from models.web import WebSearchRequest
-
-web_app = FastAPI(title="Web Search MCP Server")
-
-# Enable CORS
-web_app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-
-@web_app.post("/search")
-async def web_search(req: WebSearchRequest):
-    """
-    Web search endpoint using DuckDuckGo.
-    Accepts tenant_id for multi-tenant support (currently not used but kept for API consistency).
-    Forces English language results by using region parameter and query modification.
-    """
-    try:
-        ddg = DDGS()
-        
-        # Modify query to prefer English results
-        # Add language hint to help get English content
-        query = req.query
-        # Only add language hint if not already present
-        if "lang:en" not in query.lower() and "site:en" not in query.lower():
-            query = f"{query} lang:en"
-        
-        # Try to use region parameter for English results
-        # Common region codes: 'us-en' for US English, 'uk-en' for UK English
-        try:
-            results = ddg.text(query, max_results=5, region='us-en')
-        except (TypeError, KeyError):
-            # If region parameter not supported, try without it
-            # The lang:en in query should still help
-            results = ddg.text(query, max_results=5)
-        
-        formatted = []
-        for r in results:
-            formatted.append({
-                "title": r.get("title"),
-                "snippet": r.get("body"),
-                "url": r.get("href"),
-            })
-
-        return {"results": formatted}
-
-    except Exception as e:
-        return {"error": str(e), "results": []}
-
-
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(web_app, host="0.0.0.0", port=8002)