From 560e56b596709fc268e1441a64a708d9f90275a4 Mon Sep 17 00:00:00 2001 From: Irina Rueegg Date: Sun, 10 May 2026 15:52:34 +0200 Subject: [PATCH] MCP-Tool fine tuning --- backend/agent/coding_agent.py | 265 ++++++------------ .../servers/mcp_server_code_execution.py | 195 ++++++++++++- .../agent/servers/mcp_server_file_search.py | 176 ++++++++---- .../agent/servers/mcp_server_web_search.py | 1 + 4 files changed, 399 insertions(+), 238 deletions(-) diff --git a/backend/agent/coding_agent.py b/backend/agent/coding_agent.py index 8631f53..0c6057d 100644 --- a/backend/agent/coding_agent.py +++ b/backend/agent/coding_agent.py @@ -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}": ""') -# else: -# params.append(f'"{pname}": ""') -# param_str = ", ".join(params) -# doc = (func.__doc__ or "").strip().split("\n")[0] -# lines.append(f" - {name}({{{param_str}}}): {doc}") -# return "\n".join(lines) -# -# -#def dispatch_tool(tool_name: str, arguments: dict) -> str: -# """Call a tool by name with the given arguments.""" -# if tool_name not in TOOL_REGISTRY: -# return f"ERROR: unknown tool '{tool_name}'. Available: {list(TOOL_REGISTRY.keys())}" -# func = TOOL_REGISTRY[tool_name] -# try: -# return func(**arguments) -# except TypeError as e: -# return f"ERROR calling {tool_name}: {e}" -# 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,22 +61,7 @@ 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) async def dispatch_tool(tool_name: str, arguments: dict) -> str: @@ -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.""" @@ -415,13 +337,7 @@ class CodingAgent: "max_tokens": 4096, "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.", diff --git a/backend/agent/servers/mcp_server_code_execution.py b/backend/agent/servers/mcp_server_code_execution.py index 62d1879..4d04a75 100644 --- a/backend/agent/servers/mcp_server_code_execution.py +++ b/backend/agent/servers/mcp_server_code_execution.py @@ -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="", 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 ─────────────────────────────────────────────────────────── diff --git a/backend/agent/servers/mcp_server_file_search.py b/backend/agent/servers/mcp_server_file_search.py index 8be2b59..7020593 100644 --- a/backend/agent/servers/mcp_server_file_search.py +++ b/backend/agent/servers/mcp_server_file_search.py @@ -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}'." - - 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}" + + 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() @@ -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__": diff --git a/backend/agent/servers/mcp_server_web_search.py b/backend/agent/servers/mcp_server_web_search.py index 0c15132..39c9fb4 100644 --- a/backend/agent/servers/mcp_server_web_search.py +++ b/backend/agent/servers/mcp_server_web_search.py @@ -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")