From e75212652ecb0dc1fb038705836fbd6c9ce8b3eb Mon Sep 17 00:00:00 2001 From: Irina Rueegg Date: Thu, 21 May 2026 16:11:21 +0200 Subject: [PATCH] validate fix --- backend/agent/mcp_server_adapter.py | 8 +- backend/agent/mcp_server_adapter_RAG.py | 114 --------- .../servers/mcp_server_code_execution.py | 242 +----------------- requirements.txt | 3 + 4 files changed, 14 insertions(+), 353 deletions(-) delete mode 100644 backend/agent/mcp_server_adapter_RAG.py diff --git a/backend/agent/mcp_server_adapter.py b/backend/agent/mcp_server_adapter.py index 8df0885..c87b5ae 100644 --- a/backend/agent/mcp_server_adapter.py +++ b/backend/agent/mcp_server_adapter.py @@ -11,7 +11,6 @@ class MCPToolAdapter: def __init__(self, config_path: str = "mcp_server_config.json"): self.config_path = config_path self.servers: Dict[str, Dict] = {} - #self.exit_stack: Dict[str, Any] = {} self.tool_registry: List[Dict[str, Any]] = [] def _load_config(self) -> Dict[str, Any]: @@ -57,10 +56,8 @@ class MCPToolAdapter: await session.initialize() print(f"Session initialized for {server_name}. Requesting tools...") result = await session.list_tools() - print(f"Tools received from {server_name}: {result}") tools = result.tools - print(f"Tools received from {server_name}: {result}") - #tools = getattr(result, 'tools', []) + print(f"Tools received from {server_name}: {len(tools)} Tools") for tool in tools: t_params = tool.inputSchema.get("properties", {}) @@ -84,9 +81,6 @@ class MCPToolAdapter: }) print(f"Registered tool '{tool.name}' from {server_name}.") - - print(f"Session for {server_name} ready. {len(tools)} tools found.") - except Exception as e: print(f"Failed to initialize {server_name}: {e}") diff --git a/backend/agent/mcp_server_adapter_RAG.py b/backend/agent/mcp_server_adapter_RAG.py deleted file mode 100644 index 6583c9b..0000000 --- a/backend/agent/mcp_server_adapter_RAG.py +++ /dev/null @@ -1,114 +0,0 @@ -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.toolnames = [] - 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', '')}", - }) - self.tool_names.append(tool["name"]) - - # 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/servers/mcp_server_code_execution.py b/backend/agent/servers/mcp_server_code_execution.py index c613653..e2d69d2 100644 --- a/backend/agent/servers/mcp_server_code_execution.py +++ b/backend/agent/servers/mcp_server_code_execution.py @@ -1,34 +1,13 @@ import ast import subprocess -import tempfile import io -from pyflakes.api import check -from pyflakes.reporter import Reporter +from pyflakes.api import check # For linting Code +from pyflakes.reporter import Reporter # For linting Code from mcp.server.fastmcp import FastMCP from pathlib import Path -# ── Sandbox venv ──────────────────────────────────────────────────────────── -#SERVER_BASE_DIR = Path(__file__).parent.resolve() -#SANDBOX_DIR = SERVER_BASE_DIR / ".mcp_sandbox" -#WORKSPACE_DIR = SERVER_BASE_DIR.parent.parent.parent.parent / "workspace" - -#def get_sandbox_paths(): -# """Bestimmt die Executables innerhalb der Venv ohne os-Modul.""" -# if not SANDBOX_DIR.exists(): -# venv.create(SANDBOX_DIR, with_pip=True) -# -# bin_folder = "Scripts" if os.name == "nt" else "bin" # Einfacher Check für Windows -# -# exe_suffix = ".exe" if os.name == "nt" else "" -# python_exe = SANDBOX_DIR / bin_folder / f"python{exe_suffix}" -# pip_exe = SANDBOX_DIR / bin_folder / f"pip{exe_suffix}" -# -# return str(python_exe), str(pip_exe) -# -#PYTHON_EXE, PIP_EXE = get_sandbox_paths() - # ── Configuration ──────────────────────────────────────────────────────────── -EXEC_TIMEOUT = 45 # seconds before killing the subprocess +EXEC_TIMEOUT = 45 # seconds before killing the subprocess MAX_OUTPUT_LENGTH = 3000 # max characters of stdout+stderr to return # ── Create the MCP server ──────────────────────────────────────────────────── @@ -67,7 +46,7 @@ FORBIDDEN_SEQUENCES = ["../", "..\\", "/etc/", "/dev/", "C:\\Windows", "C:\\Program Files", "C:\\Users", "compile(", "__import__", "os.", "sys.", "subprocess."] -ALLOWED_PACKAGES = ["pygame", "numpy", "pandas"] +ALLOWED_PACKAGES = ["pygame", "numpy", "pandas"] # ── Static Analysis ──────────────────────────────────────────────────── def check_code_safety(code: str) -> str | None: @@ -189,7 +168,7 @@ def analyse_structure(code: str) -> str: @mcp.tool() -def lint_code(code: str) -> tuple[str, bool]: +def lint_code(code: str) -> str: """ Runs a fast static analysis check to catch syntax errors, unused imports, or undefined variables without executing the code. @@ -207,14 +186,14 @@ def lint_code(code: str) -> tuple[str, bool]: try: check(code, filename="", reporter=reporter) except Exception as e: - return (f"Critical error during linting: {str(e)}", False) + return f"Critical error during linting: {str(e)}" errors = error_buffer.getvalue().strip() warnings = warning_buffer.getvalue().strip() # Ergebnis-String zusammenbauen if not errors and not warnings: - return ("Linting complete: No issues found. The code is syntactically sound.", True) + return "Linting complete: No issues found. The code is syntactically sound." report = ["--- Linting Report ---"] @@ -228,155 +207,7 @@ def lint_code(code: str) -> tuple[str, bool]: report.append("\nAdvice: Please fix these issues before attempting to execute the code.") - return ("\n".join(report), False) - - -#@mcp.tool() -#def list_sandbox_packages() -> str: -# """ -# Lists all Python-Packages, that are installed in the Sandbox and their Version. -# Helpful to determine if packages like 'pygame', 'numpy' or similair are already available -# """ -# try: -# result = subprocess.run( -# [PIP_EXE, "list"], -# capture_output=True, -# text=True, -# timeout=10 -# ) -# -# if result.returncode != 0: -# return f"Error while listing the packages: {result.stderr}" -# -# if not result.stdout.strip(): -# return "The Sandbox environment is empty (only Standard-Libraries are available)." -# -# return f"Installed Packages: {result.stdout}" -# -# except Exception as e: -# return f"Error trying to list packages from the Sandbox venv: {str(e)}" -# -# -#@mcp.tool() -#def install_package_into_sandbox(package_name: str) -> str: -# """ -# Install a Python package into the sandbox environment using pip. -# -# Args: -# package_name: The name of the package to install (e.g., "requests"). -# -# Returns: -# A success message or an error message if installation fails. -# """ -# clean_name = "".join(e for e in package_name if e.isalnum() or e in "-_.") -# -# if clean_name in BLOCKED_IMPORTS: -# return f"Error: Installation of package '{clean_name}' is blocked due to security policies." -# -# if clean_name in BLOCKED_BUILTINS: -# return f"Error: Installation of package '{clean_name}' is blocked due to security policies." -# -# if not clean_name: -# return "Error: Invalid package name provided." -# -# try: -# result = subprocess.run( -# [PIP_EXE, "install", clean_name], -# capture_output=True, -# text=True, -# timeout=EXEC_TIMEOUT -# ) -# -# if result.returncode == 0: -# return f"Package '{clean_name}' installed successfully in the sandbox." -# else: -# return (f"Error installing package '{clean_name}':\n" -# f"{result.stdout}\n{result.stderr}") -# -# except subprocess.TimeoutExpired: -# return f"Error: Package installation exceeded time limit of {EXEC_TIMEOUT} seconds and was terminated." -# except Exception as e: -# return f"Error during package installation: {e}" -# -# -#@mcp.tool() -#def reset_sandbox() -> str: -# """Deletes the complete Sandbox and reopens it from Zero (Full Reset).""" -# global PYTHON_EXE, PIP_EXE -# -# try: -# if SANDBOX_DIR.exists(): -# shutil.rmtree(SANDBOX_DIR) -# -# PYTHON_EXE, PIP_EXE = get_sandbox_paths() -# -# return "Sandbox was reseet completely" -# -# except Exception as e: -# return f"Reset failed: {e}" - - -#@mcp.tool() -#def run_python_code_sandboxed(code: str) -> str: -# """ -# Runs Python code in a sandboxed environment. -# -# The sandbox blocks dangerous operations (filesystem, network, process -# control). Code is killed after 45 seconds. Use print() to produce -# output, which is captured and returned (up to 3000 chars). If the code -# is deemed unsafe by static analysis, it will not be executed and an error -# message will be returned instead. -# -# Args: -# code: The Python code or the File content of a python file to be executed in str format -# -# Returns: -# Combined stdout+stderr, or an error message in str format. -# """ -# -# static_safety = check_code_safety(code) -# if static_safety: -# return f"Code rejected:{static_safety}" -# -# run_id = datetime.now().strftime("%Y%m%d_%H%M%S") -# jail_dir = WORKSPACE_DIR / f"sandbox_run_{run_id}" -# -# try: -# jail_dir.mkdir(parents=True, exist_ok=True) -# -# custom_env = { -# "PYTHONPATH": str(WORKSPACE_DIR), -# "PATH": str(Path(PYTHON_EXE).parent), -# "HOME": str(jail_dir), -# "TMPDIR": str(jail_dir) -# } -# -# result = subprocess.run( -# [PYTHON_EXE, "-c", code], -# cwd=str(WORKSPACE_DIR), -# env=custom_env, -# 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]..." -# -# if not output.strip(): -# return "Code executed successfully (no output)." -# -# return 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}" -# -# finally: -# if jail_dir.exists(): -# shutil.rmtree(jail_dir) + return "\n".join(report) @mcp.tool() def run_python_sandboxed(code: str) -> str: @@ -422,61 +253,6 @@ def run_python_sandboxed(code: str) -> str: except Exception as e: return f"Error during code execution: {e}" -#@mcp.tool() -#def run_python_code(code: str) -> str: -# """ -# Execute Python code safely inside a temporary directory. -# -# Args: -# code: The Python code or the File content of a python file to be executed in str format -# -# Returns: -# Combined stdout+stderr, or an error message in str format. -# """ -# -# static_safety = check_code_safety(code) -# if static_safety: -# return f"Code rejected:{static_safety}" -# -# linted_code = lint_code(code) -# if not linted_code[1]: -# return f"Code is not executable. Errror while linting:{linted_code[0]}" -# -# # 2. Create isolated temp directory -# with tempfile.TemporaryDirectory() as tmp_dir: -# -# tmp_path = Path(tmp_dir) -# -# script_file = tmp_path / "main.py" -# -# script_file.write_text(code, encoding="utf-8") -# -# try: -# result = subprocess.run( -# ["python", str(script_file)], -# capture_output=True, -# text=True, -# timeout=EXEC_TIMEOUT, -# cwd=tmp_dir -# ) -# -# output = result.stdout + result.stderr -# -# if len(output) > MAX_OUTPUT_LENGTH: -# output = output[:MAX_OUTPUT_LENGTH] -# output += "\n...[output truncated]..." -# -# if not output.strip(): -# return "Code executed successfully." -# -# return output -# -# except subprocess.TimeoutExpired: -# return f"Execution stopped: Timeout after {EXEC_TIMEOUT} seconds." -# -# except Exception as e: -# return f"Execution error: {e}" - @mcp.tool() def python_code_validation(code: str) -> str: """ @@ -500,6 +276,8 @@ def python_code_validation(code: str) -> str: return f"Valid Syntax, but with safety concerns: {static_analysis_result}; code execution is not allowed." except Exception as e: return f"Error during code safety analysis: {e}" + + return "Code is valid and can be executed in the sandbox" # ── Run the server ─────────────────────────────────────────────────────────── diff --git a/requirements.txt b/requirements.txt index 850c7a9..c98c067 100644 --- a/requirements.txt +++ b/requirements.txt @@ -23,3 +23,6 @@ python-dotenv>=1.0.0 #For code editor functionality streamlit-ace>=0.1.0 + +#Whitelisted Imports from Agent-Sandbox +pygame \ No newline at end of file