diff --git a/backend/agent/coding_agent.py b/backend/agent/coding_agent.py index f2d8898..12952e8 100644 --- a/backend/agent/coding_agent.py +++ b/backend/agent/coding_agent.py @@ -13,16 +13,23 @@ step-by-step methods so Streamlit can drive the loop via session_state: agent.reject(feedback) # skip action, inject user feedback """ -import ast -import inspect import json import os -import subprocess -import sys +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 + +# ── mcp server initialization ──────────────────────────────────────────────────────────────── +adapter = MCPToolAdapter() +print("MCPToolAdapter created. Listing all tools from servers...") +asyncio.run(adapter.initialize_all_servers()) +print("listed tools from all servers") load_dotenv() @@ -38,179 +45,71 @@ MAX_HISTORY_CHARS = 80_000 # ═════════════════════════════════════════════════════════════════════════════ -# PART A – TOOL FUNCTIONS +# Tool dispatching and result handling # ═════════════════════════════════════════════════════════════════════════════ -# -# Each tool is a plain Python function decorated with @register_tool. -# The decorator adds the function to TOOL_REGISTRY so the dispatcher -# can call it by name at runtime. -TOOL_REGISTRY: dict[str, callable] = {} +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.") -def register_tool(func): - """Decorator – adds a function to the global tool registry.""" - TOOL_REGISTRY[func.__name__] = func - return func + descriptions = [] + for tool in all_tools: + pprint.pprint(f"{tool}") + descriptions.append(f"- {tool['tool_name']}: {tool['tool_description']}") + + return "\n".join(descriptions) +async def dispatch_tool(tool_name: str, arguments: dict) -> str: + """Call a tool by name with the given arguments using the MCP adapter.""" -@register_tool -def read_file(path: str) -> str: - """Read a .py or .txt file from the workspace and return its contents.""" - target = (WORKSPACE / path).resolve() - if not str(target).startswith(str(WORKSPACE.resolve())): - return "ERROR: path is outside the workspace." - if not target.exists(): - return f"ERROR: file '{path}' not found." - if target.suffix not in (".py", ".txt"): - return f"ERROR: can only read .py and .txt files, got '{target.suffix}'." - return target.read_text() - - -@register_tool -def write_file(path: str, content: str) -> str: - """Write content to a .py or .txt file in the workspace.""" - target = (WORKSPACE / path).resolve() - if not str(target).startswith(str(WORKSPACE.resolve())): - return "ERROR: path is outside the workspace." - if target.suffix not in (".py", ".txt"): - return f"ERROR: can only write .py and .txt files, got '{target.suffix}'." - target.parent.mkdir(parents=True, exist_ok=True) - target.write_text(content) - return f"OK: wrote {len(content)} chars to {path}." - - -@register_tool -def list_files(file_glob: str = "*") -> str: - """List files in the workspace matching the glob pattern.""" - found = sorted(WORKSPACE.glob(file_glob)) - found = [f.relative_to(WORKSPACE) for f in found if f.is_file()] - if not found: - return f"No files matching '{file_glob}' in workspace." - return "\n".join(str(f) for f in found) - - -@register_tool -def grep_search(pattern: str, file_glob: str = "*.py") -> str: - """Search for a pattern in workspace files and return matching lines with line numbers.""" - matches = [] - for filepath in sorted(WORKSPACE.glob(file_glob)): - if filepath.suffix not in (".py", ".txt"): - continue - try: - lines = filepath.read_text().splitlines() - except Exception: - continue - for i, line in enumerate(lines, 1): - if pattern in line: - rel = filepath.relative_to(WORKSPACE) - matches.append(f"{rel}:{i}: {line}") - if not matches: - return f"No matches for '{pattern}' in {file_glob}." - return "\n".join(matches) - - -@register_tool -def run_python(path: str) -> str: - """Execute a Python file in the workspace and return stdout and stderr.""" - target = (WORKSPACE / path).resolve() - if not str(target).startswith(str(WORKSPACE.resolve())): - return "ERROR: path is outside the workspace." - if not target.exists(): - return f"ERROR: file '{path}' not found." - result = subprocess.run( - [sys.executable, str(target)], - capture_output=True, text=True, timeout=30, - cwd=str(WORKSPACE), - ) - output = "" - if result.stdout: - output += f"STDOUT:\n{result.stdout}" - if result.stderr: - output += f"STDERR:\n{result.stderr}" - output += f"\nExit code: {result.returncode}" - return output.strip() - - -@register_tool -def validate_python(path: str) -> str: - """Check whether a Python file has valid syntax using ast.parse.""" - target = (WORKSPACE / path).resolve() - if not str(target).startswith(str(WORKSPACE.resolve())): - return "ERROR: path is outside the workspace." - if not target.exists(): - return f"ERROR: file '{path}' not found." - source = target.read_text() + if tool_name == "done": + # Handle the "done" tool locally since it's not an MCP tool + summary = arguments.get("summary", "Task completed.") + return f"DONE: {summary}" + try: - ast.parse(source) - return "OK: syntax is valid." - except SyntaxError as e: - return f"SYNTAX ERROR: {e}" + print(f"Trying to call tool '{tool_name}' in dispatch_tool through MCPToolAdapter...") + result = await adapter.call_tool(tool_name, arguments) + print(f"Raw result from tool '{tool_name}': {result}") -@register_tool -def done(summary: str) -> str: - """Signal that the agent has finished its task.""" - return f"DONE: {summary}" - - -# ═════════════════════════════════════════════════════════════════════════════ -# PART B – TOOL DISPATCHER -# ═════════════════════════════════════════════════════════════════════════════ - -def build_tool_description() -> str: - """Auto-generate tool descriptions from function signatures and docstrings.""" - lines = [] - for name, func in TOOL_REGISTRY.items(): - sig = inspect.signature(func) - params = [] - for pname, param in sig.parameters.items(): - if param.default is inspect.Parameter.empty: - params.append(f'"{pname}": ""') - else: - params.append(f'"{pname}": ""') - param_str = ", ".join(params) - doc = (func.__doc__ or "").strip().split("\n")[0] - lines.append(f" - {name}({{{param_str}}}): {doc}") - return "\n".join(lines) - - -def dispatch_tool(tool_name: str, arguments: dict) -> str: - """Call a tool by name with the given arguments.""" - if tool_name not in TOOL_REGISTRY: - return f"ERROR: unknown tool '{tool_name}'. Available: {list(TOOL_REGISTRY.keys())}" - func = TOOL_REGISTRY[tool_name] - try: - return func(**arguments) - except TypeError as e: - return f"ERROR calling {tool_name}: {e}" + if result.isError: + texts = [block.text for block in result.content if block.type == "text"] + 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: - return f"ERROR in {tool_name}: {type(e).__name__}: {e}" - + return f"Error calling tool '{tool_name}': {e}" # ═════════════════════════════════════════════════════════════════════════════ -# PART C – SYSTEM PROMPT +# SYSTEM PROMPT # ═════════════════════════════════════════════════════════════════════════════ SYSTEM_PROMPT = f"""\ You are a coding agent that helps users with Python programming tasks. You work inside a workspace directory and have access to tools. +You solve tasks by interacting with a workspace through a +dynamic set of tools via the Model Context Protocol (MCP). - -You can: - - Read .py and .txt files from the workspace - - Write .py and .txt files to the workspace - - List files in the workspace - - Search for patterns in files using grep - - Execute Python files and see their output - - Validate Python syntax using ast.parse - - Signal completion when the task is done + +You have access to a workspace where you can manage files, +analyze code, and execute Python scripts. +You can call tools to interact with the workspace and get feedback. +You can write and read files, list directory contents, search for patterns, +validate Python syntax, and run Python code. +You can access web search and page fetching tools to gather information from the internet. +You can use these capabilities to iteratively work towards completing the user's task. -{build_tool_description()} +{build_all_tool_description()} + For every user request, follow this workflow: 1. PLAN: Think about what steps are needed. List them in "thought". @@ -252,7 +151,7 @@ Example: # ═════════════════════════════════════════════════════════════════════════════ -# PART D – CODING AGENT CLASS +# CODING AGENT CLASS # ═════════════════════════════════════════════════════════════════════════════ def truncate_result(result: str) -> str: @@ -293,9 +192,100 @@ def trim_messages(messages: list) -> list: } return head + [reminder] + tail +def _repair_json_strings(text: str) -> str: + """ + Replace unescaped control characters (newline, tab, carriage return) + inside JSON string values with their proper escape sequences. + + LLMs frequently emit literal newlines inside long string values, which + is invalid JSON. This function fixes that without touching structural + whitespace outside strings. + """ + result: list[str] = [] + in_string = False + escape = False + _escapes = {'\n': '\\n', '\r': '\\r', '\t': '\\t'} + for ch in text: + if escape: + result.append(ch) + escape = False + continue + if ch == '\\' and in_string: + result.append(ch) + escape = True + continue + if ch == '"': + in_string = not in_string + result.append(ch) + continue + if in_string and ch in _escapes: + result.append(_escapes[ch]) + continue + result.append(ch) + return ''.join(result) + + +def extract_json(text: str) -> str: + """ + Extract and repair a JSON object or array from an LLM response that may + contain extra prose, markdown code fences, or unescaped control characters. + + Strategy: + 1. Strip markdown ```json ... ``` or ``` ... ``` fences. + 2. Find the first '{' or '[' and extract to the matching closing bracket. + 3. Repair unescaped newlines/tabs inside string values. + + Returns the cleaned JSON string, or the original text as a fallback + (so json.loads can raise a meaningful error with context). + """ + + if text is None: + return "" + + # 1. Strip markdown fences + fenced = re.sub(r"```(?:json)?\s*([\s\S]*?)\s*```", r"\1", text.strip()) + if fenced != text.strip(): + return _repair_json_strings(fenced.strip()) + + # 2. Find first JSON container and extract to matching close + extracted = text + for start_char, end_char in [('{', '}'), ('[', ']')]: + idx = text.find(start_char) + if idx == -1: + continue + depth = 0 + in_string = False + escape = False + for i, ch in enumerate(text[idx:], start=idx): + if escape: + escape = False + continue + if ch == '\\' and in_string: + escape = True + continue + if ch == '"': + in_string = not in_string + continue + if in_string: + continue + if ch == start_char: + depth += 1 + elif ch == end_char: + depth -= 1 + if depth == 0: + extracted = text[idx: i + 1] + break + break + + # 3. Repair unescaped control characters inside string values + return _repair_json_strings(extracted) + def _strip_code_fences(text: str) -> str: """Remove markdown code fences (```json ... ```) from a string.""" + if text is None: + return "" + text = text.strip() if text.startswith("```"): lines = text.split("\n") @@ -335,8 +325,11 @@ class CodingAgent: self.api_key = os.getenv("API_KEY") self.model = os.getenv("MODEL") + #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"} if self.api_key and self.api_key != "EMPTY": headers["Authorization"] = f"Bearer {self.api_key}" @@ -348,7 +341,7 @@ class CodingAgent: "max_tokens": 4096, "stream": False, } - + response = requests.post(self.api_url, headers=headers, json=payload, timeout=60) if response.status_code != 200: @@ -371,7 +364,7 @@ class CodingAgent: self.is_done = False self.iteration = 0 - def propose_next_action(self) -> dict: + async def propose_next_action(self) -> dict: """Ask the LLM what to do next. Returns the parsed action dict without executing anything. @@ -387,14 +380,15 @@ class CodingAgent: if self.iteration >= MAX_ITERATIONS: return {"thought": "Max iterations reached.", "tool": "done", "arguments": {"summary": "Stopped: max iterations reached."}} - + self.iteration += 1 self.messages = trim_messages(self.messages) try: raw = self._call_api(self.messages) raw = _strip_code_fences(raw) - action = json.loads(raw) + cleaned = extract_json(raw) + action = json.loads(cleaned) except json.JSONDecodeError: action = { "thought": "Could not parse LLM response as JSON.", @@ -413,7 +407,7 @@ class CodingAgent: self.pending_action = {"raw": raw, "action": action} return action - def approve(self) -> dict: + async def approve(self) -> dict: """Execute the pending action and return the result. Returns: @@ -442,7 +436,7 @@ class CodingAgent: } # Execute the tool - result = dispatch_tool(tool_name, arguments) + result = await dispatch_tool(tool_name, arguments) result = truncate_result(result) # Build feedback – nudge agent to replan on errors @@ -506,3 +500,39 @@ class CodingAgent: ), }) self.pending_action = None + +def main(): + """Example of how to use the CodingAgent in a simple loop.""" + agent = CodingAgent() + task = "Write a Python function that returns the nth Fibonacci number." + agent.start_task(task) + + if agent.pending_action: + print(f"Initial proposed action: {agent.pending_action['action']}") + + + while not agent.is_done: + action = asyncio.run(agent.propose_next_action()) + print(f"Proposed action: {action}") + + if action["tool"] == "done": + print("Task completed.") + break + else: + user_feedback = input("Approve this action? (y/n) ") + if user_feedback.lower() == "y": + result = asyncio.run(agent.approve()) + print(f"Tool result: {result}") + elif user_feedback.lower() == "n": + feedback = input("Enter feedback for the agent: ") + agent.reject(feedback) + + + if result["is_done"]: + print("Task completed.") + break + +if __name__ == "__main__": + main() + + diff --git a/backend/agent/mcp_server_adapter.py b/backend/agent/mcp_server_adapter.py new file mode 100644 index 0000000..8df0885 --- /dev/null +++ b/backend/agent/mcp_server_adapter.py @@ -0,0 +1,152 @@ +import asyncio +import json +import sys +from typing import List, Dict, Any +from pathlib import Path + +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client + +class MCPToolAdapter: + def __init__(self, config_path: str = "mcp_server_config.json"): + self.config_path = config_path + self.servers: Dict[str, Dict] = {} + #self.exit_stack: Dict[str, Any] = {} + self.tool_registry: List[Dict[str, Any]] = [] + + def _load_config(self) -> Dict[str, Any]: + """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}") + return {} + + try: + with open(path, 'r') as f: + return json.load(f) + except json.JSONDecodeError as e: + print(f"Error decoding JSON config: {e}") + return {} + + async def initialize_all_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())}") + + for server_name, params in config.items(): + print(f"Testing connection to {server_name}...") + + self.servers[server_name] = params + server_script = str(Path(__file__).parent / params["args"][0]) + if params.get("command") in ["py", "python", "python3"]: + server_command = sys.executable + else: + server_command = params["command"] + + server_params = StdioServerParameters( + command=server_command, + args=[server_script], + ) + + try: + # Verbindung aufbauen + async with stdio_client(server_params) as (read_stream, write_stream): + print(f"Connected to {server_name}. Initializing session...") + async with ClientSession(read_stream, write_stream) as session: + await session.initialize() + print(f"Session initialized for {server_name}. Requesting tools...") + result = await session.list_tools() + print(f"Tools received from {server_name}: {result}") + tools = result.tools + print(f"Tools received from {server_name}: {result}") + #tools = getattr(result, 'tools', []) + + for tool in tools: + t_params = tool.inputSchema.get("properties", {}) + if t_params: + param_lines = [] + for pname, pinfo in t_params.items(): + ptype = pinfo.get("type", "any") + pdesc = pinfo.get("description", "") + param_lines.append(f" - {pname} ({ptype}): {pdesc}") + param_str = "\n".join(param_lines) + else: + param_str = " (none)" + + t_definition = f"- {tool.name}: {tool.description}\nParameters:\n{param_str}" + + + self.tool_registry.append({ + "server": server_name, + "tool_name": tool.name, + "tool_description": t_definition + }) + + print(f"Registered tool '{tool.name}' from {server_name}.") + + print(f"Session for {server_name} ready. {len(tools)} tools found.") + + + except Exception as e: + print(f"Failed to initialize {server_name}: {e}") + + def get_all_tools(self) -> List[Dict[str, Any]]: + """Gibt alle gesammelten Tools zurück.""" + return self.tool_registry + + async def call_tool(self, tool_name: str, arguments: Dict[str, Any]): + """Findet den richtigen Server für ein Tool und führt es aus.""" + # Suche in der Registry nach dem passenden Server + 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.") + return f"Error: Tool '{tool_name}' not found in registry." + + server_name = tool_entry["server"] + s_params = self.servers.get(server_name) + + if s_params: + server_script = str(Path(__file__).parent / s_params["args"][0]) + if s_params.get("command") in ["py", "python", "python3"]: + server_command = sys.executable + else: + server_command = s_params["command"] + + server_params = StdioServerParameters( + command=server_command, + args=[server_script], + ) + + try: + async with stdio_client(server_params) as (read_stream, write_stream): + async with ClientSession(read_stream, write_stream) as session: + await session.initialize() + result = await session.call_tool(tool_name, arguments) + return result + except Exception as 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(): + adapter = MCPToolAdapter() + asyncio.run(adapter.initialize_all_servers()) + print("All servers initialized. Registered tools:") + for tool in adapter.get_all_tools(): + print(f"- {tool['tool_name']} (from {tool['server']})") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/backend/agent/mcp_server_adapter_RAG.py b/backend/agent/mcp_server_adapter_RAG.py new file mode 100644 index 0000000..6583c9b --- /dev/null +++ b/backend/agent/mcp_server_adapter_RAG.py @@ -0,0 +1,114 @@ +import asyncio +import json +# import os +import numpy as np +from typing import List, Dict, Any +from pathlib import Path + +from sentence_transformers import SentenceTransformer # embedder +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client + +class MCPToolRAGAdapter: + def __init__ (self, config_path: str = "mcp_server_config.json"): + self.config_path = config_path + self.tools = [] + self.toolnames = [] + self.embedder = SentenceTransformer('all-MiniLM-L6-v2') # for embedding tool descriptions + self.sessions = {} + self.exit_stack = {} + self.tool_registry = {} + self.tool_embeddings = None + + def _load_config(self) -> Dict[str, Any]: + config_path = Path(__file__).parent / self.config_path + if not config_path.exists(): + return {} + + try: + with open(self.config_path, 'r') as f: + return json.load(f) + except json.JSONDecodeError as e: + print(f"Error decoding JSON config: {e}") + return {} + + async def initialize_all_sessions(self): + """Initialize all MCP sessions defined in the config file and index their tools.""" + config = self._load_config() + for server_name, params in config.items(): + print(f"initializing session for {server_name} with params: {params}") + server_params = StdioServerParameters( + commanf=params["command"], + args=params.get("args", []), + # env=params.get("env", {}), + ) + + # Verbindung aufbauen (Kontext-Manager manuell handhaben für Langzeit-Sessions) + transport_gen = stdio_client(server_params) + read, write = await transport_gen.__aenter__() + session = ClientSession(read, write) + await session.__aenter__() + await session.initialize() + + self.sessions[server_name] = session + self.exit_stack[server_name] = (transport_gen, session) # Zum späteren sauberen Schließen speichern + print(f"Session for {server_name} initialized successfully.") + + # call tools and index thme + result = await session.list_tools() + tools = result.get("tools", []) + + for tool in tools: + self.tool_registry.append({ + "server": server_name, + "tool_name": tool["name"], + "definition": tool, + "search_text": f"{tool['name']}: {tool.get('description', '')}", + }) + self.tool_names.append(tool["name"]) + + # embeddings for all tools in this session + if self.tool_registry: + texts = [t["search_text"] for t in self.tool_registry] + self.tool_embeddings = self.embedder.encode(texts) + print(f"Indexing completed. {len(texts)} tools ready.") + + def get_relevant_tools(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]: + """Given a user query, return the most relevant tools based on semantic similarity.""" + if not self.tool_embeddings or not self.tool_registry: + print("No tools indexed yet.") + return [] + + query_embedding = self.embedder.encode([query]) + similarities = np.dot(self.tool_embeddings, query_embedding.T).flatten() + top_indices = np.argsort(similarities)[-top_k:][::-1] + + relevant_tools = [self.tool_registry[i] for i in top_indices] + + return relevant_tools + + async def call_tool(self, tool_name: str, arguments: Dict): + """ Finds the right server for the tool and calls it with the provided arguments. """ + for item in self.tool_registry: + if item["definition"].name == tool_name: + server_name = item["server"] + session = self.sessions.get(server_name) + if session: + try: + result = await session.call_tool(tool_name, arguments) + return result + except Exception as e: + print(f"Error calling tool {tool_name} on server {server_name}: {e}") + return f"Error calling tool: {e}" + + return f"Tool '{tool_name}' not found in registry." + + async def shutdown_all_sessions(self): + """Gracefully shutdown all MCP sessions.""" + for server_name, (transport_gen, session) in self.exit_stack.items(): + try: + await session.__aexit__(None, None, None) + await transport_gen.__aexit__(None, None, None) + print(f"Session for {server_name} shut down successfully.") + except Exception as e: + print(f"Error shutting down session for {server_name}: {e}") diff --git a/backend/agent/mcp_server_config.json b/backend/agent/mcp_server_config.json new file mode 100644 index 0000000..76ede4d --- /dev/null +++ b/backend/agent/mcp_server_config.json @@ -0,0 +1,19 @@ +{"FileSearchServer": { + "command": "py", + "args": ["servers/mcp_server_file_search.py"] +}, + +"WebSearchServer": { + "command": "py", + "args": ["servers/mcp_server_web_search.py"], + "env": { + "DDGS_API_KEY": "your_ddgs_api_key_here" + } +}, + +"CodeExecutionServer": { + "command": "py", + "args": ["servers/mcp_server_code_execution.py"] +} + +} \ No newline at end of file diff --git a/backend/agent/servers/mcp_server_code_execution.py b/backend/agent/servers/mcp_server_code_execution.py new file mode 100644 index 0000000..e04308a --- /dev/null +++ b/backend/agent/servers/mcp_server_code_execution.py @@ -0,0 +1,401 @@ +import ast +from datetime import datetime +import subprocess +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" + +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() + +# ── Configuration ──────────────────────────────────────────────────────────── +EXEC_TIMEOUT = 10 # seconds before killing the subprocess +MAX_OUTPUT_LENGTH = 3000 # max characters of stdout+stderr to return + +# ── Create the MCP server ──────────────────────────────────────────────────── +mcp = FastMCP("CodeExecutionServer") + +# ── Blocked Imports and Builtins ──────────────────────────────────────────────────── +BLOCKED_IMPORTS = { +# Filesystem access: + "os", "pathlib", "shutil", "glob", "tempfile", "fileinput", +# Process execution: + "subprocess", "multiprocessing", "threading", +# Network access: + "socket", "http", "urllib", "requests", "ftplib", "smtplib","xmlrpc", "asyncio", +# System internals: + "sys", "ctypes", "importlib", "code", "codeop", "compileall", +# Serialization exploits: + "pickle", "shelve", "marshal", +# Other dangerous: + "signal", "resource", "pty", "fcntl", "termios", "webbrowser", "antigravity" +} + +BLOCKED_BUILTINS = { +# Code execution: + "exec", "eval", "compile", "__import__", +# File access: + "open", +# Process control: + "exit", "quit", "breakpoint", +# Attribute manipulation: + "getattr", "setattr", "delattr", +# Introspection escapes: + "globals", "locals", "vars", "memoryview", "type" +} + +FORBIDDEN_SEQUENCES = ["../", "..\\", "/etc/", "/dev/", + "C:\\Windows", "C:\\Program Files", "C:\\Users", + "compile(", "__import__", "os.", "sys.", "subprocess."] + +# ── Static Analysis ──────────────────────────────────────────────────── +def check_code_safety(code: str) -> str | None: + """ + Statically analyze Python code for forbidden imports and builtins with ast. + + Args: + code: The Python code to analyze. + + Returns: + str or None + Error message if forbidden code found, None if safe. + """ + + try: + tree = ast.parse(code) + except SyntaxError as e: + return f"SyntaxError: {e}" + + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + top_level_module = alias.name.split('.')[0] + if top_level_module in BLOCKED_IMPORTS: + return (f"Blocked import: Import of '{alias.name}' is not allowed." + f"line {node.lineno}") + + elif isinstance(node, ast.ImportFrom): + if node.module: + top_level = node.module.split(".")[0] + if top_level in BLOCKED_IMPORTS: + 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: + return f"Blocked builtin: Use of builtin '{node.func.id}' is not allowed." + + for seq in FORBIDDEN_SEQUENCES: + if seq in code: + return f"Blocked: Suspect path sequence '{seq}' detected." + + return None # No violations found + +@mcp.tool() +def analyse_structure(code: str) -> str: + """ + Analyze the structure of Python code and return a summary of its components. + + Args: + code: The Python code to analyze in str format. + Returns: + A summary of the code's structure, including functions, classes, and imports. + """ + try: + tree = ast.parse(code) + except SyntaxError as e: + return f"Syntax Error: Invalid Python code provided. Line {e.lineno}: {e.msg}" + except Exception as e: + return f"Error parsing code: {str(e)}" + + analysis = { + "imports": [], + "classes": [], + "functions": [] + } + + for node in tree.body: + if isinstance(node, ast.Import): + for alias in node.names: + analysis["imports"].append(f"import {alias.name}") + + elif isinstance(node, ast.ImportFrom): + module = node.module or "" + for alias in node.names: + analysis["imports"].append(f"from {module} import {alias.name}") + + elif isinstance(node, ast.ClassDef): + methods = [n.name for n in node.body if isinstance(n, ast.FunctionDef)] + analysis["classes"].append({ + "name": node.name, + "methods": methods + }) + + elif isinstance(node, ast.FunctionDef): + args = [arg.arg for arg in node.args.args] + analysis["functions"].append({ + "name": node.name, + "args": args + }) + + # Zusammenfassung als String formatieren + lines = ["--- Code Structure Analysis ---"] + + if analysis["imports"]: + lines.append("\n[Imports]") + lines.extend([f" - {imp}" for imp in analysis["imports"]]) + + if analysis["classes"]: + lines.append("\n[Classes]") + for cls in analysis["classes"]: + lines.append(f" - class {cls['name']}:") + if cls["methods"]: + lines.extend([f" * method: {m}" for m in cls["methods"]]) + else: + lines.append(" * (no methods)") + + if analysis["functions"]: + lines.append("\n[Top-Level Functions]") + for func in analysis["functions"]: + args_str = ", ".join(func["args"]) + lines.append(f" - def {func['name']}({args_str})") + + if not any([analysis["imports"], analysis["classes"], analysis["functions"]]): + return "Analysis complete: No top-level imports, classes, or functions found." + + return "\n".join(lines) + + +@mcp.tool() +def lint_code(code: str) -> str: + """ + Runs a fast static analysis check to catch syntax errors, unused imports, + or undefined variables without executing the code. + + Args: + code: The Python code to lint in str format. + Returns: + A report of linting issues or a success message if the code is clean. + """ + error_buffer = io.StringIO() + warning_buffer = io.StringIO() + + reporter = Reporter(warning_buffer, error_buffer) + + try: + check(code, filename="", reporter=reporter) + except Exception as e: + return f"Critical error during linting: {str(e)}" + + errors = error_buffer.getvalue().strip() + warnings = warning_buffer.getvalue().strip() + + # Ergebnis-String zusammenbauen + if not errors and not warnings: + return "Linting complete: No issues found. The code is syntactically sound." + + report = ["--- Linting Report ---"] + + if errors: + report.append("\n[Syntax Errors / Critical Issues]") + report.append(errors) + + if warnings: + report.append("\n[Logical Issues (Unused imports, Undefined names, etc.)]") + report.append(warnings) + + report.append("\nAdvice: Please fix these issues before attempting to execute the code.") + + return "\n".join(report) + + +@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: + """ + Run Python code in a sandboxed environment. + + The sandbox blocks dangerous operations (filesystem, network, process + control). Code is killed after 10 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 to execute 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 python_code_validation(code: str) -> str: + """ + Validate Python code for syntax and safety without executing it. + This tool performs static analysis to check for syntax errors. + + Args: + code: The Python code to validate in str format. + Returns: + A message indicating the validation result. + And if sandboxed test execution is allowed. + """ + try: + ast.parse(code) + except SyntaxError as e: + return f"SyntaxError: {e}" + + try: + static_analysis_result = check_code_safety(code) + if static_analysis_result: + return f"Valid Syntax, but with safety concerns: {static_analysis_result}; code execution is not allowed." + except Exception as e: + return f"Error during code safety analysis: {e}" + + +# ── Run the server ─────────────────────────────────────────────────────────── + +if __name__ == "__main__": + mcp.run(transport="stdio") diff --git a/backend/agent/servers/mcp_server_file_search.py b/backend/agent/servers/mcp_server_file_search.py new file mode 100644 index 0000000..65180bc --- /dev/null +++ b/backend/agent/servers/mcp_server_file_search.py @@ -0,0 +1,218 @@ +from pathlib import Path +from mcp.server.fastmcp import FastMCP + +# ── Configuration ──────────────────────────────────────────────────────────── +project_dir = Path(__file__).resolve().parent.parent.parent.parent +ALLOWED_DIR = project_dir / "workspace" +ALLOWED_FILE_TYPES = [".py",".js",".html",".css",".json",".yaml",".yml",".sh",".md",".txt",".tex",".c",".cpp",".java"] + +# ── Create the MCP server ──────────────────────────────────────────────────── +mcp = FastMCP("FileSearchServer") + + +# ── Helper: path validation ────────────────────────────────────────────────── +def _safe_path(requested: str) -> Path: + """Resolve and validate a path is inside ALLOWED_DIR.""" + resolved = (ALLOWED_DIR / requested).resolve() + if not str(resolved).startswith(str(ALLOWED_DIR)): + raise ValueError( + f"Access denied: '{requested}' resolves outside " + f"the allowed directory '{ALLOWED_DIR}'" + ) + return resolved + + +# ── MCP Tools ──────────────────────────────────────────────────────────────── + +@mcp.tool() +def list_files() -> str: + """List all files in the project directory (recursively). + + Returns a newline-separated list of relative file paths. + """ + files = sorted( + f.relative_to(ALLOWED_DIR) + for f in ALLOWED_DIR.rglob("*") + if f.is_file() and "__pycache__" not in f.parts + ) + if not files: + return "No files found in the project directory." + return "\n".join(str(f) for f in files) + + +@mcp.tool() +def get_file_tree(dir_path: str=ALLOWED_DIR) -> str: + """Get a tree representation of the project directory. + + Args: + path: The directory path to display (default is the allowed directory). + + Returns: + A string representing the directory structure, similar to 'tree' command output. + """ + try: + safe_dir = _safe_path(dir_path) + if not safe_dir: + return f"Error: Invalid directory path '{dir_path}'." + elif not safe_dir.exists(): + return f"Error: Directory '{dir_path}' does not exist." + elif not safe_dir.is_dir(): + return f"Error: '{dir_path}' is not a valid directory within the allowed path." + except ValueError as e: + return f"Error: {e}" + + + 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 = [] + for i, entry in enumerate(entries): + connector = "└── " if i == len(entries) - 1 else "├── " + lines.append(f"{prefix}{connector}{entry.name}") + if entry.is_dir(): + extension = " " if i == len(entries) - 1 else "│ " + lines.append(_tree(entry, prefix + extension)) + return "\n".join(lines) + + return _tree(dir_path) + + +@mcp.tool() +def search_files(query: str) -> str: + """Search for files whose name or content contains the query string. + + Args: + query: The search term (case-insensitive). + + Returns: + A formatted string of search results, or a message if no matches found. + """ + query_lower = query.lower() + results = [] + + for f in sorted(ALLOWED_DIR.rglob("*")): + if not f.is_file() or "__pycache__" in f.parts: + continue + rel = f.relative_to(ALLOWED_DIR) + + if query_lower in str(rel).lower(): + results.append(f"[name match] {rel}") + + try: + content = f.read_text(encoding="utf-8") + for i, line in enumerate(content.splitlines(), 1): + if query_lower in line.lower(): + snippet = line.strip()[:100] + results.append(f"[content] {rel}:{i} -- {snippet}") + except (UnicodeDecodeError, PermissionError): + pass + + if not results: + return f"No matches found for '{query}'." + return "\n".join(results[:30]) # limit to 30 matches + + +@mcp.tool() +def read_file(path: str) -> str: + """Read the contents of a file. + + Args: + path: Relative path to the file within the project directory. + + Returns: + The file content as a string, or an error message if the file cannot be read. + """ + try: + resolved = _safe_path(path) + except ValueError as e: + return f"Error: {e}" + + if not resolved.exists(): + return f"Error: File '{path}' does not exist." + if not resolved.is_file(): + return f"Error: '{path}' is not a file." + + try: + return resolved.read_text(encoding="utf-8") + except UnicodeDecodeError: + return f"Error: '{path}' is not a text file (binary content)." + except PermissionError: + return f"Error: Permission denied when trying to read '{path}'." + except Exception as e: + return f"Error reading file '{path}': {e}" + +@mcp.tool() +def write_new_file(path: str, content: str) -> str: + """Write content to a new file in the allowed directory. + Existing files cannot be overwritten with this tool. + + Args: + path: Relative path to the file within the allowed directory. + content: The content to write to the file. + + Returns: + A success or error message. + """ + + try: + resolved = _safe_path(path) + except ValueError as e: + return f"Error: {e}" + + if resolved.exists(): + 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: + 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") + return f"OK: wrote {len(content)} chars to {path}." + + except FileNotFoundError as e: + print(f"FileNotFoundError for {path}: {e}") + return f"Error: {e}" + except PermissionError as e: + print(f"PermissionError for {path}: {e}") + return f"Error: {e}" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +def create_new_directory(path: str) -> str: + """Create a new empty directory in the allowed directory. + + Args: + path: Relative path to the directory within the allowed directory. + + Returns: + A success or error message. + """ + try: + resolved = _safe_path(path) + except ValueError as e: + return f"Error: {e}" + + if resolved.exists(): + return f"Error: File '{path}' already exists." + + if resolved.suffix != None and resolved.suffix != "": + return f"Error: can only create directories, got '{resolved.suffix}'." + + try: + resolved.parent.mkdir(parents=True, exist_ok=True) + resolved.mkdir() + return f"OK: created empty directory at {path}." + except Exception as e: + return f"Error creating dictionary file '{path}': {e}" + + +# ── Run the server ─────────────────────────────────────────────────────────── + +if __name__ == "__main__": + mcp.run(transport="stdio") + \ No newline at end of file diff --git a/backend/agent/servers/mcp_server_web_search.py b/backend/agent/servers/mcp_server_web_search.py new file mode 100644 index 0000000..39c9fb4 --- /dev/null +++ b/backend/agent/servers/mcp_server_web_search.py @@ -0,0 +1,120 @@ +from urllib.parse import urlparse +from mcp.server.fastmcp import FastMCP + +# ── Configuration ──────────────────────────────────────────────────────────── +MAX_PAGE_LENGTH = 4000 # max characters to return from a fetched page +REQUEST_TIMEOUT = 10 # seconds + +# ── Create the MCP server ──────────────────────────────────────────────────── +mcp = FastMCP("WebSearchServer") + + +# ── Helper: URL validation (SSRF prevention) ───────────────────────────────── + +def _validate_url(url: str) -> str: + """Validate a URL to prevent SSRF attacks.""" + parsed = urlparse(url) + + if parsed.scheme not in ("http", "https"): + raise ValueError( + f"Blocked scheme '{parsed.scheme}'. Only http and https are allowed." + ) + + hostname = parsed.hostname or "" + + blocked_hosts = {"localhost", "127.0.0.1", "0.0.0.0", "169.254.169.254"} + if hostname in blocked_hosts: + raise ValueError(f"Blocked internal host: {hostname}") + + private_prefixes = ( + "10.", "172.16.", "172.17.", "172.18.", "172.19.", + "172.20.", "172.21.", "172.22.", "172.23.", "172.24.", + "172.25.", "172.26.", "172.27.", "172.28.", "172.29.", + "172.30.", "172.31.", "192.168.", + ) + for prefix in private_prefixes: + if hostname.startswith(prefix): + raise ValueError(f"Blocked private IP range: {hostname}") + + return url + + +# ── MCP Tools ──────────────────────────────────────────────────────────────── + +@mcp.tool() +def web_search(query: str, max_results: int = 5) -> str: + """Search the web using DuckDuckGo. + + Args: + query: The search query. + max_results: Maximum number of results to return (default 5). + Returns: + A formatted string of search results, or a message if no matches found. + """ + try: + from ddgs import DDGS + results = DDGS().text(query, max_results=max_results) + + if not results: + return f"No results found for: {query}" + + formatted = [] + for r in results: + formatted.append( + f"Title: {r['title']}\n" + f"URL: {r['href']}\n" + f"Snippet: {r['body']}" + ) + return "\n---\n".join(formatted) + + except Exception as e: + return f"Search error: {e}" + + +@mcp.tool() +def fetch_page(url: str) -> str: + """Fetch a web page and extract its text content. + + Args: + url: The URL to fetch. + Returns: + The text content of the fetched page, or an error message. + """ + try: + url = _validate_url(url) + except ValueError as e: + return f"URL blocked: {e}" + + try: + import requests + from bs4 import BeautifulSoup + + response = requests.get( + url, + timeout=REQUEST_TIMEOUT, + headers={"User-Agent": "Mozilla/5.0 (Lightweight Web Search MCP Server)"}, + ) + + if response.status_code != 200: + return f"HTTP error {response.status_code} fetching {url}" + + soup = BeautifulSoup(response.text, "html.parser") + + for tag in soup(["script", "style", "nav", "footer"]): + tag.decompose() + + text = soup.get_text(separator="\n", strip=True) + + if len(text) > MAX_PAGE_LENGTH: + text = text[:MAX_PAGE_LENGTH] + "\n\n[... truncated ...]" + + return text if text else "Page fetched but no text content found." + + except Exception as e: + return f"Error fetching page: {e}" + + +# ── Run the server ─────────────────────────────────────────────────────────── + +if __name__ == "__main__": + mcp.run(transport="stdio") diff --git a/backend/managers/file_manager.py b/backend/managers/file_manager.py index b1d223f..d86cff3 100644 --- a/backend/managers/file_manager.py +++ b/backend/managers/file_manager.py @@ -10,6 +10,17 @@ class FileManager: self.base_path.mkdir(exist_ok=True) def create_folder(self, relative_path: str, name: str) -> bool: + """ + Creates a new folder at the specified relative path. + The relative_path should be the path to the folder relative to the base path, + and name should be the name of the new folder (without any slashes). + + Args: + relative_path (str): The relative path (without base path) where the new folder should be created. + name (str): The name of the new folder to create (should not contain slashes). + Returns: + bool: True if folder was created successfully, False otherwise. + """ if not name: st.error(f"Invalid folder name: {name}") return False @@ -40,7 +51,18 @@ class FileManager: st.error(f"Error creating folder {relative_path}: {str(e)}") return False - def create_file(self, relative_path: str, name: str) -> bool: + def create_file(self, relative_path: str, name: str) -> bool: + """ + Creates a new file at the specified relative path. + The relative_path should be the path to the folder relative to the base path where the file should be created, + and name should be the name of the new file (without any slashes). + + Args: + relative_path (str): The relative path (without base path) where the new file should be created. + name (str): The name of the new file to create (should not contain slashes). + Returns: + bool: True if file was created successfully, False otherwise. + """ if not name or name.strip() == "" : st.error(f"Invalid file name: {name}") return False @@ -71,6 +93,15 @@ class FileManager: return False def read_file(self, relative_path: Path) -> str: + """ + Reads the content of a file. + The relative_path should be the path to the file relative to the base path. + + Args: + relative_path (str): The relative path (without base path) to the file to read, including the file name + Returns: + str: The content of the file, or an empty string if there was an error. + """ file_path = (relative_path).resolve() if not file_path.exists(): @@ -93,7 +124,17 @@ class FileManager: st.error(f"Error reading file {relative_path}: {str(e)}") return "" - def save_file(self, relative_path: str, content: str): + def save_file(self, relative_path: str, content: str) -> bool: + """ + Saves content to a file. + The relative_path should be the path to the file relative to the base path. + + Args: + relative_path (str): The relative path(without base path) to the file to save, including the file name + content (str): The content to write to the file + Returns: + bool: True if save was successful, False otherwise. + """ file_path = (Path(relative_path)).resolve() if not str(file_path).startswith(str(self.base_path.resolve())): @@ -109,6 +150,16 @@ class FileManager: return False def rename_file(self, old_relative_path: str, new_name: str) -> bool: + """ + Renames a file while keeping the same extension. + The new_name should not include the extension, as it will be preserved from the old name. + + Args: + old_relative_path (str): The current relative path (without base path) of the file to rename, including the file name. + new_name (str): The new name for the file, without extension. + Returns: + bool: True if rename was successful, False otherwise. + """ if not new_name or new_name.strip() == "": st.error(f"Invalid file name: {new_name}") return False @@ -119,7 +170,7 @@ class FileManager: if not Path(new_name).suffix == file_type: new_name = Path(new_name).with_suffix(file_type) # Ensure the file extension remains the same - old_file_path = (Path(old_relative_path)).resolve() + old_file_path = (Path(self.base_path / old_relative_path)).resolve() new_file_path = old_file_path.parent / new_name if not str(old_file_path).startswith(str(self.base_path.resolve())) or not str(new_file_path).startswith(str(self.base_path.resolve())): @@ -131,9 +182,18 @@ class FileManager: return True except FileNotFoundError: st.error(f"File not found: {old_relative_path}") - return False + return False + - def delete_folder(self, relative_path) -> bool: + def delete_folder(self, relative_path: str) -> bool: + """Deletes a folder and all its contents. + The relative_path should be the path to the folder relative to the base path. + + Args: + relative_path (str): The relative path (without base path) to the folder to delete. + Returns: + bool: True if deletion was successful, False otherwise. + """ folder_path = (self.base_path / relative_path).resolve() if not str(folder_path).startswith(str(self.base_path.resolve())): @@ -152,15 +212,25 @@ class FileManager: st.error(f"Error deleting folder {relative_path}: {str(e)}") return False - def delete_file(self, relative_path): - file_path = Path(relative_path).resolve() + def delete_file(self, relative_path: str) -> bool: + """Deletes a file. + The relative_path should be the path to the file relative to the base path. + + Args: + relative_path (str): The relative path (without base path) to the file to delete, including the file name. + Returns: + bool: True if deletion was successful, False otherwise. + """ + file_path = Path(relative_path) + abs_file_path = (Path(self.base_path) / file_path).resolve() + print(f"Absolute file path resolved to: {abs_file_path}") # Debugging info - if not str(file_path).startswith(str(self.base_path.resolve())): + if not str(abs_file_path).startswith(str(self.base_path.resolve())): st.error(f"Access denied: {relative_path}") return False - + try: - file_path.unlink() + abs_file_path.unlink() return True except FileNotFoundError: st.error(f"File not found: {relative_path}") @@ -170,6 +240,14 @@ class FileManager: return False def get_file_tree(self): + """ + Builds a nested dictionary representing the file tree starting from the base path. + Directories are represented as keys with dictionary values, + and files are represented as keys with None + + Returns: + dict: A nested dictionary representing the file tree. + """ def build_tree(path: Path): tree = {} diff --git a/frontend/chat.py b/frontend/chat.py index 7716a04..84cee58 100644 --- a/frontend/chat.py +++ b/frontend/chat.py @@ -2,8 +2,18 @@ import streamlit as st from backend.managers.chat_manager import ChatManager from backend.managers.system_prompter import SystemPrompter +import asyncio + # ── Agent Mode helpers ──────────────────────────────────────────────────────── +def _run_async(coro): + """Hilfsfunktion um async Code in sync Streamlit auszuführen""" + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + return loop.run_until_complete(coro) def _start_agent(task: str): """Initialise a fresh CodingAgent, start the task, @@ -11,7 +21,7 @@ def _start_agent(task: str): from backend.agent.coding_agent import CodingAgent agent = CodingAgent() agent.start_task(task) - action = agent.propose_next_action() + action = _run_async(agent.propose_next_action()) st.session_state.coding_agent = agent st.session_state.agent_pending_action = action st.session_state.agent_status = "waiting_approval" @@ -24,7 +34,7 @@ def _approve_action(): agent = st.session_state.coding_agent pending = st.session_state.agent_pending_action - result = agent.approve() + result = _run_async(agent.approve()) st.session_state.agent_log.append({ "thought": pending.get("thought", ""), @@ -37,7 +47,7 @@ def _approve_action(): st.session_state.agent_status = "done" st.session_state.agent_pending_action = None else: - next_action = agent.propose_next_action() + next_action = _run_async(agent.propose_next_action()) st.session_state.agent_pending_action = next_action st.session_state.agent_status = "waiting_approval" @@ -46,7 +56,7 @@ def _reject_action(feedback: str): """Reject the pending action with optional feedback, then replan.""" agent = st.session_state.coding_agent agent.reject(feedback or "Please try a different approach.") - next_action = agent.propose_next_action() + next_action = _run_async(agent.propose_next_action()) st.session_state.agent_pending_action = next_action st.session_state.agent_status = "waiting_approval" @@ -55,7 +65,7 @@ def _followup_agent(question: str): """Inject a follow-up question into the finished agent and resume the loop.""" agent = st.session_state.coding_agent agent.follow_up(question) - action = agent.propose_next_action() + action = _run_async(agent.propose_next_action()) st.session_state.agent_pending_action = action st.session_state.agent_status = "waiting_approval" @@ -103,6 +113,8 @@ 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()) diff --git a/frontend/editor.py b/frontend/editor.py index 4797f58..67d0d05 100644 --- a/frontend/editor.py +++ b/frontend/editor.py @@ -13,6 +13,7 @@ LANG_MAP = { } + # ── Modals ──────────────────────────────────────────────────────────────────── @st.dialog("Rename File") @@ -29,7 +30,8 @@ def _rename_dialog(file_path: str): elif "/" in new_name or "\\" in new_name: st.warning("Name must not contain slashes.") else: - if fm.rename_file(file_path, new_name.strip()): + rel_path = str(Path(file_path).relative_to(fm.base_path)) + if fm.rename_file(rel_path, new_name.strip()): ext = Path(file_path).suffix new_file_path = str(Path(file_path).parent / (Path(new_name.strip()).stem + ext)) i = st.session_state.open_files.index(file_path) @@ -47,17 +49,19 @@ def _rename_dialog(file_path: str): @st.dialog("Delete File") -def _delete_dialog(file_path: str): +def _delete_dialog(abs_file_path: str): fm = FileManager() - st.warning(f"Delete **{Path(file_path).name}**? This cannot be undone.") + file_name = Path(abs_file_path).name + relative_path = str(Path(abs_file_path).relative_to(fm.base_path)) + st.warning(f"Delete **{file_name}**? This cannot be undone.") col1, col2 = st.columns(2) with col1: if st.button("Delete", type="primary", use_container_width=True): - if fm.delete_file(file_path): - st.session_state.open_files.remove(file_path) - st.session_state.files_content.pop(file_path, None) - if st.session_state.active_file == file_path: + if fm.delete_file(relative_path): + st.session_state.open_files.remove(abs_file_path) + st.session_state.files_content.pop(abs_file_path, None) + if st.session_state.active_file == abs_file_path: st.session_state.active_file = ( st.session_state.open_files[0] if st.session_state.open_files else None diff --git a/frontend/sidebar.py b/frontend/sidebar.py index 8cc1975..fa0a1b6 100644 --- a/frontend/sidebar.py +++ b/frontend/sidebar.py @@ -46,10 +46,16 @@ def _delete_folder_dialog(folder_rel: str, folder_name: str): @st.dialog("Add File") def _add_file_dialog(parent_path: str = ""): - name = st.text_input("File name:", placeholder="e.g. script.py") - col1, col2 = st.columns(2) - with col1: - if st.button("Create", type="primary", use_container_width=True): + with st.form("add_file_form"): + name = st.text_input("File name:", placeholder="e.g. script.py") + col1, col2 = st.columns(2) + + with col1: + submitted = st.form_submit_button("Create", type="primary", use_container_width=True) + with col2: + cancel = st.form_submit_button("Cancel", use_container_width=True) + + if submitted: if not name.strip(): st.warning("Please enter a file name.") elif "/" in name or "\\" in name: @@ -57,17 +63,23 @@ def _add_file_dialog(parent_path: str = ""): else: if fm.create_file(parent_path, name.strip()): st.rerun() - with col2: - if st.button("Cancel", use_container_width=True): + + if cancel: st.rerun() @st.dialog("Add Folder") def _add_folder_dialog(parent_path: str = ""): - name = st.text_input("Folder name:", placeholder="e.g. utils") - col1, col2 = st.columns(2) - with col1: - if st.button("Create", type="primary", use_container_width=True): + with st.form("add_folder_form"): + name = st.text_input("Folder name:", placeholder="e.g. utils") + col1, col2 = st.columns(2) + + with col1: + submitted = st.form_submit_button("Create", type="primary", use_container_width=True) + with col2: + cancel = st.form_submit_button("Cancel", use_container_width=True) + + if submitted: if not name.strip(): st.warning("Please enter a folder name.") elif "/" in name or "\\" in name: @@ -75,6 +87,98 @@ def _add_folder_dialog(parent_path: str = ""): else: if fm.create_folder(parent_path, name.strip()): st.rerun() + + if cancel: + st.rerun() + + +@st.dialog("Rename File") +def _rename_file_dialog(relative_file_path: str, file_name: str): + """ + Dialog to rename a file. + + Args: + relative_file_path (str): The current relative path (without base path) to the file to rename, including the file name. + file_name (str): The current name of the file, including the extension. + """ + st.write(f"Current name: **{file_name}**") + + with st.form("rename_file_form"): + new_name = st.text_input( + "New name:", + value=Path(relative_file_path).stem + ) + + col1, col2 = st.columns(2) + + with col1: + submitted = st.form_submit_button( + "Confirm", + type="primary", + use_container_width=True + ) + + with col2: + cancel = st.form_submit_button( + "Cancel", + use_container_width=True + ) + + if submitted: + if not new_name.strip(): + st.warning("Please enter a name.") + elif "/" in new_name or "\\" in new_name or "." in new_name: + st.warning("Name must not contain slashes.") + else: + if fm.rename_file(relative_file_path, new_name.strip()): + absolute_file_path = str(Path(fm.base_path / relative_file_path)) + ext = Path(absolute_file_path).suffix + new_file_path = str( + Path(absolute_file_path).parent / (Path(new_name.strip()).stem + ext) + ) + + if absolute_file_path in st.session_state.open_files: + i = st.session_state.open_files.index(absolute_file_path) + st.session_state.open_files[i] = new_file_path + + if absolute_file_path in st.session_state.files_content: + st.session_state.files_content[new_file_path] = \ + st.session_state.files_content.pop(absolute_file_path) + + if st.session_state.active_file == absolute_file_path: + st.session_state.active_file = new_file_path + + st.rerun() + else: + st.error("Rename failed. Check that the file still exists.") + st.error(f"Attempted to rename: {relative_file_path} to {new_name.strip()}") + + if cancel: + st.rerun() + + +@st.dialog("Delete File") +def _delete_file_dialog(relative_file_path: str, file_name: str): + st.warning(f"Delete **{file_name}**? This cannot be undone.") + + col1, col2 = st.columns(2) + with col1: + if st.button("Delete", type="primary", use_container_width=True): + if fm.delete_file(relative_file_path): + abs_file_path = str(Path(fm.base_path) / relative_file_path) + print(f"Deleting file at absolute path: {abs_file_path}") # Debugging info + print(f"Current open files before deletion: {st.session_state.open_files}") # Debugging info + + st.session_state.open_files.remove(abs_file_path) + st.session_state.files_content.pop(abs_file_path, None) + if st.session_state.active_file == abs_file_path: + st.session_state.active_file = ( + st.session_state.open_files[0] + if st.session_state.open_files else None + ) + st.rerun() + else: + st.error("Delete failed. Check that the file still exists.") with col2: if st.button("Cancel", use_container_width=True): st.rerun() @@ -186,6 +290,17 @@ def render_sidebar(): _add_folder_dialog(folder_rel) if st.button("Delete Folder", key="btn_delete_folder", use_container_width=True): _delete_folder_dialog(folder_rel, folder_name) + + if st.session_state.get("active_file"): + active_file_name = Path(st.session_state.active_file).name + file_rel = str(Path(st.session_state.active_file).relative_to(fm.base_path)) + + with st.container(border=True): + st.write(f"**File actions:** {active_file_name}") + if st.button("Rename File", key="btn_rename_file", use_container_width=True): + _rename_file_dialog(file_rel, active_file_name) + if st.button("Delete File", key="btn_delete_active_file", use_container_width=True): + _delete_file_dialog(file_rel, active_file_name) with add_more: with st.popover("⚙️ Explorer Options", key="popover_options", use_container_width=True): diff --git a/frontend/state.py b/frontend/state.py index 6021fd1..b301321 100644 --- a/frontend/state.py +++ b/frontend/state.py @@ -19,12 +19,20 @@ def init_state(): # Editor state initialization if "open_files" not in st.session_state: + """A list of currently open file paths - absolute paths only. The order determines the tab order in the UI. + Format: [ "path/to/file1.py", "path/to/file2.js", ... ] + """ st.session_state.open_files = [] if "files_content" not in st.session_state: + """A dictionary mapping file paths to their current content in the editor. + Format: { "path/to/file.py": "file content as string", ... } + """ st.session_state.files_content = {} if "active_file" not in st.session_state: + """The currently active file in the editor (absolute path in string e.g. "/workspace/path/to/file.py"). + Should be one of the paths in open_files or None if no file is open.""" st.session_state.active_file = None if "active_tab" not in st.session_state: diff --git a/requirements.txt b/requirements.txt index ada1cab..850c7a9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,6 +3,8 @@ streamlit>=1.28.0 # AI/LLM Integration openai>=1.0.0 +mcp>=0.1.0 +ddgs>=0.1.0 # Web & API requests>=2.31.0