mcp servers with config and adapter files

backend/agent/mcp_server_adapter.py

@@ -0,0 +1,112 @@
+import  asyncio
+import json
+import os
+import numpy as np
+from typing import List, Dict, Any
+from pathlib import Path
+
+from sentence_transformers import SentenceTransformer # embedder
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+class MCPToolRAGAdapter:
+    def __init__ (self, config_path: str = "mcp_server_config.json"):
+        self.config_path = config_path
+        self.tools = []
+        self.embedder = SentenceTransformer('all-MiniLM-L6-v2') # for embedding tool descriptions
+        self.sessions = {}
+        self.exit_stack = {}
+        self.tool_registry = {}
+        self.tool_embeddings = None
+    
+    def _load_config(self) -> Dict[str, Any]:
+        config_path = Path(__file__).parent / self.config_path
+        if not config_path.exists():
+            return {}
+        
+        try:
+            with open(self.config_path, 'r') as f:
+             return json.load(f)
+        except json.JSONDecodeError as e:
+            print(f"Error decoding JSON config: {e}")
+            return {}
+    
+    async def initialize_all_sessions(self):
+        """Initialize all MCP sessions defined in the config file and index their tools."""
+        config = self._load_config()
+        for server_name, params in config.items():
+            print(f"initializing session for {server_name} with params: {params}")
+            server_params = StdioServerParameters(
+                commanf=params["command"],
+                args=params.get("args", []),
+                # env=params.get("env", {}),
+            )
+            
+            # Verbindung aufbauen (Kontext-Manager manuell handhaben für Langzeit-Sessions)
+            transport_gen = stdio_client(server_params)
+            read, write = await transport_gen.__aenter__()
+            session = ClientSession(read, write)
+            await session.__aenter__()
+            await session.initialize()
+
+            self.sessions[server_name] = session
+            self.exit_stack[server_name] = (transport_gen, session)    # Zum späteren sauberen Schließen speichern
+            print(f"Session for {server_name} initialized successfully.")
+
+            # call tools and index thme
+            result = await session.list_tools()
+            tools = result.get("tools", [])
+
+            for tool in tools:
+                self.tool_registry.append({
+                    "server": server_name,
+                    "tool_name": tool["name"],
+                    "definition": tool,
+                    "search_text": f"{tool['name']}: {tool.get('description', '')}",
+                })
+            
+        # embeddings for all tools in this session
+        if self.tool_registry:
+            texts = [t["search_text"] for t in self.tool_registry]
+            self.tool_embeddings = self.embedder.encode(texts)
+            print(f"Indexing completed. {len(texts)} tools ready.")
+
+    def get_relevant_tools(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]:
+        """Given a user query, return the most relevant tools based on semantic similarity."""
+        if not self.tool_embeddings or not self.tool_registry:
+            print("No tools indexed yet.")
+            return []
+        
+        query_embedding = self.embedder.encode([query])
+        similarities = np.dot(self.tool_embeddings, query_embedding.T).flatten()
+        top_indices = np.argsort(similarities)[-top_k:][::-1]
+
+        relevant_tools = [self.tool_registry[i] for i in top_indices]
+
+        return relevant_tools
+    
+    async def call_tool(self, tool_name: str, arguments: Dict):
+        """ Finds the right server for the tool and calls it with the provided arguments. """
+        for item in self.tool_registry:
+            if item["definition"].name == tool_name:
+                server_name = item["server"]
+                session = self.sessions.get(server_name)
+                if session:
+                    try:
+                        result = await session.call_tool(tool_name, arguments)
+                        return result
+                    except Exception as e:
+                        print(f"Error calling tool {tool_name} on server {server_name}: {e}")
+                        return f"Error calling tool: {e}"
+        
+        return f"Tool '{tool_name}' not found in registry."
+    
+    async def shutdown_all_sessions(self):
+        """Gracefully shutdown all MCP sessions."""
+        for server_name, (transport_gen, session) in self.exit_stack.items():
+            try:
+                await session.__aexit__(None, None, None)
+                await transport_gen.__aexit__(None, None, None)
+                print(f"Session for {server_name} shut down successfully.")
+            except Exception as e:
+                print(f"Error shutting down session for {server_name}: {e}")

backend/agent/mcp_server_config.json

@@ -0,0 +1,17 @@
+{"Filemanager": {
+    "command": "python",
+    "args": ["servers/mcp_server_file_search.py"]
+},
+"WebSearch": {
+    "command": "python",
+    "args": ["servers/mcp_server_web_search.py"],
+    "env": {
+        "DDGS_API_KEY": "your_ddgs_api_key_here"
+    }
+},
+"CodeExecution": {
+    "command": "python",
+    "args": ["servers/mcp_server_code_execution.py"]
+}
+
+}

backend/agent/servers/mcp_server_code_execution.py

@@ -0,0 +1,121 @@
+import ast
+import subprocess
+from mcp.server.fastmcp import FastMCP
+
+# ── Configuration ────────────────────────────────────────────────────────────
+EXEC_TIMEOUT = 10        # seconds before killing the subprocess
+MAX_OUTPUT_LENGTH = 3000  # max characters of stdout+stderr to return
+
+mcp = FastMCP("SafeExecServer")
+
+BLOCKED_IMPORTS = {
+#    Filesystem access: 
+    "os", "pathlib", "shutil", "glob", "tempfile", "fileinput",
+#   Process execution: 
+    "subprocess", "multiprocessing", "threading",
+#   Network access: 
+    "socket", "http", "urllib", "requests", "ftplib", "smtplib","xmlrpc", "asyncio",
+#   System internals: 
+    "sys", "ctypes", "importlib", "code", "codeop", "compileall",
+#   Serialization exploits: 
+    "pickle", "shelve", "marshal",
+#   Other dangerous: 
+    "signal", "resource", "pty", "fcntl", "termios", "webbrowser", "antigravity"
+}
+
+BLOCKED_BUILTINS = {
+#   Code execution: 
+    "exec", "eval", "compile", "__import__",
+#   File access:
+    "open",
+#   Process control:
+    "exit", "quit", "breakpoint",
+#   Attribute manipulation:
+    "getattr", "setattr", "delattr",
+#   Introspection escapes:
+    "globals", "locals", "vars", "memoryview", "type"
+}
+
+
+def check_code_safety(code: str) -> str | None:
+    """
+    Statically analyze Python code for forbidden imports and builtins with ast.
+
+
+    Parameters
+    ----------
+    code : str
+        The Python code to check.
+
+    Returns
+    -------
+    str or None
+        Error message if forbidden code found, None if safe.
+    """
+        
+    try:
+        tree = ast.parse(code)
+    except SyntaxError as e:
+        return f"SyntaxError: {e}"
+    
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                if alias.name.split('.')[0] in BLOCKED_IMPORTS:
+                    return f"Error: Import of '{alias.name}' is not allowed."
+
+        elif isinstance(node, ast.ImportFrom):
+            if node.module and node.module.split('.')[0] in BLOCKED_IMPORTS:
+                return f"Error: Import from '{node.module}' is not allowed."
+
+        elif isinstance(node, ast.Call):
+            if isinstance(node.func, ast.Name) and node.func.id in BLOCKED_BUILTINS:
+                return f"Error: Use of builtin '{node.func.id}' is not allowed."
+        
+    return None  # No violations found
+
+@mcp.tool()
+def run_python_sandboxed(code: str) -> str:
+    """
+    Run Python code in a sandboxed subprocess.
+
+    Defense layers:
+      1. Static analysis (check_code_safety)
+      2. Subprocess isolation (child process)
+      3. Timeout (killed after EXEC_TIMEOUT seconds)
+      4. Output truncation (max MAX_OUTPUT_LENGTH chars)
+
+    Parameters
+    ----------
+    code : str
+        The Python code to execute.
+
+    Returns
+    -------
+    str
+        Combined stdout+stderr, or an error message.
+    """
+
+    static_safety = check_code_safety(code)
+    if static_safety is not None:
+        return static_safety
+    
+    try:
+        result = subprocess.run(
+            ["python3", "-c", code],
+            capture_output=True, text=True, timeout=EXEC_TIMEOUT)
+        output = result.stdout + result.stderr
+        if len(output) > MAX_OUTPUT_LENGTH:
+            output = output[:MAX_OUTPUT_LENGTH] + "\n...[output truncated]..."
+        return output if output.strip() else "Code executed successfully (no output)."
+    except subprocess.TimeoutExpired:
+        return f"Error: Code execution exceeded time limit of {EXEC_TIMEOUT} seconds and was terminated."
+    except Exception as e:
+        return f"Error during code execution: {e}"
+    
+
+
+# ── Run the server ───────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    mcp.run(transport="stdio")

backend/agent/servers/mcp_server_file_search.py

@@ -0,0 +1,99 @@
+from pathlib import Path
+from mcp.server.fastmcp import FastMCP
+
+# ── Configuration ────────────────────────────────────────────────────────────
+project_dir = Path(__file__).resolve().parent.parent
+ALLOWED_DIR = project_dir / "workspace"
+
+mcp = FastMCP("FileSearchServer")
+
+
+# ── Helper: path validation ──────────────────────────────────────────────────
+def _safe_path(requested: str) -> Path:
+    """Resolve and validate a path is inside ALLOWED_DIR."""
+    resolved = (ALLOWED_DIR / requested).resolve()
+    if not str(resolved).startswith(str(ALLOWED_DIR)):
+        raise ValueError(
+            f"Access denied: '{requested}' resolves outside "
+            f"the allowed directory '{ALLOWED_DIR}'"
+        )
+    return resolved
+
+
+# ── MCP Tools ────────────────────────────────────────────────────────────────
+
+@mcp.tool()
+def list_files() -> str:
+    """List all files in the project directory (recursively).
+
+    Returns a newline-separated list of relative file paths.
+    """
+    files = sorted(
+        f.relative_to(ALLOWED_DIR)
+        for f in ALLOWED_DIR.rglob("*")
+        if f.is_file() and "__pycache__" not in f.parts
+    )
+    if not files:
+        return "No files found in the project directory."
+    return "\n".join(str(f) for f in files)
+
+
+@mcp.tool()
+def read_file(path: str) -> str:
+    """Read the contents of a file.
+
+    Args:
+        path: Relative path to the file within the project directory.
+    """
+    try:
+        resolved = _safe_path(path)
+    except ValueError as e:
+        return f"Error: {e}"
+
+    if not resolved.exists():
+        return f"Error: File '{path}' does not exist."
+    if not resolved.is_file():
+        return f"Error: '{path}' is not a file."
+
+    try:
+        return resolved.read_text(encoding="utf-8")
+    except UnicodeDecodeError:
+        return f"Error: '{path}' is not a text file (binary content)."
+
+
+@mcp.tool()
+def search_files(query: str) -> str:
+    """Search for files whose name or content contains the query string.
+
+    Args:
+        query: The search term (case-insensitive).
+    """
+    query_lower = query.lower()
+    results = []
+
+    for f in sorted(ALLOWED_DIR.rglob("*")):
+        if not f.is_file() or "__pycache__" in f.parts:
+            continue
+        rel = f.relative_to(ALLOWED_DIR)
+
+        if query_lower in str(rel).lower():
+            results.append(f"[name match] {rel}")
+
+        try:
+            content = f.read_text(encoding="utf-8")
+            for i, line in enumerate(content.splitlines(), 1):
+                if query_lower in line.lower():
+                    snippet = line.strip()[:100]
+                    results.append(f"[content] {rel}:{i} -- {snippet}")
+        except (UnicodeDecodeError, PermissionError):
+            pass
+
+    if not results:
+        return f"No matches found for '{query}'."
+    return "\n".join(results[:30])  # limit to 30 matches
+
+
+# ── Run the server ───────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    mcp.run(transport="stdio")

backend/agent/servers/mcp_server_web_search.py

@@ -0,0 +1,115 @@
+from urllib.parse import urlparse
+from mcp.server.fastmcp import FastMCP
+
+# ── Configuration ────────────────────────────────────────────────────────────
+MAX_PAGE_LENGTH = 4000   # max characters to return from a fetched page
+REQUEST_TIMEOUT = 10     # seconds
+
+mcp = FastMCP("WebSearchServer")
+
+
+# ── Helper: URL validation (SSRF prevention) ─────────────────────────────────
+
+def _validate_url(url: str) -> str:
+    """Validate a URL to prevent SSRF attacks."""
+    parsed = urlparse(url)
+
+    if parsed.scheme not in ("http", "https"):
+        raise ValueError(
+            f"Blocked scheme '{parsed.scheme}'. Only http and https are allowed."
+        )
+
+    hostname = parsed.hostname or ""
+
+    blocked_hosts = {"localhost", "127.0.0.1", "0.0.0.0", "169.254.169.254"}
+    if hostname in blocked_hosts:
+        raise ValueError(f"Blocked internal host: {hostname}")
+
+    private_prefixes = (
+        "10.", "172.16.", "172.17.", "172.18.", "172.19.",
+        "172.20.", "172.21.", "172.22.", "172.23.", "172.24.",
+        "172.25.", "172.26.", "172.27.", "172.28.", "172.29.",
+        "172.30.", "172.31.", "192.168.",
+    )
+    for prefix in private_prefixes:
+        if hostname.startswith(prefix):
+            raise ValueError(f"Blocked private IP range: {hostname}")
+
+    return url
+
+
+# ── MCP Tools ────────────────────────────────────────────────────────────────
+
+@mcp.tool()
+def web_search(query: str, max_results: int = 5) -> str:
+    """Search the web using DuckDuckGo.
+
+    Args:
+        query: The search query.
+        max_results: Maximum number of results to return (default 5).
+    """
+    try:
+        from ddgs import DDGS
+        results = DDGS().text(query, max_results=max_results)
+
+        if not results:
+            return f"No results found for: {query}"
+
+        formatted = []
+        for r in results:
+            formatted.append(
+                f"Title: {r['title']}\n"
+                f"URL: {r['href']}\n"
+                f"Snippet: {r['body']}"
+            )
+        return "\n---\n".join(formatted)
+
+    except Exception as e:
+        return f"Search error: {e}"
+
+
+@mcp.tool()
+def fetch_page(url: str) -> str:
+    """Fetch a web page and extract its text content.
+
+    Args:
+        url: The URL to fetch.
+    """
+    try:
+        url = _validate_url(url)
+    except ValueError as e:
+        return f"URL blocked: {e}"
+
+    try:
+        import requests
+        from bs4 import BeautifulSoup
+
+        response = requests.get(
+            url,
+            timeout=REQUEST_TIMEOUT,
+            headers={"User-Agent": "Mozilla/5.0 (Lightweight Web Search MCP Server)"},
+        )
+
+        if response.status_code != 200:
+            return f"HTTP error {response.status_code} fetching {url}"
+
+        soup = BeautifulSoup(response.text, "html.parser")
+
+        for tag in soup(["script", "style", "nav", "footer"]):
+            tag.decompose()
+
+        text = soup.get_text(separator="\n", strip=True)
+
+        if len(text) > MAX_PAGE_LENGTH:
+            text = text[:MAX_PAGE_LENGTH] + "\n\n[... truncated ...]"
+
+        return text if text else "Page fetched but no text content found."
+
+    except Exception as e:
+        return f"Error fetching page: {e}"
+
+
+# ── Run the server ───────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    mcp.run(transport="stdio")

requirements.txt

@@ -3,6 +3,8 @@ streamlit>=1.28.0
 
 # AI/LLM Integration
 openai>=1.0.0
+mcp>=0.1.0
+ddgs>=0.1.0
 
 # Web & API
 requests>=2.31.0