meulilivio/AISE1_Project_Irina_Livio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

validate fix

Browse Source
This commit is contained in:
Irina Rueegg 2026-05-21 16:11:21 +02:00
parent c8299f89c3
commit e75212652e
4 changed files with 14 additions and 353 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
backend/agent/mcp_server_adapter.py
View File

@ -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}")

114
backend/agent/mcp_server_adapter_RAG.py
View File

@ -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}")

242
backend/agent/servers/mcp_server_code_execution.py
View File

@ -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="<agent_code>", 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 ───────────────────────────────────────────────────────────

3
requirements.txt
View File

@ -23,3 +23,6 @@ python-dotenv>=1.0.0
#For code editor functionality
streamlit-ace>=0.1.0
#Whitelisted Imports from Agent-Sandbox
pygame