MCP-Tool fine tuning
This commit is contained in:
Irina Rueegg
parent
e5831f69a8
commit
560e56b596
@ -24,9 +24,7 @@ import asyncio
|
||||
import pprint
|
||||
|
||||
import requests
|
||||
import httpx
|
||||
from dotenv import load_dotenv
|
||||
#from backend.agent.mcp_server_adapter import MCPToolRAGAdapter
|
||||
#from mcp_server_adapter import MCPToolAdapter # Import from current directory for easier testing without package structure
|
||||
from backend.agent.mcp_server_adapter import MCPToolAdapter
|
||||
|
||||
@ -50,155 +48,8 @@ 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 register_tool(func):
|
||||
# """Decorator – adds a function to the global tool registry."""
|
||||
# TOOL_REGISTRY[func.__name__] = func
|
||||
# return func
|
||||
#
|
||||
#
|
||||
#@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()
|
||||
# try:
|
||||
# ast.parse(source)
|
||||
# return "OK: syntax is valid."
|
||||
# except SyntaxError as e:
|
||||
# return f"SYNTAX ERROR: {e}"
|
||||
#
|
||||
#
|
||||
#@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}"
|
||||
# except Exception as e:
|
||||
# return f"ERROR in {tool_name}: {type(e).__name__}: {e}"
|
||||
#
|
||||
|
||||
def build_all_tool_description() -> str:
|
||||
"""Get relevant tools from the MCP servers based on the query."""
|
||||
@ -210,21 +61,6 @@ def build_all_tool_description() -> str:
|
||||
for tool in all_tools:
|
||||
pprint.pprint(f"{tool}")
|
||||
descriptions.append(f"- {tool['tool_name']}: {tool['tool_description']}")
|
||||
#
|
||||
#params = tool.tool_description.inputSchema.get("properties", {})
|
||||
#if params:
|
||||
# param_lines = []
|
||||
# for pname, pinfo in 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)"
|
||||
#descriptions.append(
|
||||
# f"- {tool.name}: {tool.description}\n"
|
||||
# f"Parameters:\n{param_str}"
|
||||
#)
|
||||
|
||||
return "\n".join(descriptions)
|
||||
|
||||
@ -357,6 +193,92 @@ 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).
|
||||
"""
|
||||
import re
|
||||
|
||||
# 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."""
|
||||
@ -416,12 +338,6 @@ class CodingAgent:
|
||||
"stream": False,
|
||||
}
|
||||
|
||||
#async with httpx.AsyncClient(timeout=30) as client:
|
||||
# response = await client.post(
|
||||
# self.api_url,
|
||||
# headers=headers,
|
||||
# json=payload,
|
||||
# timeout=60)
|
||||
response = requests.post(self.api_url, headers=headers, json=payload, timeout=60)
|
||||
|
||||
if response.status_code != 200:
|
||||
@ -467,7 +383,8 @@ class CodingAgent:
|
||||
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.",
|
||||
|
||||
@ -1,13 +1,18 @@
|
||||
import ast
|
||||
import subprocess
|
||||
import io
|
||||
from pyflakes.api import check
|
||||
from pyflakes.reporter import Reporter
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
# ── 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",
|
||||
@ -36,7 +41,7 @@ BLOCKED_BUILTINS = {
|
||||
"globals", "locals", "vars", "memoryview", "type"
|
||||
}
|
||||
|
||||
|
||||
# ── Static Analysis ────────────────────────────────────────────────────
|
||||
def check_code_safety(code: str) -> str | None:
|
||||
"""
|
||||
Statically analyze Python code for forbidden imports and builtins with ast.
|
||||
@ -57,26 +62,153 @@ def check_code_safety(code: str) -> str | None:
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
for alias in node.names:
|
||||
if alias.name.split('.')[0] in BLOCKED_IMPORTS:
|
||||
return f"Error: Import of '{alias.name}' is not allowed."
|
||||
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 and node.module.split('.')[0] in BLOCKED_IMPORTS:
|
||||
return f"Error: Import from '{node.module}' is not allowed."
|
||||
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) and node.func.id in BLOCKED_BUILTINS:
|
||||
return f"Error: Use of builtin '{node.func.id}' is not allowed."
|
||||
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."
|
||||
|
||||
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 run_python_sandboxed(code: str) -> str:
|
||||
"""
|
||||
Run Python code in a sandboxed subprocess.
|
||||
Run Python code in a sandboxed environment.
|
||||
|
||||
Static analysis (check_code_safety), Subprocess isolation (child process)
|
||||
Timeout (killed after EXEC_TIMEOUT seconds), Output truncation (max MAX_OUTPUT_LENGTH chars)
|
||||
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.
|
||||
@ -86,23 +218,56 @@ def run_python_sandboxed(code: str) -> str:
|
||||
"""
|
||||
|
||||
static_safety = check_code_safety(code)
|
||||
if static_safety is not None:
|
||||
return static_safety
|
||||
if static_safety:
|
||||
return f"Code rejected:{static_safety}"
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["python3", "-c", code],
|
||||
capture_output=True, text=True, timeout=EXEC_TIMEOUT)
|
||||
["python", "-c", code],
|
||||
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]..."
|
||||
return output if output.strip() else "Code executed successfully (no output)."
|
||||
|
||||
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}"
|
||||
|
||||
|
||||
@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 ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@ -2,9 +2,11 @@ from pathlib import Path
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
# ── Configuration ────────────────────────────────────────────────────────────
|
||||
project_dir = Path(__file__).resolve().parent.parent
|
||||
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")
|
||||
|
||||
|
||||
@ -39,63 +41,39 @@ def list_files() -> str:
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def read_file(path: str) -> str:
|
||||
"""Read the contents of a file.
|
||||
def get_file_tree(dir_path: str=ALLOWED_DIR) -> str:
|
||||
"""Get a tree representation of the project directory.
|
||||
|
||||
Args:
|
||||
path: Relative path to the file within the project directory.
|
||||
path: The directory path to display (default is the allowed directory).
|
||||
|
||||
Returns:
|
||||
The file content as a string, or an error message if the file cannot be read.
|
||||
A string representing the directory structure, similar to 'tree' command output.
|
||||
"""
|
||||
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)."
|
||||
|
||||
@mcp.tool()
|
||||
def write_file(path: str, content: str) -> str:
|
||||
"""Write content to a .py or .txt file in the allowed directory.
|
||||
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)
|
||||
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}"
|
||||
|
||||
|
||||
if resolved.suffix not in (".py", ".txt"):
|
||||
return f"ERROR: can only write .py and .txt files, got '{resolved.suffix}'."
|
||||
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)
|
||||
|
||||
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}"
|
||||
return _tree(dir_path)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@ -133,6 +111,106 @@ def search_files(query: str) -> str:
|
||||
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:
|
||||
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__":
|
||||
|
||||
@ -5,6 +5,7 @@ from mcp.server.fastmcp import FastMCP
|
||||
MAX_PAGE_LENGTH = 4000 # max characters to return from a fetched page
|
||||
REQUEST_TIMEOUT = 10 # seconds
|
||||
|
||||
# ── Create the MCP server ────────────────────────────────────────────────────
|
||||
mcp = FastMCP("WebSearchServer")
|
||||
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user