From 2a65efb046ae661d18e423a0e8334560ca9f8bbb Mon Sep 17 00:00:00 2001 From: Irina Rueegg Date: Wed, 6 May 2026 19:01:12 +0200 Subject: [PATCH] mcp servers with config and adapter files --- backend/agent/mcp_server_adapter.py | 112 ++++++++++++++++ backend/agent/mcp_server_config.json | 17 +++ .../servers/mcp_server_code_execution.py | 121 ++++++++++++++++++ .../agent/servers/mcp_server_file_search.py | 99 ++++++++++++++ .../agent/servers/mcp_server_web_search.py | 115 +++++++++++++++++ requirements.txt | 2 + 6 files changed, 466 insertions(+) create mode 100644 backend/agent/mcp_server_adapter.py create mode 100644 backend/agent/mcp_server_config.json create mode 100644 backend/agent/servers/mcp_server_code_execution.py create mode 100644 backend/agent/servers/mcp_server_file_search.py create mode 100644 backend/agent/servers/mcp_server_web_search.py diff --git a/backend/agent/mcp_server_adapter.py b/backend/agent/mcp_server_adapter.py new file mode 100644 index 0000000..c3c779b --- /dev/null +++ b/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}") diff --git a/backend/agent/mcp_server_config.json b/backend/agent/mcp_server_config.json new file mode 100644 index 0000000..087e5cb --- /dev/null +++ b/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"] +} + +} \ No newline at end of file diff --git a/backend/agent/servers/mcp_server_code_execution.py b/backend/agent/servers/mcp_server_code_execution.py new file mode 100644 index 0000000..7f610b7 --- /dev/null +++ b/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") diff --git a/backend/agent/servers/mcp_server_file_search.py b/backend/agent/servers/mcp_server_file_search.py new file mode 100644 index 0000000..55d3231 --- /dev/null +++ b/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") \ No newline at end of file diff --git a/backend/agent/servers/mcp_server_web_search.py b/backend/agent/servers/mcp_server_web_search.py new file mode 100644 index 0000000..8b92731 --- /dev/null +++ b/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") diff --git a/requirements.txt b/requirements.txt index ada1cab..850c7a9 100644 --- a/requirements.txt +++ b/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