mcp code execution fix
This commit is contained in:
Irina Rueegg
parent
901fbab543
commit
c8299f89c3
@ -1,35 +1,34 @@
|
||||
import ast
|
||||
from datetime import datetime
|
||||
import subprocess
|
||||
import tempfile
|
||||
import io
|
||||
from pyflakes.api import check
|
||||
from pyflakes.reporter import Reporter
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
from pathlib import Path
|
||||
import venv
|
||||
import shutil
|
||||
|
||||
# ── 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"
|
||||
#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 Path("C:/").exists() else "bin" # Einfacher Check für Windows
|
||||
|
||||
python_exe = SANDBOX_DIR / bin_folder / "python"
|
||||
pip_exe = SANDBOX_DIR / bin_folder / "pip"
|
||||
|
||||
return str(python_exe), str(pip_exe)
|
||||
|
||||
PYTHON_EXE, PIP_EXE = get_sandbox_paths()
|
||||
#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 = 10 # 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 ────────────────────────────────────────────────────
|
||||
@ -68,6 +67,8 @@ FORBIDDEN_SEQUENCES = ["../", "..\\", "/etc/", "/dev/",
|
||||
"C:\\Windows", "C:\\Program Files", "C:\\Users",
|
||||
"compile(", "__import__", "os.", "sys.", "subprocess."]
|
||||
|
||||
ALLOWED_PACKAGES = ["pygame", "numpy", "pandas"]
|
||||
|
||||
# ── Static Analysis ────────────────────────────────────────────────────
|
||||
def check_code_safety(code: str) -> str | None:
|
||||
"""
|
||||
@ -188,7 +189,7 @@ def analyse_structure(code: str) -> str:
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def lint_code(code: str) -> str:
|
||||
def lint_code(code: str) -> tuple[str, bool]:
|
||||
"""
|
||||
Runs a fast static analysis check to catch syntax errors, unused imports,
|
||||
or undefined variables without executing the code.
|
||||
@ -206,14 +207,14 @@ def lint_code(code: str) -> str:
|
||||
try:
|
||||
check(code, filename="<agent_code>", reporter=reporter)
|
||||
except Exception as e:
|
||||
return f"Critical error during linting: {str(e)}"
|
||||
return (f"Critical error during linting: {str(e)}", False)
|
||||
|
||||
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."
|
||||
return ("Linting complete: No issues found. The code is syntactically sound.", True)
|
||||
|
||||
report = ["--- Linting Report ---"]
|
||||
|
||||
@ -227,88 +228,158 @@ def lint_code(code: str) -> str:
|
||||
|
||||
report.append("\nAdvice: Please fix these issues before attempting to execute the code.")
|
||||
|
||||
return "\n".join(report)
|
||||
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)
|
||||
|
||||
@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:
|
||||
"""Löscht die gesamte Sandbox und erstellt sie neu (Full Reset)."""
|
||||
if SANDBOX_DIR.exists():
|
||||
shutil.rmtree(SANDBOX_DIR)
|
||||
get_sandbox_paths()
|
||||
return "Sandbox wurde komplett zurückgesetzt."
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def run_python_code_sandboxed(code: str) -> str:
|
||||
def run_python_sandboxed(code: str) -> str:
|
||||
"""
|
||||
Run Python code in a sandboxed environment.
|
||||
|
||||
@ -329,23 +400,9 @@ def run_python_code_sandboxed(code: str) -> str:
|
||||
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,
|
||||
["python", "-c", code],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=EXEC_TIMEOUT)
|
||||
@ -365,10 +422,60 @@ def run_python_code_sandboxed(code: str) -> str:
|
||||
except Exception as e:
|
||||
return f"Error during code execution: {e}"
|
||||
|
||||
finally:
|
||||
if jail_dir.exists():
|
||||
shutil.rmtree(jail_dir)
|
||||
|
||||
#@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:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user