Docstring anpassungen mcp server

This commit is contained in:
Irina Rueegg 2026-05-06 19:25:48 +02:00
parent 2a65efb046
commit 2a37ae3aa4
5 changed files with 59 additions and 26 deletions

View File

@ -1,6 +1,6 @@
import asyncio import asyncio
import json import json
import os # import os
import numpy as np import numpy as np
from typing import List, Dict, Any from typing import List, Dict, Any
from pathlib import Path from pathlib import Path

View File

@ -1,15 +1,15 @@
{"Filemanager": { {"FileSearchServer": {
"command": "python", "command": "python",
"args": ["servers/mcp_server_file_search.py"] "args": ["servers/mcp_server_file_search.py"]
}, },
"WebSearch": { "WebSearchServer": {
"command": "python", "command": "python",
"args": ["servers/mcp_server_web_search.py"], "args": ["servers/mcp_server_web_search.py"],
"env": { "env": {
"DDGS_API_KEY": "your_ddgs_api_key_here" "DDGS_API_KEY": "your_ddgs_api_key_here"
} }
}, },
"CodeExecution": { "CodeExecutionServer": {
"command": "python", "command": "python",
"args": ["servers/mcp_server_code_execution.py"] "args": ["servers/mcp_server_code_execution.py"]
} }

View File

@ -6,7 +6,7 @@ from mcp.server.fastmcp import FastMCP
EXEC_TIMEOUT = 10 # seconds before killing the subprocess EXEC_TIMEOUT = 10 # seconds before killing the subprocess
MAX_OUTPUT_LENGTH = 3000 # max characters of stdout+stderr to return MAX_OUTPUT_LENGTH = 3000 # max characters of stdout+stderr to return
mcp = FastMCP("SafeExecServer") mcp = FastMCP("CodeExecutionServer")
BLOCKED_IMPORTS = { BLOCKED_IMPORTS = {
# Filesystem access: # Filesystem access:
@ -41,15 +41,11 @@ def check_code_safety(code: str) -> str | None:
""" """
Statically analyze Python code for forbidden imports and builtins with ast. Statically analyze Python code for forbidden imports and builtins with ast.
Args:
code: The Python code to analyze.
Parameters Returns:
---------- str or None
code : str
The Python code to check.
Returns
-------
str or None
Error message if forbidden code found, None if safe. Error message if forbidden code found, None if safe.
""" """
@ -79,21 +75,14 @@ def run_python_sandboxed(code: str) -> str:
""" """
Run Python code in a sandboxed subprocess. Run Python code in a sandboxed subprocess.
Defense layers: Static analysis (check_code_safety), Subprocess isolation (child process)
1. Static analysis (check_code_safety) Timeout (killed after EXEC_TIMEOUT seconds), Output truncation (max MAX_OUTPUT_LENGTH chars)
2. Subprocess isolation (child process)
3. Timeout (killed after EXEC_TIMEOUT seconds)
4. Output truncation (max MAX_OUTPUT_LENGTH chars)
Parameters Args:
---------- code: The Python code to execute in str format.
code : str
The Python code to execute.
Returns Returns:
------- Combined stdout+stderr, or an error message in str format.
str
Combined stdout+stderr, or an error message.
""" """
static_safety = check_code_safety(code) static_safety = check_code_safety(code)

View File

@ -44,6 +44,9 @@ def read_file(path: str) -> str:
Args: Args:
path: Relative path to the file within the project directory. 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: try:
resolved = _safe_path(path) resolved = _safe_path(path)
@ -60,6 +63,40 @@ def read_file(path: str) -> str:
except UnicodeDecodeError: except UnicodeDecodeError:
return f"Error: '{path}' is not a text file (binary content)." 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)
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}'."
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() @mcp.tool()
def search_files(query: str) -> str: def search_files(query: str) -> str:
@ -67,6 +104,9 @@ def search_files(query: str) -> str:
Args: Args:
query: The search term (case-insensitive). query: The search term (case-insensitive).
Returns:
A formatted string of search results, or a message if no matches found.
""" """
query_lower = query.lower() query_lower = query.lower()
results = [] results = []

View File

@ -47,6 +47,8 @@ def web_search(query: str, max_results: int = 5) -> str:
Args: Args:
query: The search query. query: The search query.
max_results: Maximum number of results to return (default 5). 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: try:
from ddgs import DDGS from ddgs import DDGS
@ -74,6 +76,8 @@ def fetch_page(url: str) -> str:
Args: Args:
url: The URL to fetch. url: The URL to fetch.
Returns:
The text content of the fetched page, or an error message.
""" """
try: try:
url = _validate_url(url) url = _validate_url(url)