Merge pull request 'MCP-setup' (#13) from MCP-setup into main
Reviewed-on: meulilivio/AISE1_Project#13
This commit is contained in:
commit
901fbab543
@ -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}": "<value>"')
|
||||
else:
|
||||
params.append(f'"{pname}": "<optional, default={param.default!r}>"')
|
||||
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).
|
||||
|
||||
<capabilities>
|
||||
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
|
||||
<capabilities>
|
||||
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.
|
||||
</capabilities>
|
||||
|
||||
<tools>
|
||||
{build_tool_description()}
|
||||
{build_all_tool_description()}
|
||||
</tools>
|
||||
|
||||
|
||||
<workflow>
|
||||
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()
|
||||
|
||||
|
||||
|
||||
152
backend/agent/mcp_server_adapter.py
Normal file
152
backend/agent/mcp_server_adapter.py
Normal file
@ -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()
|
||||
114
backend/agent/mcp_server_adapter_RAG.py
Normal file
114
backend/agent/mcp_server_adapter_RAG.py
Normal file
@ -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}")
|
||||
19
backend/agent/mcp_server_config.json
Normal file
19
backend/agent/mcp_server_config.json
Normal file
@ -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"]
|
||||
}
|
||||
|
||||
}
|
||||
401
backend/agent/servers/mcp_server_code_execution.py
Normal file
401
backend/agent/servers/mcp_server_code_execution.py
Normal file
@ -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="<agent_code>", 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")
|
||||
218
backend/agent/servers/mcp_server_file_search.py
Normal file
218
backend/agent/servers/mcp_server_file_search.py
Normal file
@ -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")
|
||||
|
||||
120
backend/agent/servers/mcp_server_web_search.py
Normal file
120
backend/agent/servers/mcp_server_web_search.py
Normal file
@ -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")
|
||||
@ -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 = {}
|
||||
|
||||
@ -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())
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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):
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user