diff --git a/.gitignore b/.gitignore
index 7309b3c..c3aa73e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -50,3 +50,6 @@ data/raw/
# Workspace
workspace/
+
+# Logs
+logs/
diff --git a/backend/agent/coding_agent.py b/backend/agent/coding_agent.py
index fd0a371..9b15f52 100644
--- a/backend/agent/coding_agent.py
+++ b/backend/agent/coding_agent.py
@@ -18,18 +18,20 @@ import os
import re
from pathlib import Path
import asyncio
-import pprint
import requests
from dotenv import load_dotenv
#from mcp_server_adapter import MCPToolAdapter # Import from current directory for easier testing without package structure
from backend.agent.mcp_server_adapter import MCPToolAdapter
+from backend.managers.debug_logger import get_logger
+logger = get_logger(__name__)
+
# ── mcp server initialization ────────────────────────────────────────────────────────────────
adapter = MCPToolAdapter()
-print("MCPToolAdapter created. Listing all tools from servers...")
+logger.info("MCPToolAdapter created. Listing all tools from servers...")
asyncio.run(adapter.initialize_all_servers())
-print("listed tools from all servers")
+logger.info("Listed tools from all servers")
load_dotenv()
@@ -52,11 +54,10 @@ def build_all_tool_description() -> str:
"""Get relevant tools from the MCP servers based on the query."""
all_tools = adapter.get_all_tools()
- print(f"Building tool description for {len(all_tools)} tools.")
+ logger.info("Building tool description for %s tools.", str(len(all_tools)))
descriptions = []
for tool in all_tools:
- pprint.pprint(f"{tool}")
descriptions.append(f"- {tool['tool_name']}: {tool['tool_description']}")
return "\n".join(descriptions)
@@ -70,20 +71,21 @@ async def dispatch_tool(tool_name: str, arguments: dict) -> str:
return f"DONE: {summary}"
try:
- print(f"Trying to call tool '{tool_name}' with arguments: {arguments}")
- print(f"Trying to call tool '{tool_name}' in dispatch_tool through MCPToolAdapter...")
+ logger.info("Calling tool '%s' in dispatch_tool through MCPToolAdapter...", tool_name)
result = await adapter.call_tool(tool_name, arguments)
- print(f"Raw result from tool '{tool_name}': {result}")
+ logger.info(f"Result from tool '%s' recieved", tool_name)
if result.isError:
texts = [block.text for block in result.content if block.type == "text"]
+ logger.warning("Result from '%s' is Error", tool_name)
return f"Tool error: {' '.join(texts)}"
texts = [block.text for block in result.content if block.type == "text"]
return "\n".join(texts)
except Exception as e:
+ logger.exception(f"Error calling tool '%s' with argument: %s", tool_name, arguments)
return f"Error calling tool '{tool_name}': {e}"
# ═════════════════════════════════════════════════════════════════════════════
@@ -157,6 +159,7 @@ Example:
def truncate_result(result: str) -> str:
"""Truncate a tool result that exceeds MAX_RESULT_LENGTH."""
+ logger.info("Result has been truncated")
if len(result) <= MAX_RESULT_LENGTH:
return result
half = MAX_RESULT_LENGTH // 2
@@ -171,6 +174,8 @@ def trim_messages(messages: list) -> list:
"""Drop old messages when history exceeds MAX_HISTORY_CHARS.
Always keeps the system prompt (index 0) and original task (index 1).
"""
+ logger.info("Message is being trimmed")
+
total = sum(len(m["content"]) for m in messages)
if total <= MAX_HISTORY_CHARS:
return messages
@@ -331,7 +336,6 @@ class CodingAgent:
#async def _call_api(self, messages: list) -> str:
def _call_api(self, messages: list) -> str:
-
"""Make a raw API call and return the response content string."""
headers = {"Content-Type": "application/json"}
@@ -345,16 +349,31 @@ class CodingAgent:
"max_tokens": 4096,
"stream": False,
}
-
- response = requests.post(self.api_url, headers=headers, json=payload, timeout=60)
-
- if response.status_code != 200:
- raise Exception(f"API Error {response.status_code}: {response.text}")
+ try:
+ response = requests.post(
+ self.api_url,
+ headers=headers,
+ json=payload,
+ timeout=60)
+ response.raise_for_status()
+ logger.info("LLM API response requested")
+
+ if response.status_code != 200:
+ logger.error("API Error %s: %s", response.status_code, response.text)
+ raise Exception(f"API Error {response.status_code}: {response.text}")
+ except requests.RequestException as exc:
+ logger.exception("API Error; HTTP-Fehler: %s", exc)
+ raise Exception(f"HTTP-Fehler: {exc}") from exc
+
data = response.json()
if "choices" in data and len(data["choices"]) > 0:
+ logger.info("valid API output, data returned")
return data["choices"][0]["message"]["content"]
+
+ logger.error("Invalid API response format")
raise Exception("Invalid API response format")
+
# ── Public interface ──────────────────────────────────────────────────────
@@ -367,6 +386,7 @@ class CodingAgent:
self.pending_action = None
self.is_done = False
self.iteration = 0
+ logger.info("New ask initialized")
async def propose_next_action(self) -> dict:
"""Ask the LLM what to do next.
@@ -393,6 +413,7 @@ class CodingAgent:
raw = _strip_code_fences(raw)
cleaned = extract_json(raw)
action = json.loads(cleaned)
+ logger.info("Propose next action successfull")
except json.JSONDecodeError:
action = {
"thought": "Could not parse LLM response as JSON.",
@@ -400,6 +421,7 @@ class CodingAgent:
"arguments": {"summary": "Stopped: JSON parse error."},
}
raw = json.dumps(action)
+ logger.critical("Parsing API response into valid JASON failed in Step 'propose_next_action'")
except Exception as e:
action = {
"thought": f"API call failed: {e}",
@@ -407,6 +429,7 @@ class CodingAgent:
"arguments": {"summary": f"Stopped: {e}"},
}
raw = json.dumps(action)
+ logger.critical("API call faliled in Step %s: %s", self.iteration, e)
self.pending_action = {"raw": raw, "action": action}
return action
@@ -429,6 +452,8 @@ class CodingAgent:
self.messages.append({"role": "assistant", "content": raw})
self.pending_action = None
+ logger.info("Messages prepared after approval")
+
# Handle completion
if tool_name == "done":
self.is_done = True
@@ -442,6 +467,7 @@ class CodingAgent:
# Execute the tool
result = await dispatch_tool(tool_name, arguments)
result = truncate_result(result)
+ logger.info("Tool called and result truncated")
# Wrap the tool output in an XML tag so the LLM can easily find it.
# Append a tag on errors to force the agent to reconsider
@@ -453,6 +479,7 @@ class CodingAgent:
"Re-examine your plan: what went wrong and what should you do differently? "
"State your revised plan in your next thought."
)
+ logger.warning("Error Message in the tool result, replan-feedback will appended")
self.messages.append({"role": "user", "content": feedback})
@@ -480,6 +507,7 @@ class CodingAgent:
"address their question accordingly."
),
})
+ logger.info("Follow-up message appended.")
def reject(self, feedback: str) -> None:
"""Reject the pending action and inject user feedback.
@@ -506,6 +534,7 @@ class CodingAgent:
),
})
self.pending_action = None
+ logger.info("Rejection message appended.")
def main():
"""Example of how to use the CodingAgent in a simple loop."""
diff --git a/backend/agent/mcp_server_adapter.py b/backend/agent/mcp_server_adapter.py
index c87b5ae..80cb4af 100644
--- a/backend/agent/mcp_server_adapter.py
+++ b/backend/agent/mcp_server_adapter.py
@@ -7,6 +7,9 @@ from pathlib import Path
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
+from backend.managers.debug_logger import get_logger
+logger = get_logger(__name__)
+
class MCPToolAdapter:
def __init__(self, config_path: str = "mcp_server_config.json"):
self.config_path = config_path
@@ -17,24 +20,25 @@ class MCPToolAdapter:
"""Lädt die Server-Konfiguration aus der JSON-Datei."""
path = Path(__file__).parent / self.config_path
if not path.exists():
- print(f"Config file not found: {path}")
+ logger.warning("Config file not found: %s", path)
return {}
try:
with open(path, 'r') as f:
- return json.load(f)
+ config_file = json.load(f)
+ logger.info("MCP-Server config loaded successfully")
+ return config_file
except json.JSONDecodeError as e:
- print(f"Error decoding JSON config: {e}")
+ logger.critical("Error decoding JSON from server config: %s", e)
return {}
async def initialize_all_servers(self):
"""Lädt die Konfiguration und fragt alle Server ab, um die Tools zu registrieren."""
- print("Initializing MCP sessions...")
config = self._load_config()
- print(f"Loaded config for servers: {list(config.keys())}")
+ logger.info("Loaded config for servers: %s", list(config.keys()))
for server_name, params in config.items():
- print(f"Testing connection to {server_name}...")
+ logger.info("Initializing connection to %s", server_name)
self.servers[server_name] = params
server_script = str(Path(__file__).parent / params["args"][0])
@@ -51,13 +55,13 @@ class MCPToolAdapter:
try:
# Verbindung aufbauen
async with stdio_client(server_params) as (read_stream, write_stream):
- print(f"Connected to {server_name}. Initializing session...")
+ logger.info("Connected to %s. Initializing session...", server_name)
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
- print(f"Session initialized for {server_name}. Requesting tools...")
+ logger.info("Session initialized for %s. Requesting tools...", server_name)
result = await session.list_tools()
tools = result.tools
- print(f"Tools received from {server_name}: {len(tools)} Tools")
+ logger.info(f"Tools received from %s: %s Tools", server_name, str(len(tools)))
for tool in tools:
t_params = tool.inputSchema.get("properties", {})
@@ -80,10 +84,10 @@ class MCPToolAdapter:
"tool_description": t_definition
})
- print(f"Registered tool '{tool.name}' from {server_name}.")
+ logger.info("Registered tool '%s' from %s.", tool.name, server_name)
except Exception as e:
- print(f"Failed to initialize {server_name}: {e}")
+ logger.exception("Failed to initialize %s: %s", server_name, str(e))
def get_all_tools(self) -> List[Dict[str, Any]]:
"""Gibt alle gesammelten Tools zurück."""
@@ -95,7 +99,7 @@ class MCPToolAdapter:
tool_entry = next((t for t in self.tool_registry if t["tool_name"] == tool_name), None)
if not tool_entry:
- print(f"Tool '{tool_name}' not found in MCP adapter registry.")
+ logger.warning("Tool '%s' not found in MCP adapter registry.", tool_name)
return f"Error: Tool '{tool_name}' not found in registry."
server_name = tool_entry["server"]
@@ -117,24 +121,15 @@ class MCPToolAdapter:
async with stdio_client(server_params) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
+ logger.info("Session successfully initialized, calling tool '%s' on server '%s", tool_name, server_name)
result = await session.call_tool(tool_name, arguments)
return result
except Exception as e:
+ logger.exception("Error calling tool '%s' on server '%s': %s", tool_name, server_name, str(e))
return f"Error calling tool '{tool_name}' on server '{server_name}': {str(e)}"
-
- return f"Error: Session for server '{server_name}' not active."
-
- async def shutdown_all_sessions(self):
- """Schließt alle offenen Verbindungen sauber."""
- 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.")
- except Exception as e:
- print(f"Error during shutdown of {server_name}: {e}")
def main():
+ """Debug Function for Tool-Registry"""
adapter = MCPToolAdapter()
asyncio.run(adapter.initialize_all_servers())
print("All servers initialized. Registered tools:")
diff --git a/backend/agent/mcp_server_config.json b/backend/agent/mcp_server_config.json
index 76ede4d..ade2521 100644
--- a/backend/agent/mcp_server_config.json
+++ b/backend/agent/mcp_server_config.json
@@ -5,10 +5,7 @@
"WebSearchServer": {
"command": "py",
- "args": ["servers/mcp_server_web_search.py"],
- "env": {
- "DDGS_API_KEY": "your_ddgs_api_key_here"
- }
+ "args": ["servers/mcp_server_web_search.py"]
},
"CodeExecutionServer": {
diff --git a/backend/agent/servers/mcp_server_code_execution.py b/backend/agent/servers/mcp_server_code_execution.py
index c2608e5..7e1dfb2 100644
--- a/backend/agent/servers/mcp_server_code_execution.py
+++ b/backend/agent/servers/mcp_server_code_execution.py
@@ -8,8 +8,11 @@ from pyflakes.reporter import Reporter # For linting Code
from mcp.server.fastmcp import FastMCP
from pathlib import Path
+#from backend.managers.debug_logger import get_logger
+#logger = get_logger(__name__)
+
# ── Configuration ────────────────────────────────────────────────────────────
-EXEC_TIMEOUT = 45 # seconds before killing the subprocess
+EXEC_TIMEOUT = 15 # seconds before killing the subprocess
MAX_OUTPUT_LENGTH = 3000 # max characters of stdout+stderr to return
# ── Create the MCP server ────────────────────────────────────────────────────
@@ -48,8 +51,9 @@ FORBIDDEN_SEQUENCES = ["../", "..\\", "/etc/", "/dev/",
"C:\\Windows", "C:\\Program Files", "C:\\Users",
"compile(", "__import__", "os.", "sys.", "subprocess."]
-ALLOWED_PACKAGES = ["pygame", "numpy", "pandas"]
-
+"""
+Pre-installed Packages in Sandbox: "pygame", "numpy", "pandas"
+"""
# ── Static Analysis ────────────────────────────────────────────────────
def check_code_safety(code: str) -> str | None:
"""
@@ -62,10 +66,13 @@ def check_code_safety(code: str) -> str | None:
str or None
Error message if forbidden code found, None if safe.
"""
+ #logger.info("Checking code safety.")
try:
tree = ast.parse(code)
+ #logger.info("Code has valid Syntax")
except SyntaxError as e:
+ #logger.exception("SyntaxError: %s", e)
return f"SyntaxError: {e}"
for node in ast.walk(tree):
@@ -73,6 +80,7 @@ def check_code_safety(code: str) -> str | None:
for alias in node.names:
top_level_module = alias.name.split('.')[0]
if top_level_module in BLOCKED_IMPORTS:
+ #logger.warning("Blocked import '%s'", alias.name)
return (f"Blocked import: Import of '{alias.name}' is not allowed."
f"line {node.lineno}")
@@ -80,16 +88,19 @@ def check_code_safety(code: str) -> str | None:
if node.module:
top_level = node.module.split(".")[0]
if top_level in BLOCKED_IMPORTS:
+ #logger.warning("Blocked import from '%s'", alias.name)
return (f"Blocked import: Import from '{node.module}' is not allowed."
f"(module '{top_level}' is blocked) line {node.lineno}")
elif isinstance(node, ast.Call):
if isinstance(node.func, ast.Name):
if node.func.id in BLOCKED_BUILTINS:
+ #logger.warning("Blocked ubiltin '%s'", node.func.id)
return f"Blocked builtin: Use of builtin '{node.func.id}' is not allowed."
for seq in FORBIDDEN_SEQUENCES:
if seq in code:
+ #logger.warning("Suspect path sequence '%s' detected.", seq)
return f"Blocked: Suspect path sequence '{seq}' detected."
return None # No violations found
@@ -104,11 +115,16 @@ def analyse_structure(code: str) -> str:
Returns:
A summary of the code's structure, including functions, classes, and imports.
"""
+ #logger.info("Tool analyse_structure is being executed on MCP code execution server")
+
try:
tree = ast.parse(code)
+ #logger.info("Code tree parsed successfully")
except SyntaxError as e:
+ #logger.warning("Syntax Error in provided code. Line %s : %s", e.lineno, e.msg)
return f"Syntax Error: Invalid Python code provided. Line {e.lineno}: {e.msg}"
except Exception as e:
+ #logger.exception("Error parsing code: %s", str(e))
return f"Error parsing code: {str(e)}"
analysis = {
@@ -164,6 +180,7 @@ def analyse_structure(code: str) -> str:
lines.append(f" - def {func['name']}({args_str})")
if not any([analysis["imports"], analysis["classes"], analysis["functions"]]):
+ #logger.info("Code analysis successfull but no top-level items found")
return "Analysis complete: No top-level imports, classes, or functions found."
return "\n".join(lines)
@@ -180,6 +197,8 @@ def lint_code(code: str) -> str:
Returns:
A report of linting issues or a success message if the code is clean.
"""
+ #logger.info("Tool lint_code is being executed on MCP code execution server")
+
error_buffer = io.StringIO()
warning_buffer = io.StringIO()
@@ -187,7 +206,9 @@ def lint_code(code: str) -> str:
try:
check(code, filename="", reporter=reporter)
+ #logger.info("Linting successfull")
except Exception as e:
+ #logger.exception("Critical error during linting: %s", str(e))
return f"Critical error during linting: {str(e)}"
errors = error_buffer.getvalue().strip()
@@ -195,6 +216,7 @@ def lint_code(code: str) -> str:
# Ergebnis-String zusammenbauen
if not errors and not warnings:
+ #logger.info("No issues found")
return "Linting complete: No issues found. The code is syntactically sound."
report = ["--- Linting Report ---"]
@@ -209,6 +231,7 @@ def lint_code(code: str) -> str:
report.append("\nAdvice: Please fix these issues before attempting to execute the code.")
+ #logger.info("There are issues with provided code. Check Report: %s", "\n".join(report))
return "\n".join(report)
@mcp.tool()
@@ -228,6 +251,7 @@ def run_python_sandboxed(code: str) -> str:
Returns:
Combined stdout+stderr, or an error message in str format.
"""
+ #logger.info("Tool run_python_sandboxed is being executed on MCP code execution server")
static_safety = check_code_safety(code)
if static_safety:
@@ -249,11 +273,15 @@ def run_python_sandboxed(code: str) -> str:
if not output.strip():
return "Code executed successfully (no output)."
+ #logger.info("Code ran successfully")
return output
except subprocess.TimeoutExpired:
+ #logger.warning("Code execution exceeded time limit of %s seconds", EXEC_TIMEOUT)
return f"Error: Code execution exceeded time limit of {EXEC_TIMEOUT} seconds and was terminated."
+
except Exception as e:
+ #logger.exception("Error during code execution: %s", str(e))
return f"Error during code execution: {e}"
@mcp.tool()
@@ -268,16 +296,22 @@ def python_code_validation(code: str) -> str:
A message indicating the validation result.
And if sandboxed test execution is allowed.
"""
+ #logger.info("Tool python_code_validation is being executed on MCP code execution server")
+
try:
ast.parse(code)
+ #logger.info("Ast parsing successfull")
except SyntaxError as e:
+ #logger.warning("Syntax Error while ast parsing code: %s", str(e))
return f"SyntaxError: {e}"
try:
static_analysis_result = check_code_safety(code)
if static_analysis_result:
+ #logger.info("Code safety issues detected")
return f"Valid Syntax, but with safety concerns: {static_analysis_result}; code execution is not allowed."
except Exception as e:
+ #logger.exception("Error during code safety analysis: %e", str(e))
return f"Error during code safety analysis: {e}"
return "Code is valid and can be executed in the sandbox"
diff --git a/backend/agent/servers/mcp_server_file_search.py b/backend/agent/servers/mcp_server_file_search.py
index 65180bc..37ba647 100644
--- a/backend/agent/servers/mcp_server_file_search.py
+++ b/backend/agent/servers/mcp_server_file_search.py
@@ -1,6 +1,9 @@
from pathlib import Path
from mcp.server.fastmcp import FastMCP
+#from backend.managers.debug_logger import get_logger
+#logger = get_logger(__name__)
+
# ── Configuration ────────────────────────────────────────────────────────────
project_dir = Path(__file__).resolve().parent.parent.parent.parent
ALLOWED_DIR = project_dir / "workspace"
@@ -15,10 +18,12 @@ 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)):
+ #logger.warning("Access denied: '%s' resolves outside allowed directory.", str(requested))
raise ValueError(
f"Access denied: '{requested}' resolves outside "
f"the allowed directory '{ALLOWED_DIR}'"
)
+ #logger.info("Requested path is safe")
return resolved
@@ -30,6 +35,8 @@ def list_files() -> str:
Returns a newline-separated list of relative file paths.
"""
+ #logger.info("Tool list_files is being executed on MCP file search server")
+
files = sorted(
f.relative_to(ALLOWED_DIR)
for f in ALLOWED_DIR.rglob("*")
@@ -50,18 +57,22 @@ def get_file_tree(dir_path: str=ALLOWED_DIR) -> str:
Returns:
A string representing the directory structure, similar to 'tree' command output.
"""
+ #logger.info("Tool get_file_tree is being executed on MCP file search server")
try:
safe_dir = _safe_path(dir_path)
if not safe_dir:
return f"Error: Invalid directory path '{dir_path}'."
elif not safe_dir.exists():
+ #logger.warning("Directory '%s' does not exist.", dir_path)
return f"Error: Directory '{dir_path}' does not exist."
elif not safe_dir.is_dir():
+ #logger.warning("'%s' is not a valid directory", dir_path)
return f"Error: '{dir_path}' is not a valid directory within the allowed path."
except ValueError as e:
+ #logger.exception("Error while checking directory and its path: %s", str(e))
return f"Error: {e}"
-
+ #logger.info("Generating file tree.")
def _tree(dir_path: Path, prefix="") -> str:
entries = sorted([e for e in dir_path.iterdir() if "__pycache__" not in e.parts], key=lambda x: (x.is_file(), x.name))
lines = []
@@ -73,7 +84,7 @@ def get_file_tree(dir_path: str=ALLOWED_DIR) -> str:
lines.append(_tree(entry, prefix + extension))
return "\n".join(lines)
- return _tree(dir_path)
+ return _tree(Path(dir_path))
@mcp.tool()
@@ -86,6 +97,8 @@ def search_files(query: str) -> str:
Returns:
A formatted string of search results, or a message if no matches found.
"""
+ #logger.info("Tool search_files is being executed on MCP file search server")
+
query_lower = query.lower()
results = []
@@ -103,11 +116,17 @@ def search_files(query: str) -> str:
if query_lower in line.lower():
snippet = line.strip()[:100]
results.append(f"[content] {rel}:{i} -- {snippet}")
- except (UnicodeDecodeError, PermissionError):
+ except UnicodeDecodeError as e:
+ #logger.warning("Decode error in file/folder '%s': %s", f, e)
+ pass
+ except PermissionError as e:
+ #logger.warning("Permission error in file/folder: '%s': %s", f, e)
pass
if not results:
+ #logger.info("No matches found for user query")
return f"No matches found for '{query}'."
+ #logger.info("Result returned, limited to 30 matches.")
return "\n".join(results[:30]) # limit to 30 matches
@@ -121,23 +140,33 @@ def read_file(path: str) -> str:
Returns:
The file content as a string, or an error message if the file cannot be read.
"""
+ #logger.info("Tool read_file is being executed on MCP file search server")
+
try:
resolved = _safe_path(path)
except ValueError as e:
return f"Error: {e}"
if not resolved.exists():
+ #logger.warning("File '%s' does not exist.", path)
return f"Error: File '{path}' does not exist."
if not resolved.is_file():
+ #logger.warning("'%s' is not a valid file.", path)
return f"Error: '{path}' is not a file."
try:
- return resolved.read_text(encoding="utf-8")
+ text = resolved.read_text(encoding="utf-8")
+ #logger.info("File read successfully.")
+ return text
+
except UnicodeDecodeError:
+ #logger.warning("'%s' is not a text file (binary content).", path)
return f"Error: '{path}' is not a text file (binary content)."
except PermissionError:
+ #logger.warning(f"Permission denied when trying to read '%s'.", path)
return f"Error: Permission denied when trying to read '{path}'."
except Exception as e:
+ #logger.exception("Error reading file '%s': %s", path, e)
return f"Error reading file '{path}': {e}"
@mcp.tool()
@@ -152,6 +181,7 @@ def write_new_file(path: str, content: str) -> str:
Returns:
A success or error message.
"""
+ #logger.info("Tool write_new_file is being executed on MCP file search server")
try:
resolved = _safe_path(path)
@@ -159,26 +189,30 @@ def write_new_file(path: str, content: str) -> str:
return f"Error: {e}"
if resolved.exists():
+ #logger.warning("Requested file path '%s' already exists, overwriting not allowed.", path)
return (f"ERROR: File '{path}' already exists."
f"Overwriting is not allowed with this tool."
f"Use a different path or filename to create a new file.")
if resolved.suffix not in ALLOWED_FILE_TYPES:
+ #logger.warning("File type not allowed: %s", resolved.suffix)
return f"ERROR: can only write {', '.join(ALLOWED_FILE_TYPES)} types, got '{resolved.suffix}'."
try:
resolved.parent.mkdir(parents=True, exist_ok=True)
resolved.write_text(content, encoding="utf-8")
+ #logger.info("File written successfully.")
return f"OK: wrote {len(content)} chars to {path}."
except FileNotFoundError as e:
- print(f"FileNotFoundError for {path}: {e}")
+ #logger.warning("FileNotFoundError for '%s': %s", path, e)
return f"Error: {e}"
except PermissionError as e:
- print(f"PermissionError for {path}: {e}")
+ #logger.warning("PermissionError for '%s': %s", path, e)
return f"Error: {e}"
except Exception as e:
+ #logger.exception("Error writing file: %s", e)
return f"Error: {e}"
@@ -192,23 +226,29 @@ def create_new_directory(path: str) -> str:
Returns:
A success or error message.
"""
+ #logger.info("Tool create_new_directory is being executed on MCP file search server")
+
try:
resolved = _safe_path(path)
except ValueError as e:
return f"Error: {e}"
if resolved.exists():
+ #logger.warning("Requested path '%s' already exists, overwriting not allowed.", path)
return f"Error: File '{path}' already exists."
if resolved.suffix != None and resolved.suffix != "":
+ #logger.warning("Can only create directories, got '%s'.", resolved.suffix)
return f"Error: can only create directories, got '{resolved.suffix}'."
try:
resolved.parent.mkdir(parents=True, exist_ok=True)
resolved.mkdir()
+ #logger.info("Directory '%s' created successfully.", path)
return f"OK: created empty directory at {path}."
except Exception as e:
- return f"Error creating dictionary file '{path}': {e}"
+ #logger.exception("Error creating directory '%s': %s", path, e)
+ return f"Error creating directory '{path}': {e}"
# ── Run the server ───────────────────────────────────────────────────────────
diff --git a/backend/agent/servers/mcp_server_web_search.py b/backend/agent/servers/mcp_server_web_search.py
index 39c9fb4..03c7a7c 100644
--- a/backend/agent/servers/mcp_server_web_search.py
+++ b/backend/agent/servers/mcp_server_web_search.py
@@ -1,10 +1,26 @@
from urllib.parse import urlparse
+import requests
+from bs4 import BeautifulSoup
+from ddgs import DDGS
from mcp.server.fastmcp import FastMCP
+#from backend.managers.debug_logger import get_logger
+#logger = get_logger(__name__)
+
# ── Configuration ────────────────────────────────────────────────────────────
MAX_PAGE_LENGTH = 4000 # max characters to return from a fetched page
REQUEST_TIMEOUT = 10 # seconds
+# ── Bolcked prefixes & Hosts ────────────────────────────────────────────────────
+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.",
+]
+
+BLOCKED_HOSTS = ["localhost", "127.0.0.1", "0.0.0.0", "169.254.169.254"]
+
# ── Create the MCP server ────────────────────────────────────────────────────
mcp = FastMCP("WebSearchServer")
@@ -14,26 +30,23 @@ mcp = FastMCP("WebSearchServer")
def _validate_url(url: str) -> str:
"""Validate a URL to prevent SSRF attacks."""
parsed = urlparse(url)
+ #logger.info("Validateing URL")
if parsed.scheme not in ("http", "https"):
+ #logger.warning("Blocked scheme '%s'. Only http and https are allowed.", parsed.scheme)
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:
+ if hostname in BLOCKED_HOSTS:
+ #logger.warning("Blocked internal host: %s", hostname)
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:
+
+ for prefix in PRIVATE_PREFIXES:
if hostname.startswith(prefix):
+ #logger.warning("Blocked private IP range: %s", hostname)
raise ValueError(f"Blocked private IP range: {hostname}")
return url
@@ -51,12 +64,16 @@ def web_search(query: str, max_results: int = 5) -> str:
Returns:
A formatted string of search results, or a message if no matches found.
"""
+ #logger.info("Tool web_search is being executed on MCP web search server")
+
try:
- from ddgs import DDGS
results = DDGS().text(query, max_results=max_results)
if not results:
+ #logger.info("DDGS API call successful, no web search results found.")
return f"No results found for: {query}"
+
+ #logger.info("DDGS API call successfull, web search results returned.")
formatted = []
for r in results:
@@ -68,6 +85,7 @@ def web_search(query: str, max_results: int = 5) -> str:
return "\n---\n".join(formatted)
except Exception as e:
+ #logger.exception("DDGS API call failed, web search error: %s", e)
return f"Search error: {e}"
@@ -80,24 +98,36 @@ def fetch_page(url: str) -> str:
Returns:
The text content of the fetched page, or an error message.
"""
+ #logger.info("Tool fetch_page is being executed on MCP web search server")
+
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)"},
)
+ response.raise_for_status()
if response.status_code != 200:
+ #logger.warning("HTTP error %s while fetching %s", response.status_code, url)
return f"HTTP error {response.status_code} fetching {url}"
+
+ #logger.info("DDGS API call successfull")
+ except requests.RequestException as e:
+ #logger.warning("HTTP-Fehler: %s", e)
+ return f"HTTP-Fehler: {e}"
+
+ except Exception as e:
+ #logger.exception("Error fetching page: %s", e)
+ return f"Error fetching page: {e}"
+
+ try:
soup = BeautifulSoup(response.text, "html.parser")
for tag in soup(["script", "style", "nav", "footer"]):
@@ -105,13 +135,16 @@ def fetch_page(url: str) -> str:
text = soup.get_text(separator="\n", strip=True)
+ #logger.info("HTML parsing with BeautifulSoup successfull")
+
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}"
+ #logger.exception("Error parsing HTML: %s", e)
+ return f"Error parsing html: {e}"
# ── Run the server ───────────────────────────────────────────────────────────
diff --git a/backend/managers/_debug_logger.py b/backend/managers/_debug_logger.py
new file mode 100644
index 0000000..af1fc13
--- /dev/null
+++ b/backend/managers/_debug_logger.py
@@ -0,0 +1,71 @@
+from datetime import datetime
+
+
+class DebugLogger:
+ """In-memory logger for code execution events.
+
+ Collects timestamped INFO and ERROR entries during a single run.
+ Call clear() before each new execution to start fresh.
+ """
+
+ def __init__(self):
+ self.logs: list[dict] = []
+
+ def log(self, message: str) -> None:
+ """Append a general info message."""
+ self.logs.append({
+ "level": "INFO",
+ "message": message,
+ "timestamp": datetime.now().strftime("%H:%M:%S"),
+ })
+
+ def log_error(self, error_message: str) -> None:
+ """Append an error message."""
+ self.logs.append({
+ "level": "ERROR",
+ "message": error_message,
+ "timestamp": datetime.now().strftime("%H:%M:%S"),
+ })
+
+ def get_logs(self) -> list[dict]:
+ """Return a copy of all collected log entries."""
+ return list(self.logs)
+
+ def clear(self) -> None:
+ """Reset the log — call before each new execution."""
+ self.logs = []
+
+ def format_debug_output(self, output: dict) -> str:
+ """Format an ExecutionEngine result dict into a human-readable string.
+
+ Args:
+ output: dict with keys 'stdout', 'stderr', and 'rc'.
+
+ Returns:
+ A formatted string ready for display in the UI.
+ """
+ lines = []
+
+ status = "SUCCESS" if output.get("rc") == 0 else "FAILED"
+ lines.append(f"[{status}] Exit code: {output.get('rc')}")
+
+ if output.get("stdout"):
+ lines.append("\n--- stdout ---")
+ lines.append(output["stdout"].rstrip())
+
+ if output.get("stderr"):
+ lines.append("\n--- stderr ---")
+ lines.append(output["stderr"].rstrip())
+
+ if not output.get("stdout") and not output.get("stderr"):
+ lines.append("No output produced.")
+
+ for entry in self.logs:
+ lines.append(f"[{entry['timestamp']}] [{entry['level']}] {entry['message']}")
+
+ return "\n".join(lines)
+
+
+if __name__ == "__main__":
+ logger = DebugLogger()
+ print(logger.get_logs())
\ No newline at end of file
diff --git a/backend/managers/chat_manager.py b/backend/managers/chat_manager.py
index a10faab..0b44bb8 100644
--- a/backend/managers/chat_manager.py
+++ b/backend/managers/chat_manager.py
@@ -5,6 +5,9 @@ from dotenv import load_dotenv
import requests
import json
+from backend.managers.debug_logger import get_logger
+logger = get_logger(__name__)
+
load_dotenv()
@@ -38,6 +41,7 @@ class ChatManager:
def clear_history(self) -> None:
"""Wipe the conversation history (starts a fresh chat)."""
+ logger.info("Chat history was cleared")
self.chat_history = []
def send_message(self, user_message: str) -> str:
@@ -49,25 +53,27 @@ class ChatManager:
# Add user message to history
self.add_message("user", user_message)
+ logger.info("Sending message to LLM API")
+
+ # Prepare request to OpenAI-compatible API
+ headers = {
+ "Content-Type": "application/json",
+ }
+
+ # Add API key if available
+ if self.api_key and self.api_key != "EMPTY":
+ headers["Authorization"] = f"Bearer {self.api_key}"
+
+ # Full history is sent so the model has multi-turn conversation context
+ payload = {
+ "model": self.model,
+ "messages": self.chat_history,
+ "temperature": 0.7,
+ "max_tokens": 2000,
+ "stream": False,
+ }
+
try:
- # Prepare request to OpenAI-compatible API
- headers = {
- "Content-Type": "application/json",
- }
-
- # Add API key if available
- if self.api_key and self.api_key != "EMPTY":
- headers["Authorization"] = f"Bearer {self.api_key}"
-
- # Full history is sent so the model has multi-turn conversation context
- payload = {
- "model": self.model,
- "messages": self.chat_history,
- "temperature": 0.7,
- "max_tokens": 2000,
- "stream": False,
- }
-
# Make API request
response = requests.post(
self.api_url, headers=headers, json=payload, timeout=30
@@ -75,9 +81,24 @@ class ChatManager:
# Check if request was successful
if response.status_code != 200:
- error_msg = f"API Error {response.status_code}: {response.text}"
- raise Exception(error_msg)
-
+ logger.warning("API HTTP status error %s: %s", response.status_code, response.text)
+ raise Exception(f"API Error {response.status_code}")
+
+ logger.info("Response recieved from API")
+
+ except requests.exceptions.Timeout as e:
+ error_msg = f"Timeout Error: {str(e)}"
+ self.add_message("assistant", f"Error: {error_msg}")
+ logger.exception("LLM API timeout: %s", e)
+ raise RuntimeError("LLM API timeout") from e
+
+ except requests.exceptions.RequestException as e:
+ error_msg = f"Connection Error: {str(e)}"
+ self.add_message("assistant", f"Error: {error_msg}")
+ logger.exception("LLM API connection failed: %s", e)
+ raise RuntimeError("LLM API connection failed") from e
+
+ try:
# Parse response
response_data = response.json()
@@ -88,23 +109,24 @@ class ChatManager:
# Add AI response to history
self.add_message("assistant", ai_message)
+ logger.info("Assistant response generated")
+
return ai_message
else:
+ logger.warning("Invalid API response format: %s", response_data)
raise Exception("Invalid API response format")
- except requests.exceptions.RequestException as e:
- error_msg = f"Connection Error: {str(e)}"
- # Add error message to history so user sees it
- self.add_message("assistant", f"Error: {error_msg}")
- raise Exception(error_msg)
+
except json.JSONDecodeError as e:
error_msg = f"JSON Decode Error: {str(e)}"
self.add_message("assistant", f"Error: {error_msg}")
+ logger.exception("JSON Decode Error: %s", e)
raise Exception(error_msg)
except Exception as e:
error_msg = f"Error: {str(e)}"
self.add_message("assistant", f"Error: {error_msg}")
- raise Exception(error_msg)
+ logger.exception("JSON parsing and message formatting failed: %s", e)
+ raise RuntimeError("JSON parsing and message formatting failed") from e
def get_chat_display(self) -> list:
"""Return a copy of the history suitable for display in the UI."""
diff --git a/backend/managers/debug_logger.py b/backend/managers/debug_logger.py
index b681f05..fd14ef9 100644
--- a/backend/managers/debug_logger.py
+++ b/backend/managers/debug_logger.py
@@ -1,66 +1,94 @@
-from datetime import datetime
+"""
+Central logging setup for the application.
+- Provides a unified logger via get_logger(__name__)
+- Writes all logs to a central rotating log file (logs/app.log)
+- Writes errors separately to logs/errors.log
+- Automatically includes the module name in each log entry
+- Supports standard logging levels: DEBUG, INFO, WARNING, ERROR, CRITICAL
+
+Usage:
+ from backend.managers.debug_logger import get_logger
+ logger = get_logger(__name__)
+
+ logger.info("Service started")
+ logger.debug("Debug details")
+ logger.error("Something went wrong")
+
+ try:
+ ...
+ except Exception:
+ logger.exception("Unexpected error")
+
+Logging levels (use consistently):
+DEBUG: Detailed technical info for developers (variables, flow, internal state).
+INFO: Normal application events (start/stop, successful operations, key milestones).
+WARNING: Something unexpected happened, but the program continues normally.
+ERROR: A specific operation failed, but the application is still running.
+CRITICAL: A severe failure that may stop the application or make it unusable.
+EXCEPTION: Same as ERROR, but used inside an `except` block and includes stacktrace
+ (via logger.exception()).
+"""
+
+import logging
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parent.parent
+LOG_DIR = BASE_DIR / "logs"
+LOG_DIR.mkdir(exist_ok=True)
class DebugLogger:
- """In-memory logger for code execution events.
- Collects timestamped INFO and ERROR entries during a single run.
- Call clear() before each new execution to start fresh.
- """
+ _initialized = False
- def __init__(self):
- self.logs: list[dict] = []
+ @classmethod
+ def setup(cls):
+ # prevents multiple setup
+ if cls._initialized:
+ return
- def log(self, message: str) -> None:
- """Append a general info message."""
- self.logs.append({
- "level": "INFO",
- "message": message,
- "timestamp": datetime.now().strftime("%H:%M:%S"),
- })
+ formatter = logging.Formatter(
+ "%(asctime)s [%(levelname)s] [%(name)s: Line %(lineno)d] %(message)s"
+ )
- def log_error(self, error_message: str) -> None:
- """Append an error message."""
- self.logs.append({
- "level": "ERROR",
- "message": error_message,
- "timestamp": datetime.now().strftime("%H:%M:%S"),
- })
+ # Main log file
+ file_handler = RotatingFileHandler(
+ LOG_DIR / "app.log",
+ maxBytes=5_000_000,
+ backupCount=5,
+ encoding="utf-8"
+ )
- def get_logs(self) -> list[dict]:
- """Return a copy of all collected log entries."""
- return list(self.logs)
+ file_handler.setFormatter(formatter)
- def clear(self) -> None:
- """Reset the log — call before each new execution."""
- self.logs = []
+ # Separate Error-Log
+ error_handler = RotatingFileHandler(
+ LOG_DIR / "errors.log",
+ maxBytes=5_000_000,
+ backupCount=3,
+ encoding="utf-8"
+ )
- def format_debug_output(self, output: dict) -> str:
- """Format an ExecutionEngine result dict into a human-readable string.
+ error_handler.setLevel(logging.ERROR)
+ error_handler.setFormatter(formatter)
- Args:
- output: dict with keys 'stdout', 'stderr', and 'rc'.
+ root_logger = logging.getLogger()
- Returns:
- A formatted string ready for display in the UI.
- """
- lines = []
+ root_logger.setLevel(logging.DEBUG)
- status = "SUCCESS" if output.get("rc") == 0 else "FAILED"
- lines.append(f"[{status}] Exit code: {output.get('rc')}")
+ root_logger.addHandler(file_handler)
+ root_logger.addHandler(error_handler)
+ #root_logger.propagate = False
- if output.get("stdout"):
- lines.append("\n--- stdout ---")
- lines.append(output["stdout"].rstrip())
+ cls._initialized = True
- if output.get("stderr"):
- lines.append("\n--- stderr ---")
- lines.append(output["stderr"].rstrip())
+ @classmethod
+ def get_logger(cls, name: str):
+ cls.setup()
+ return logging.getLogger(name)
- if not output.get("stdout") and not output.get("stderr"):
- lines.append("No output produced.")
- for entry in self.logs:
- lines.append(f"[{entry['timestamp']}] [{entry['level']}] {entry['message']}")
-
- return "\n".join(lines)
+# praktische shortcut function
+def get_logger(name: str):
+ return DebugLogger.get_logger(name)
diff --git a/backend/managers/execution_engine.py b/backend/managers/execution_engine.py
index b4a449a..8ac9f68 100644
--- a/backend/managers/execution_engine.py
+++ b/backend/managers/execution_engine.py
@@ -1,6 +1,9 @@
import subprocess
from pathlib import Path
+from backend.managers.debug_logger import get_logger
+logger = get_logger(__name__)
+
# Maximum time (seconds) a subprocess is allowed to run before being killed.
RUN_TIMEOUT = 30
@@ -41,6 +44,8 @@ class ExecutionEngine:
]
else:
return {"stdout": "", "stderr": f"Unsupported file type: {suffix}", "rc": 1}
+
+ logger.info("Running file %s with suffix %s", active_file.name, suffix)
try:
proc = subprocess.run(
@@ -50,12 +55,16 @@ class ExecutionEngine:
text=True,
timeout=RUN_TIMEOUT,
)
+ logger.info("File ran successfully.")
return {"stdout": proc.stdout, "stderr": proc.stderr, "rc": proc.returncode}
except subprocess.TimeoutExpired:
+ logger.warning("Time out afte %s s", RUN_TIMEOUT)
return {"stdout": "", "stderr": f"Timed out after {RUN_TIMEOUT}s", "rc": -1}
except FileNotFoundError as e:
# Raised when the interpreter/compiler binary is not found on PATH
+ logger.warning("Interpreter/compiler binary is not found on PATH: %s", e)
return {"stdout": "", "stderr": str(e), "rc": -1}
except Exception as e:
+ logger.exception("Error while running %s: %s", active_file.name, e)
return {"stdout": "", "stderr": str(e), "rc": -1}
diff --git a/backend/managers/file_manager.py b/backend/managers/file_manager.py
index add29f3..12d9e65 100644
--- a/backend/managers/file_manager.py
+++ b/backend/managers/file_manager.py
@@ -7,6 +7,9 @@ touching the filesystem, preventing path-traversal attacks.
import streamlit as st
from pathlib import Path
+from backend.managers.debug_logger import get_logger
+logger = get_logger(__name__)
+
# The workspace folder is created at module load so it always exists.
WORKSPACE = Path("workspace")
WORKSPACE.mkdir(exist_ok=True)
@@ -28,12 +31,16 @@ class FileManager:
Returns:
bool: True if folder was created successfully, False otherwise.
"""
+ logger.info("Creating folder at %s named %s", relative_path, name)
+
if not name:
+ logger.warning("Invalid folder name")
st.error(f"Invalid folder name: {name}")
return False
# Slashes in the name would silently create nested paths — reject them.
if "/" in name or "\\" in name:
+ logger.warning("'/' or '\\' in foldername not allowed")
st.error(f"Invalid folder name (no slashes allowed): {name}")
return False
@@ -52,11 +59,14 @@ class FileManager:
try:
folder_path.mkdir(exist_ok=False)
+ logger.info("Folder created successfully.")
return True
except FileExistsError:
+ logger.warning("Folder already exists.")
st.warning(f"Folder already exists: {relative_path}")
return False
except Exception as e:
+ logger.exception("Error creating folder %s: %s", relative_path, str(e))
st.error(f"Error creating folder {relative_path}: {str(e)}")
return False
@@ -71,14 +81,18 @@ class FileManager:
name (str): The name of the new file to create (should not contain slashes).
Returns:
bool: True if file was created successfully, False otherwise.
- """
+ """
+ logger.info("Creating file at %s named %s", relative_path, name)
+
if not name or name.strip() == "" :
+ logger.warning("Invalid folder name")
st.error(f"Invalid file name: {name}")
return False
name = Path(name)
if not name.suffix:
name = name.with_suffix(".txt") # Default to .txt if no extension provided
+ logger.info("No suffix was provided, creating .txt file")
if relative_path:
relative_path = Path(relative_path)
@@ -94,11 +108,14 @@ class FileManager:
try:
file_path.touch(exist_ok=False)
+ logger.info("File created successfully.")
return True
except FileExistsError:
+ logger.warning("Folder already exists.")
st.warning(f"File already exists: {relative_path}")
return False
except Exception as e:
+ logger.exception("Error creating file %s: %s", relative_path, str(e))
st.error(f"Error creating file {relative_path}: {str(e)}")
return False
@@ -113,27 +130,35 @@ class FileManager:
Returns:
str: The content of the file, or an empty string if there was an error.
"""
+ logger.info("Reading file at %s.", relative_path)
file_path = (relative_path).resolve()
if not file_path.exists():
st.error(f"File not found: {relative_path}")
+ logger.warning("Filepath does not exist.")
return ""
if not file_path.is_file():
st.error(f"Path is not a file: {relative_path}")
+ logger.warning("Path is not a file.")
return ""
# Ensure the resolved path is still inside the workspace (prevents path traversal).
if not str(file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}")
+ logger.warning("Access denied. File ist outside WORKSPACE")
return ""
try:
with open(file_path, "r") as f:
- return f.read()
+ content = f.read()
+ logger.info("File read successfully.")
+ return content
except FileNotFoundError:
st.error(f"File not found: {relative_path}")
+ logger.warning("File not found")
return ""
except Exception as e:
st.error(f"Error reading file {relative_path}: {str(e)}")
+ logger.exception("Error reading file at %s: %s", relative_path, e)
return ""
def save_file(self, relative_path: str, content: str) -> bool:
@@ -148,19 +173,24 @@ class FileManager:
Returns:
bool: True if save was successful, False otherwise.
"""
+ logger.info("Saving file at %s.", relative_path)
+
file_path = (Path(relative_path)).resolve()
# Ensure the resolved path is still inside the workspace (prevents path traversal).
if not str(file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}")
+ logger.warning("Access denied. File outside WORKSPACE.")
return False
try:
with open(file_path, "w") as f:
f.write(content)
+ logger.info("File written successfully.")
return True
except Exception as e:
st.error(f"Error saving file {relative_path}: {str(e)}")
+ logger.exception("Error saving file %s: %s", relative_path, e)
return False
def rename_file(self, old_relative_path: str, new_name: str) -> bool:
@@ -174,8 +204,11 @@ class FileManager:
Returns:
bool: True if rename was successful, False otherwise.
"""
+ logger.info("Rename file at %s to %s.", old_relative_path, new_name)
+
if not new_name or new_name.strip() == "":
st.error(f"Invalid file name: {new_name}")
+ logger.warning("New Name is empty.")
return False
file_type = Path(old_relative_path).suffix
@@ -191,13 +224,20 @@ class FileManager:
# Both old and new paths must stay inside the workspace.
if not str(old_file_path).startswith(str(self.base_path.resolve())) or not str(new_file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {old_relative_path}")
+ logger.warning("Access denied, file outside WORKSPACE.")
return False
try:
old_file_path.rename(new_file_path)
+ logger.info("Renamed successfully.")
return True
except FileNotFoundError:
st.error(f"File not found: {old_relative_path}")
+ logger.warning("Original file not found.")
+ return False
+ except Exception as e:
+ st.error(f"Error renaming file {old_relative_path} to {new_name}: {str(e)}")
+ logger.exception("Error deleting folder %s to %s: %s", old_relative_path, new_name, str(e))
return False
@@ -210,23 +250,29 @@ class FileManager:
Returns:
bool: True if deletion was successful, False otherwise.
"""
+ logger.info("Deleting folder %s.", relative_path)
+
folder_path = (self.base_path / relative_path).resolve()
# Ensure the resolved path is still inside the workspace (prevents path traversal).
if not str(folder_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}")
+ logger.warning("Access denied, folder outside WORKSPACE.")
return False
if not folder_path.exists():
st.error(f"Folder not found: {relative_path}")
+ logger.warning("Folder path not found.")
return False
try:
import shutil
shutil.rmtree(folder_path)
+ logger.info("Folder deleted successfully.")
return True
except Exception as e:
st.error(f"Error deleting folder {relative_path}: {str(e)}")
+ logger.exception("Error deleting folder %s: %s", relative_path, str(e))
return False
def delete_file(self, relative_path: str) -> bool:
@@ -238,21 +284,26 @@ class FileManager:
Returns:
bool: True if deletion was successful, False otherwise.
"""
+ logger.info("Deleting file %s.", relative_path)
file_path = Path(relative_path)
abs_file_path = (Path(self.base_path) / file_path).resolve()
if not str(abs_file_path).startswith(str(self.base_path.resolve())):
st.error(f"Access denied: {relative_path}")
+ logger.warning("Access denied, file outside WORKSPACE.")
return False
try:
abs_file_path.unlink()
+ logger.info("File deleted successfully.")
return True
except FileNotFoundError:
st.error(f"File not found: {relative_path}")
+ logger.warning("File not found")
return False
except Exception as e:
st.error(f"Error deleting file {relative_path}: {str(e)}")
+ logger.exception("Error deleting folder %s: %s", relative_path, str(e))
return False
def get_file_tree(self):
@@ -264,6 +315,8 @@ class FileManager:
Returns:
dict: A nested dictionary representing the file tree.
"""
+ logger.info("Getting file tree ...")
+
def build_tree(path: Path):
tree = {}
diff --git a/backend/managers/search_manager.py b/backend/managers/search_manager.py
index e69de29..7a11780 100644
--- a/backend/managers/search_manager.py
+++ b/backend/managers/search_manager.py
@@ -0,0 +1,2 @@
+from backend.managers.debug_logger import get_logger
+logger = get_logger(__name__)
\ No newline at end of file
diff --git a/backend/managers/system_prompter.py b/backend/managers/system_prompter.py
index d9610b7..46e09cd 100644
--- a/backend/managers/system_prompter.py
+++ b/backend/managers/system_prompter.py
@@ -1,5 +1,8 @@
"""Builds the system prompt that is sent to the AI at the start of each chat session."""
+from backend.managers.debug_logger import get_logger
+logger = get_logger(__name__)
+
# Prevents very large files from flooding the context window with tokens.
MAX_FILE_CHARS = 4000
@@ -22,6 +25,7 @@ class SystemPrompter:
Returns:
A ready-to-use system prompt string.
"""
+ logger.info("Generating system prompt.")
base = (
"You are an expert code assistant integrated into a lightweight code editor. "
"Help the user with code suggestions, debugging, explanations, and improvements. "
@@ -29,6 +33,7 @@ class SystemPrompter:
)
if file_context:
+ logger.info("Appending file context.")
name = file_context.get("name", "unknown")
content = file_context.get("content", "")
diff --git a/frontend/app.py b/frontend/app.py
index 140f7c9..4d9dc6c 100644
--- a/frontend/app.py
+++ b/frontend/app.py
@@ -17,6 +17,9 @@ from pathlib import Path
# where streamlit is launched from.
sys.path.insert(0, str(Path(__file__).parent.parent))
+from backend.managers.debug_logger import get_logger
+logger = get_logger(__name__)
+
from frontend.sidebar import render_sidebar
from frontend.editor import render_editor
from frontend.chat import render_chat
@@ -51,8 +54,10 @@ def main():
# Switch between the two main views based on the sidebar radio button
if st.session_state.get("radio_interface_options") == "Code Editor":
+ logger.info("Editor mode")
render_editor()
elif st.session_state.get("radio_interface_options") == "Chat with AI Assistant":
+ logger.info("Chat/Agent mode")
render_chat()
diff --git a/frontend/chat.py b/frontend/chat.py
index 43ab8bc..8457268 100644
--- a/frontend/chat.py
+++ b/frontend/chat.py
@@ -3,8 +3,13 @@
import streamlit as st
from backend.managers.chat_manager import ChatManager
from backend.managers.system_prompter import SystemPrompter
+from backend.agent.coding_agent import CodingAgent
+
+from backend.managers.debug_logger import get_logger
+logger = get_logger(__name__)
import asyncio
+import json
# ── Agent Mode helpers ────────────────────────────────────────────────────────
@@ -22,7 +27,7 @@ def _start_agent(task: str):
Stores the agent and its state in session_state so Streamlit can reference
them across reruns without losing progress.
"""
- from backend.agent.coding_agent import CodingAgent
+ logger.info("Starting coding agent.")
agent = CodingAgent()
agent.start_task(task)
action = _run_async(agent.propose_next_action())
@@ -38,6 +43,7 @@ def _approve_action():
pending = st.session_state.agent_pending_action
result = _run_async(agent.approve())
+ logger.info("Approve action and propose next step.")
# Append a record to the log so the user can review every completed step.
st.session_state.agent_log.append({
@@ -62,6 +68,7 @@ def _reject_action(feedback: str):
The pending action is discarded; the agent receives the user's feedback and
proposes a different approach on the next call to propose_next_action().
"""
+ logger.info("Rejecting proposed action.")
agent = st.session_state.coding_agent
agent.reject(feedback or "Please try a different approach.")
next_action = _run_async(agent.propose_next_action())
@@ -71,6 +78,7 @@ def _reject_action(feedback: str):
def _followup_agent(question: str):
"""Continue a finished task by injecting a follow-up question and resuming the loop."""
+ logger.info("Asking follow up question")
agent = st.session_state.coding_agent
agent.follow_up(question)
action = _run_async(agent.propose_next_action())
@@ -80,6 +88,7 @@ def _followup_agent(question: str):
def _reset_agent():
"""Clear all agent state and return to the idle (task input) screen."""
+ logger.info("Resetting Agent")
st.session_state.coding_agent = None
st.session_state.agent_status = "idle"
st.session_state.agent_log = []
@@ -88,6 +97,62 @@ def _reset_agent():
# ── Agent Mode UI ─────────────────────────────────────────────────────────────
+def _render_arguments(args: dict):
+ if not args:
+ return
+
+ with st.expander("📦 Arguments", expanded=False):
+
+ if args.get("path"):
+ st.markdown("##### 📁 Path")
+ st.code(args["path"])
+
+ if args.get("dir_path"):
+ st.markdown("##### 🌳 Directory")
+ st.code(args["dir_path"])
+
+ if args.get("query"):
+ st.markdown("##### 🔎 Query")
+ st.code(args["query"])
+
+ if args.get("url"):
+ st.markdown("##### 🌐 URL")
+ st.code(args["url"])
+
+ if args.get("content"):
+ st.markdown("##### 📝 Content")
+ st.code(args["content"])
+
+ if args.get("code"):
+ st.markdown("##### 🐍 Python Code")
+ st.code(args["code"], language="python")
+
+ if args.get("max_results") is not None:
+ st.markdown("##### 📊 Max Results")
+ st.code(str(args["max_results"]))
+
+ known_keys = {
+ "path",
+ "dir_path",
+ "query",
+ "content",
+ "url",
+ "code",
+ "max_results",
+ }
+
+ extra_args = {
+ k: v for k, v in args.items()
+ if k not in known_keys
+ }
+
+ if extra_args:
+ st.markdown("##### ⚙️ Other")
+ st.code(
+ json.dumps(extra_args, indent=2),
+ language="json"
+ )
+
def render_agent_mode():
"""Render the step-by-step agent UI.
@@ -96,6 +161,7 @@ def render_agent_mode():
- "waiting_approval" → show proposed action, Approve / Reject / Abort
- "done" → success message, follow-up input, New Task button
"""
+ logger.info("Agent mode.")
# The toggle must always render so Streamlit keeps agent_mode=True in session_state.
st.toggle("Agent Mode", key="agent_mode")
@@ -110,8 +176,8 @@ def render_agent_mode():
with st.chat_message("assistant"):
st.markdown(f"**Step {i + 1} — `{step['tool']}`**")
st.caption(f"Thought: {step['thought']}")
- if step.get("arguments"):
- st.json(step["arguments"])
+ if step.get("arguments"):
+ _render_arguments(step["arguments"])
result_text = step.get("result", "")
# Colour the result based on whether the tool succeeded or failed.
if result_text.startswith("ERROR") or result_text.startswith("SYNTAX ERROR"):
@@ -130,8 +196,6 @@ def render_agent_mode():
placeholder="e.g. Write a function that sorts a list and saves it to sorted.py",
)
if st.button("Start Agent", type="primary", use_container_width=True):
- #loop = asyncio.new_event_loop()
- #asyncio.set_event_loop(loop)
if task.strip():
with st.spinner("Agent is thinking..."):
_start_agent(task.strip())
@@ -149,15 +213,7 @@ def render_agent_mode():
args = pending.get("arguments", {})
if args:
- # Show file content separately as a code block for readability;
- # other arguments are displayed as JSON.
- if "content" in args:
- display_args = {k: v for k, v in args.items() if k != "content"}
- if display_args:
- st.json(display_args)
- st.code(args["content"], language="python")
- else:
- st.json(args)
+ _render_arguments(args)
feedback = st.text_input(
"Rejection feedback (optional):",
@@ -219,6 +275,7 @@ def render_normal_chat():
Each subsequent message appends to the same conversation so the AI retains
full context throughout the session.
"""
+ logger.info("Chat mode")
# Replay the conversation history as chat bubbles (skip system messages).
for message in st.session_state.chat_history:
role = message["role"]
diff --git a/frontend/editor.py b/frontend/editor.py
index 9f9c8b8..35d5a40 100644
--- a/frontend/editor.py
+++ b/frontend/editor.py
@@ -6,7 +6,8 @@ from pathlib import Path
from backend.managers.file_manager import FileManager
from backend.managers.execution_engine import ExecutionEngine
-from backend.managers.debug_logger import DebugLogger
+from backend.managers.debug_logger import get_logger
+logger = get_logger(__name__)
# Maps file extensions to Ace editor language modes for syntax highlighting.
LANG_MAP = {
@@ -51,8 +52,10 @@ def _rename_dialog(file_path: str):
if st.session_state.active_file == file_path:
st.session_state.active_file = new_file_path
st.rerun()
+ logger.info("Rename file %s to %s successfull", file_path, new_name )
else:
- st.error("Rename failed. Check that the file still exists.")
+ logger.warning("Rename failed.")
+ st.error("Rename failed. Check that the file %s still exists.", file_path)
with col2:
if st.button("Cancel", use_container_width=True):
st.rerun()
@@ -77,8 +80,10 @@ def _delete_dialog(abs_file_path: str):
if st.session_state.open_files else None
)
st.rerun()
+ logger.info("Deleting file %s successfull.", abs_file_path)
else:
st.error("Delete failed. Check that the file still exists.")
+ logger.warning("Deleting file %s failed.", abs_file_path)
with col2:
if st.button("Cancel", use_container_width=True):
st.rerun()
@@ -96,18 +101,16 @@ def run_active_file():
return
execution_engine = ExecutionEngine()
- debug_logger = DebugLogger()
-
- debug_logger.clear()
- debug_logger.log(f"Executing code from {active_file}...")
+
+ logger.info("Executing code from %s...", active_file)
with st.spinner(f"Running {Path(active_file).name}..."):
output = execution_engine.run_code(Path(active_file))
if output["rc"] == 0:
- debug_logger.log("Execution completed successfully.")
+ logger.info("Execution completed successfully.")
else:
- debug_logger.log_error(f"Execution failed with exit code {output['rc']}.")
+ logger.error("Execution failed with exit code %s.", output['rc'])
st.session_state.code_execution_output = {
"stdout": output["stdout"],
diff --git a/requirements.txt b/requirements.txt
index 3916858..acaf29e 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -25,5 +25,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
+#MCP-Code execution tools
+pyflakes>=0.1.0
+pygame>=0.1.0
\ No newline at end of file
diff --git a/tests/test_debug_logger.py b/tests/test_debug_logger.py
index 1ee6eca..d0de98a 100644
--- a/tests/test_debug_logger.py
+++ b/tests/test_debug_logger.py
@@ -6,7 +6,7 @@ import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
-from backend.managers.debug_logger import DebugLogger
+from backend.managers._debug_logger import DebugLogger
# ── Fixtures ──────────────────────────────────────────────────────────────────
diff --git a/tests/test_execution_engine.py b/tests/test_execution_engine.py
index e69de29..14e32ce 100644
--- a/tests/test_execution_engine.py
+++ b/tests/test_execution_engine.py
@@ -0,0 +1,428 @@
+import pytest
+import subprocess
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from backend.managers.execution_engine import ExecutionEngine
+
+
+# =========================================================
+# FIXTURE
+# =========================================================
+
+@pytest.fixture()
+def engine():
+ return ExecutionEngine()
+
+
+# =========================================================
+# BASIC TESTS (1–10)
+# =========================================================
+
+
+# ---------------------------------------------------------
+# 1. Python-Datei wird korrekt ausgeführt
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_python_file_success(mock_run, engine, tmp_path):
+ file = tmp_path / "test.py"
+ file.write_text("print('hello')")
+
+ mock_run.return_value = Mock(
+ stdout="hello\n",
+ stderr="",
+ returncode=0
+ )
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == 0
+ assert "hello" in result["stdout"]
+
+
+# ---------------------------------------------------------
+# 2. Python-Datei mit Fehler
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_python_file_error(mock_run, engine, tmp_path):
+ file = tmp_path / "broken.py"
+ file.write_text("1/0")
+
+ mock_run.return_value = Mock(
+ stdout="",
+ stderr="ZeroDivisionError",
+ returncode=1
+ )
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == 1
+ assert "ZeroDivisionError" in result["stderr"]
+
+
+# ---------------------------------------------------------
+# 3. LaTeX-Datei wird kompiliert
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_tex_file_success(mock_run, engine, tmp_path):
+ file = tmp_path / "doc.tex"
+ file.write_text("\\documentclass{article}")
+
+ mock_run.return_value = Mock(
+ stdout="PDF created",
+ stderr="",
+ returncode=0
+ )
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == 0
+ assert "PDF created" in result["stdout"]
+
+
+# ---------------------------------------------------------
+# 4. Unsupported File Type
+# ---------------------------------------------------------
+
+def test_run_unsupported_file(engine, tmp_path):
+ file = tmp_path / "test.js"
+ file.write_text("console.log('x')")
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == 1
+ assert "Unsupported file type" in result["stderr"]
+
+
+# ---------------------------------------------------------
+# 5. Timeout wird behandelt
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_timeout(mock_run, engine, tmp_path):
+ file = tmp_path / "slow.py"
+ file.write_text("while True: pass")
+
+ mock_run.side_effect = subprocess.TimeoutExpired(
+ cmd=["py"],
+ timeout=30
+ )
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == -1
+ assert "Timed out" in result["stderr"]
+
+
+# ---------------------------------------------------------
+# 6. Fehlender Interpreter
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_missing_interpreter(mock_run, engine, tmp_path):
+ file = tmp_path / "test.py"
+ file.write_text("print(1)")
+
+ mock_run.side_effect = FileNotFoundError("py not found")
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == -1
+ assert "py not found" in result["stderr"]
+
+
+# ---------------------------------------------------------
+# 7. Allgemeine Exception
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_general_exception(mock_run, engine, tmp_path):
+ file = tmp_path / "test.py"
+ file.write_text("print(1)")
+
+ mock_run.side_effect = RuntimeError("unexpected")
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == -1
+ assert "unexpected" in result["stderr"]
+
+
+# ---------------------------------------------------------
+# 8. subprocess.run wird mit cwd ausgeführt
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_uses_correct_cwd(mock_run, engine, tmp_path):
+ folder = tmp_path / "project"
+ folder.mkdir()
+
+ file = folder / "main.py"
+ file.write_text("print(1)")
+
+ mock_run.return_value = Mock(
+ stdout="",
+ stderr="",
+ returncode=0
+ )
+
+ engine.run_code(file)
+
+ _, kwargs = mock_run.call_args
+
+ assert kwargs["cwd"] == folder.resolve()
+
+
+# ---------------------------------------------------------
+# 9. subprocess.run nutzt capture_output
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_capture_output_enabled(mock_run, engine, tmp_path):
+ file = tmp_path / "test.py"
+ file.write_text("print(1)")
+
+ mock_run.return_value = Mock(
+ stdout="",
+ stderr="",
+ returncode=0
+ )
+
+ engine.run_code(file)
+
+ _, kwargs = mock_run.call_args
+
+ assert kwargs["capture_output"] is True
+
+
+# ---------------------------------------------------------
+# 10. subprocess.run nutzt text=True
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_text_mode_enabled(mock_run, engine, tmp_path):
+ file = tmp_path / "test.py"
+ file.write_text("print(1)")
+
+ mock_run.return_value = Mock(
+ stdout="",
+ stderr="",
+ returncode=0
+ )
+
+ engine.run_code(file)
+
+ _, kwargs = mock_run.call_args
+
+ assert kwargs["text"] is True
+
+
+# =========================================================
+# EDGE CASE TESTS (11–20)
+# =========================================================
+
+
+# ---------------------------------------------------------
+# 11. Unicode Output
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_unicode_output(mock_run, engine, tmp_path):
+ file = tmp_path / "unicode.py"
+ file.write_text("print('🔥 Grüezi 世界')")
+
+ mock_run.return_value = Mock(
+ stdout="🔥 Grüezi 世界\n",
+ stderr="",
+ returncode=0
+ )
+
+ result = engine.run_code(file)
+
+ assert "🔥 Grüezi 世界" in result["stdout"]
+
+
+# ---------------------------------------------------------
+# 12. Leerer stdout/stderr
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_empty_output(mock_run, engine, tmp_path):
+ file = tmp_path / "empty.py"
+ file.write_text("x = 1")
+
+ mock_run.return_value = Mock(
+ stdout="",
+ stderr="",
+ returncode=0
+ )
+
+ result = engine.run_code(file)
+
+ assert result["stdout"] == ""
+ assert result["stderr"] == ""
+
+
+# ---------------------------------------------------------
+# 13. Sehr langer stdout
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_large_output(mock_run, engine, tmp_path):
+ file = tmp_path / "large.py"
+ file.write_text("print('A')")
+
+ mock_run.return_value = Mock(
+ stdout="A" * 100000,
+ stderr="",
+ returncode=0
+ )
+
+ result = engine.run_code(file)
+
+ assert len(result["stdout"]) == 100000
+
+
+# ---------------------------------------------------------
+# 14. Dateiname mit Leerzeichen
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_filename_with_spaces(mock_run, engine, tmp_path):
+ file = tmp_path / "my script.py"
+ file.write_text("print(1)")
+
+ mock_run.return_value = Mock(
+ stdout="ok",
+ stderr="",
+ returncode=0
+ )
+
+ engine.run_code(file)
+
+ args, _ = mock_run.call_args
+
+ assert "my script.py" in args[0]
+
+
+# ---------------------------------------------------------
+# 15. Dateiname mit Unicode
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_unicode_filename(mock_run, engine, tmp_path):
+ file = tmp_path / "🔥_test.py"
+ file.write_text("print(1)")
+
+ mock_run.return_value = Mock(
+ stdout="ok",
+ stderr="",
+ returncode=0
+ )
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == 0
+
+
+# ---------------------------------------------------------
+# 16. .tex nutzt pdflatex
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_tex_uses_pdflatex(mock_run, engine, tmp_path):
+ file = tmp_path / "doc.tex"
+ file.write_text("x")
+
+ mock_run.return_value = Mock(
+ stdout="",
+ stderr="",
+ returncode=0
+ )
+
+ engine.run_code(file)
+
+ args, _ = mock_run.call_args
+
+ assert args[0][0] == "pdflatex"
+
+
+# ---------------------------------------------------------
+# 17. .py nutzt py Interpreter
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_python_uses_py_interpreter(mock_run, engine, tmp_path):
+ file = tmp_path / "main.py"
+ file.write_text("print(1)")
+
+ mock_run.return_value = Mock(
+ stdout="",
+ stderr="",
+ returncode=0
+ )
+
+ engine.run_code(file)
+
+ args, _ = mock_run.call_args
+
+ assert args[0][0] == "py"
+
+
+# ---------------------------------------------------------
+# 18. Relative Pfade funktionieren
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_relative_paths(mock_run, engine, tmp_path):
+ sub = tmp_path / "src"
+ sub.mkdir()
+
+ file = sub / "main.py"
+ file.write_text("print(1)")
+
+ mock_run.return_value = Mock(
+ stdout="ok",
+ stderr="",
+ returncode=0
+ )
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == 0
+
+
+# ---------------------------------------------------------
+# 19. Großgeschriebenes Suffix blockiert
+# ---------------------------------------------------------
+
+def test_uppercase_suffix_not_supported(engine, tmp_path):
+ file = tmp_path / "SCRIPT.PY"
+ file.write_text("print(1)")
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == 1
+
+
+# ---------------------------------------------------------
+# 20. Leere Datei ausführen
+# ---------------------------------------------------------
+
+@patch("subprocess.run")
+def test_run_empty_file(mock_run, engine, tmp_path):
+ file = tmp_path / "empty.py"
+ file.write_text("")
+
+ mock_run.return_value = Mock(
+ stdout="",
+ stderr="",
+ returncode=0
+ )
+
+ result = engine.run_code(file)
+
+ assert result["rc"] == 0
\ No newline at end of file
diff --git a/tests/test_mcp_server_code_execution.py b/tests/test_mcp_server_code_execution.py
index e69de29..8bef986 100644
--- a/tests/test_mcp_server_code_execution.py
+++ b/tests/test_mcp_server_code_execution.py
@@ -0,0 +1,269 @@
+import pytest
+from unittest.mock import Mock, patch
+import subprocess
+
+from backend.agent.servers import mcp_server_code_execution as server
+
+
+# =========================================================
+# BASIC TESTS (1–10)
+# =========================================================
+
+
+# ---------------------------------------------------------
+# 1. Erlaubter Code besteht Safety Check
+# ---------------------------------------------------------
+
+def test_check_code_safety_valid():
+ code = "print('hello')"
+
+ result = server.check_code_safety(code)
+
+ assert result is None
+
+
+# ---------------------------------------------------------
+# 2. Blockierter Import wird erkannt
+# ---------------------------------------------------------
+
+def test_check_code_safety_blocked_import():
+ code = "import os"
+
+ result = server.check_code_safety(code)
+
+ assert "Blocked import" in result
+
+
+# ---------------------------------------------------------
+# 3. Blockierter Builtin wird erkannt
+# ---------------------------------------------------------
+
+def test_check_code_safety_blocked_builtin():
+ code = "eval('2+2')"
+
+ result = server.check_code_safety(code)
+
+ assert "Blocked builtin" in result
+
+
+# ---------------------------------------------------------
+# 4. analyse_structure erkennt Funktionen
+# ---------------------------------------------------------
+
+def test_analyse_structure_function():
+ code = """
+def hello(name):
+ return name
+"""
+
+ result = server.analyse_structure(code)
+
+ assert "def hello(name)" in result
+
+
+# ---------------------------------------------------------
+# 5. analyse_structure erkennt Klassen
+# ---------------------------------------------------------
+
+def test_analyse_structure_class():
+ code = """
+class User:
+ def login(self):
+ pass
+"""
+
+ result = server.analyse_structure(code)
+
+ assert "class User" in result
+ assert "method: login" in result
+
+
+# ---------------------------------------------------------
+# 6. lint_code erkennt Undefined Variable
+# ---------------------------------------------------------
+
+def test_lint_code_undefined_variable():
+ code = "print(x)"
+
+ result = server.lint_code(code)
+
+ assert "undefined name 'x'" in result.lower()
+
+
+# ---------------------------------------------------------
+# 7. lint_code erkennt sauberen Code
+# ---------------------------------------------------------
+
+def test_lint_code_clean():
+ code = """
+x = 1
+print(x)
+"""
+
+ result = server.lint_code(code)
+
+ assert "No issues found" in result
+
+
+# ---------------------------------------------------------
+# 8. python_code_validation validiert sicheren Code
+# ---------------------------------------------------------
+
+def test_python_code_validation_safe():
+ code = "print('safe')"
+
+ result = server.python_code_validation(code)
+
+ assert "can be executed" in result
+
+
+# ---------------------------------------------------------
+# 9. run_python_sandboxed führt Code aus
+# ---------------------------------------------------------
+
+def test_run_python_sandboxed_success():
+ code = "print('hello world')"
+
+ result = server.run_python_sandboxed(code)
+
+ assert "hello world" in result
+
+
+# ---------------------------------------------------------
+# 10. run_python_sandboxed ohne Output
+# ---------------------------------------------------------
+
+def test_run_python_sandboxed_no_output():
+ code = "x = 5"
+
+ result = server.run_python_sandboxed(code)
+
+ assert "no output" in result.lower()
+
+
+# =========================================================
+# EDGE CASE TESTS (11–20)
+# =========================================================
+
+
+# ---------------------------------------------------------
+# 11. Syntaxfehler erkennen
+# ---------------------------------------------------------
+
+def test_check_code_safety_syntax_error():
+ code = "def broken("
+
+ result = server.check_code_safety(code)
+
+ assert "SyntaxError" in result
+
+
+# ---------------------------------------------------------
+# 12. ImportFrom blockieren
+# ---------------------------------------------------------
+
+def test_check_code_safety_import_from():
+ code = "from os import path"
+
+ result = server.check_code_safety(code)
+
+ assert "Blocked import" in result
+
+
+# ---------------------------------------------------------
+# 13. Gefährliche Path-Sequenzen erkennen
+# ---------------------------------------------------------
+
+def test_check_code_safety_path_traversal():
+ code = "print('../etc/passwd')"
+
+ result = server.check_code_safety(code)
+
+ assert "Suspect path sequence" in result
+
+
+# ---------------------------------------------------------
+# 14. __import__ erkennen
+# ---------------------------------------------------------
+
+def test_check_code_safety_import_escape():
+ code = "__import__('os')"
+
+ result = server.check_code_safety(code)
+
+ assert "Blocked" in result
+
+
+# ---------------------------------------------------------
+# 15. subprocess Escape erkennen
+# ---------------------------------------------------------
+
+def test_check_code_safety_subprocess_escape():
+ code = "subprocess.run(['ls'])"
+
+ result = server.check_code_safety(code)
+
+ assert "Suspect path sequence" in result
+
+
+# ---------------------------------------------------------
+# 16. Endlosschleife Timeout
+# ---------------------------------------------------------
+
+def test_run_python_sandboxed_timeout():
+ code = """
+while True:
+ pass
+"""
+
+ result = server.run_python_sandboxed(code)
+
+ assert "time limit" in result.lower()
+
+
+# ---------------------------------------------------------
+# 17. Sehr großer Output wird gekürzt
+# ---------------------------------------------------------
+
+def test_run_python_sandboxed_large_output():
+ code = "print('A' * 10000)"
+
+ result = server.run_python_sandboxed(code)
+
+ assert "truncated" in result.lower()
+
+
+# ---------------------------------------------------------
+# 18. Unicode Output funktioniert
+# ---------------------------------------------------------
+
+def test_run_python_sandboxed_unicode():
+ code = "print('🔥 Grüezi 世界')"
+
+ result = server.run_python_sandboxed(code)
+
+ assert "🔥 Grüezi 世界" in result
+
+
+# ---------------------------------------------------------
+# 19. analyse_structure bei leerem Code
+# ---------------------------------------------------------
+
+def test_analyse_structure_empty():
+ code = ""
+
+ result = server.analyse_structure(code)
+
+ assert "No top-level imports" in result
+
+
+# ---------------------------------------------------------
+# 20. Sandbox behandelt Runtime Errors
+# ---------------------------------------------------------
+
+def test_run_python_sandboxed_runtime_error():
+ code = "1 / 0"
+
+ result = server.run_python_sandboxed(code)
+
+ assert "ZeroDivisionError" in result
\ No newline at end of file