From 5183f9606aa29c73e3e4f0abe6b9b935631edb12 Mon Sep 17 00:00:00 2001 From: shenyue Date: Mon, 8 Dec 2025 14:30:55 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81pyproject.toml=E5=92=8Cyaml?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.py | 21 ++- src/client/MCPClient.py | 314 ++++++++++++++++++++++++++++++---------- 2 files changed, 249 insertions(+), 86 deletions(-) diff --git a/main.py b/main.py index 9ebaabb..5fa6cbf 100644 --- a/main.py +++ b/main.py @@ -3,7 +3,7 @@ import argparse def parse_args(): parser = argparse.ArgumentParser( - description="MCP Server initialization args", + description="MCP Testkit initialization args", formatter_class=argparse.ArgumentDefaultsHelpFormatter ) @@ -17,6 +17,12 @@ def parse_args(): default="./mcp-servers-perf.json", help="Path to MCP Server config file" ) + gen_parser.add_argument( + "--logdir", + type=str, + default=".logs", + help="Folderpath to log file" + ) #val-cases子命令 val_parser = subparsers.add_parser('val-cases', help='Validate test cases') @@ -49,23 +55,22 @@ def parse_args(): ) rep_parser.add_argument( "--detailed", - type=bool, - default=False, - help="Output detailed report or not" + action="store_true", + help="Output detailed report" ) return parser.parse_args() -async def gen_cases(config_path): +async def gen_cases(config_path, logdir): from src.test_generator.TestGenerator import TestGenerator - generator = TestGenerator(config_path=config_path) + generator = TestGenerator(config_path=config_path, log_name=logdir) return await generator.run() - async def val_cases(config_path, testcase_path): from src.validator.Response_validator_withenv import ResponseValidator_withenv validator = ResponseValidator_withenv(config_path=config_path, testcase_path=testcase_path) return await validator.run() + async def rep_cases(valpath, config_path, detailed): from src.reporter.Reporter import Reporter Reporter = Reporter(valpath, config_path, detailed) @@ -73,7 +78,7 @@ async def rep_cases(valpath, config_path, detailed): async def main(): args = parse_args() if args.command == 'gen-cases': - await gen_cases(args.config) + await gen_cases(args.config, args.logdir) if args.command == 'val-cases': await val_cases(args.config, args.testpath) if args.command == 'rep-cases': diff --git a/src/client/MCPClient.py b/src/client/MCPClient.py index a33f197..e289dd8 100644 --- a/src/client/MCPClient.py +++ b/src/client/MCPClient.py @@ -6,7 +6,7 @@ import subprocess import time from typing import Any, Optional from contextlib import AsyncExitStack - +import yaml import docker from docker.errors import NotFound, APIError @@ -35,8 +35,17 @@ class MCPClient: # Docker相关配置 self.use_docker = use_docker - self.abs_script_path = self.get_command_script_path() - self.host_mcp_path = self.abs_script_path.split('src')[0] if self.abs_script_path else "" + self.host_mcp_path = self.config.get("cwd") + self.abs_script_path = self.get_command_script_path() + + if not self.host_mcp_path and self.abs_script_path: + self.host_mcp_path = self.abs_script_path.split('src')[0] + + if self.host_mcp_path: + self.host_mcp_path = os.path.abspath(self.host_mcp_path) + else: + logging.warning("未找到有效的 host_mcp_path（cwd 未配置且脚本路径推导失败）") + self.container_mcp_path = "/app/" self.server_port = config.get("port", 8080) @@ -44,7 +53,6 @@ class MCPClient: self.docker_process = None self.container_id = None self.container_name = None - self.client = docker.from_env() # 初始化全局容器注册表 @@ -55,21 +63,27 @@ class MCPClient: def get_command_script_path(self) -> str: """获取命令脚本路径""" try: - server_args = self.config['args'] - work_dir = os.getcwd() + #config中的cwd → args中的--directory参数 → 当前工作目录（os.getcwd()） + server_args = self.config.get('args', []) + work_dir = self.config.get("cwd") + if work_dir: + work_dir = os.path.abspath(work_dir) source_file = None - for i, arg in enumerate(server_args): - if arg == "--directory" and i + 1 < len(server_args): - work_dir = os.path.abspath(server_args[i + 1]) - break + if not work_dir: + for i, arg in enumerate(server_args): + if arg == "--directory" and i + 1 < len(server_args): + work_dir = os.path.abspath(server_args[i + 1]) + break + if not work_dir: + work_dir = os.getcwd() for i, arg in enumerate(server_args): if arg.endswith(".py"): source_file = arg if i > 0 and server_args[i-1] == "run": break - elif arg == "run" and i + 1 < len(server_args) and server_args[i+1].endswith(".py"): + elif arg == "run" and i + 1 < len(server_args): source_file = server_args[i+1] break @@ -78,38 +92,50 @@ class MCPClient: return None if os.path.isabs(source_file): - absolute_path = source_file + absolute_paths = [source_file] else: - absolute_path = os.path.join(work_dir, source_file) + absolute_paths = [os.path.join(work_dir, source_file)] - if os.path.exists(absolute_path): - return absolute_path - else: - logging.error(f"源代码文件不存在：{absolute_path}") - return None + if not os.path.exists(absolute_paths[0]): + absolute_paths.append(f"{absolute_paths[0]}.py") + for abs_path in absolute_paths: + if os.path.exists(abs_path): + return abs_path + logging.error(f"源代码文件不存在（尝试路径：{absolute_paths}）") + return None except Exception as e: logging.error(f"获取脚本路径出错: {e}") - return None - + return None async def initialize(self) -> None: """初始化服务器""" if self._is_initialized: logging.warning(f"服务器 {self.name} 已经初始化") return - + TIMEOUT_SECONDS = 300 try: - logging.info(f"开始初始化服务器 {self.name}") - if self.use_docker: - await self._initialize_docker() + await asyncio.wait_for( + self._initialize_docker(), + timeout=TIMEOUT_SECONDS + ) else: - await self._initialize_host_server() + await asyncio.wait_for( + self._initialize_host_server(), + timeout=TIMEOUT_SECONDS + ) self._is_initialized = True + logging.info(f"服务器 {self.name} 初始化成功") + except asyncio.TimeoutError: + error_msg = f"服务器 {self.name} 初始化超时（超过{TIMEOUT_SECONDS}秒）" + logging.error(error_msg) + await self.cleanup() + raise + except Exception as e: - logging.error(f"初始化失败: {e}") + logging.error(f"服务器 {self.name} 初始化失败: {e}") await self.cleanup() raise @@ -152,6 +178,8 @@ class MCPClient: docker_cmd.extend([ "-v", f"{self.host_mcp_path}:{self.container_mcp_path}" ]) + docker_cmd.extend(["-v", "/var/run/docker.sock:/var/run/docker.sock"]) + docker_cmd.extend(["-v", "/usr/bin/docker:/usr/bin/docker"]) # 添加环境变量 env_vars = { @@ -167,7 +195,7 @@ class MCPClient: docker_cmd.extend(["-a", "stdout", "-a", "stderr"]) # 添加Docker镜像 - self.docker_image = "val:latest" + self.docker_image = "openeuler-167:latest" docker_cmd.append(self.docker_image) startup_script = self._build_correct_bash_script() @@ -178,53 +206,141 @@ class MCPClient: def _build_correct_bash_script(self) -> str: """构建启动脚本""" container_command = self._get_container_command() + system_deps_str = self._get_yaml_dependencies() + if not system_deps_str: + system_deps_str = "" + + self.env_script = self._clean_llm_generated_script(self.env_script) script = f'''set -e + echo "=== 快速环境检查 ===" >&2 + echo "Python版本: $(python --version 2>&1 || python3 --version 2>&1)" >&2 + echo "工作目录: $(pwd)" >&2 + echo "文件列表:" >&2 + ls -la >&2 + echo "" >&2 + + echo "=== 检查并安装项目依赖 ===" >&2 + + if command -v dnf &> /dev/null; then + PACKAGE_MANAGER="dnf" + elif command -v yum &> /dev/null; then + PACKAGE_MANAGER="yum" + else + echo "❌ 不支持的包管理器，无法安装系统依赖" >&2 + exit 1 + fi - echo "=== 快速环境检查 ===" - echo "Python版本: $(python --version)" - echo "工作目录: $(pwd)" - echo "文件列表:" - ls -la - echo "" - - # 只安装项目特定的新依赖 - echo "=== 检查并安装项目依赖 ===" - if [ -f requirements.txt ]; then - echo "发现requirements.txt文件" - pip install -qq --upgrade-strategy only-if-needed -r requirements.txt - if [ $? -eq 0 ]; then - echo "依赖安装完成（无异常）" + DEPS="{system_deps_str}" + if [ -n "$DEPS" ]; then + echo "安装系统依赖：$DEPS" >&2 + $PACKAGE_MANAGER install -y $DEPS >&2 2>&1 else - echo "依赖安装失败！（可去掉 -qq 参数重新执行查看详细错误）" + echo "✅ 无依赖需要安装" >&2 fi - else - echo "未找到requirements.txt文件" - fi - echo "=== 执行自定义环境部署 ===" - {self.env_script} + if [ -f "pyproject.toml" ]; then + echo "发现 pyproject.toml，使用 uv 安装依赖..." >&2 + if ! command -v uv &> /dev/null; then + echo "未找到 uv，正在安装..." >&2 + curl -sSf https://astral.sh/uv/install.sh | sh >&2 + export PATH="$HOME/.cargo/bin:$PATH" + fi + if [ ! -d ".venv" ]; then + echo "未发现虚拟环境，正在用 uv 创建..." >&2 + export UV_PYTHON_DOWNLOAD_MIRROR="https://mirrors.tuna.tsinghua.edu.cn/python/" + export UV_INDEX_URL="https://mirrors.aliyun.com/pypi/simple/" + uv venv >&2 + fi + source .venv/bin/activate + + uv pip install --quiet . + if [ $? -eq 0 ]; then + echo "pyproject.toml 依赖安装完成（无异常）" >&2 + else + echo "uv 安装依赖失败！（可去掉 --quiet 参数重新执行查看详细错误）" >&2 + exit 1 + fi + + elif [ -f "requirements.txt" ]; then + echo "发现 requirements.txt，使用 pip 安装依赖..." >&2 + pip install -qq --upgrade-strategy only-if-needed -r requirements.txt + if [ $? -eq 0 ]; then + echo "requirements.txt 依赖安装完成（无异常）" >&2 + else + echo "pip 安装依赖失败！（可去掉 -qq 参数重新执行查看详细错误）" >&2 + exit 1 + fi + + elif [ -f "src/requirements.txt" ]; then + echo "发现 src/requirements.txt，使用 pip 安装依赖..." >&2 + pip install -qq --upgrade-strategy only-if-needed -r src/requirements.txt + if [ $? -eq 0 ]; then + echo "src/requirements.txt 依赖安装完成（无异常）" >&2 + else + echo "pip 安装依赖失败！（可去掉 -qq 参数重新执行查看详细错误）" >&2 + exit 1 + fi - echo "=== 启动MCP服务器 ===" - echo "执行命令: {container_command}" - exec {container_command}''' + else + echo "未找到 pyproject.toml 或 requirements.txt，跳过依赖安装" >&2 + fi + + echo "=== 执行自定义环境部署 ===" >&2 + {self.env_script} >&2 + echo "=== 启动MCP服务器 ===" >&2 + echo "执行命令: {container_command}" >&2 + exec {container_command}''' return script def _get_container_command(self) -> str: """获取容器内的命令字符串""" command = self.config.get("command", "python") - if command == "uv": - command = "uv run" - script_rel_path = 'src'+self.abs_script_path.split('src')[-1] - return command + " " + script_rel_path + args = self.config.get("args", []) + command_parts = [command] + args + return " ".join(command_parts) + + def _get_yaml_dependencies(self) -> Optional[str]: + """ + 从 mcp-rpm.yaml 中提取系统依赖和 Python 包依赖，合并为字符串返回 + :return: 合并后的依赖字符串（格式："依赖1, 依赖2, ..."），失败返回 None + """ + if not self.host_mcp_path or not os.path.isdir(self.host_mcp_path): + logging.warning("主机 MCP 路径为空或不存在，无法提取依赖") + return None + + existing_deps = ['uv', 'jq', 'python3', 'python3-mcp'] + mcp_yaml_path = os.path.join(self.host_mcp_path, "mcp-rpm.yaml") + + if not os.path.exists(mcp_yaml_path): + logging.warning(f"未找到系统配置文件：{mcp_yaml_path}") + return None + try: + with open(mcp_yaml_path, "r", encoding="utf-8") as f: + mcp_rpm_config = yaml.safe_load(f) + except yaml.YAMLError as e: + logging.error(f"解析 {mcp_yaml_path} 失败：YAML 格式错误 - {str(e)}") + return None + except Exception as e: + logging.error(f"读取 {mcp_yaml_path} 失败：{str(e)}") + return None + + mcp_dependencies = mcp_rpm_config.get("dependencies", {}) + system_deps = mcp_dependencies.get("system", []) + package_deps = mcp_dependencies.get("packages", []) + + combined_deps = list(set(system_deps + package_deps)) + if not combined_deps: + logging.info(f"{mcp_yaml_path} 中未配置任何依赖") + return "" + + combined_deps = [dep for dep in combined_deps if dep not in existing_deps] + return " ".join(combined_deps) async def _initialize_docker(self): """初始化Docker中的MCP服务器，支持输出显示""" - original_docker_command = self._build_docker_command() - - # 修改Docker命令，使用tee同时输出到终端和MCP - docker_command = self._add_output_redirection(original_docker_command) + docker_command = self._build_docker_command() logging.info(f"启动Docker命令: {' '.join(docker_command)}") @@ -269,27 +385,69 @@ class MCPClient: except Exception as e: logging.error(f"Docker MCP服务器初始化失败: {str(e)}") raise + def _clean_llm_generated_script(self, llm_script: str) -> str: + """ + 自动化净化大模型生成的 Bash 脚本 + 核心优化： + 1. 去重 dnf/yum 的 -y 参数（避免重复） + 2. 排除 exit 命令（不添加重定向，确保正常退出） + 3. 精准匹配命令，不干扰代码块和关键字 + 4. 保留所有核心功能，仅净化输出 + """ + if not llm_script: + return "" + + lines = llm_script.split('\n') + cleaned_lines = [] + + ignore_patterns = [ + r'^\s*$', # 空行 + r'^\s*#', # 注释行 + r'^\s*(if|then|else|fi|case|esac|for|do|done|while|until)', # 代码块关键字 + r'^\s*export\s+', # 变量导出 + r'^\s*[A-Za-z0-9_]+\s*=', # 变量赋值 + r'^\s*exit\s+\d+', # exit 命令（不添加重定向） + ] + ignore_re = re.compile('|'.join(ignore_patterns)) - def _add_output_redirection(self, docker_command): - """为Docker命令添加输出重定向""" - # 找到bash脚本部分 - if len(docker_command) >= 3 and docker_command[-2] == "-c": - # 修改bash脚本，添加tee命令 - original_script = docker_command[-1] - - # 将输出同时发送到stderr（显示在终端）和stdout（给MCP） - modified_script = f''' - # 设置输出重定向 - exec > >(tee /dev/stderr) - exec 2>&1 - - # 原始脚本 - {original_script} - ''' - new_command = docker_command[:-1] + [modified_script] - return new_command - - return docker_command + pkg_manager_re = re.compile(r'^\s*(dnf|yum)\s+(install|remove)\s+', re.IGNORECASE) + pkg_flag_re = re.compile(r'(-y|--yes|-q|--quiet)\s+', re.IGNORECASE) + + echo_re = re.compile(r'^\s*echo\s+', re.IGNORECASE) + + redirected_re = re.compile(r'(>\&2|\/dev\/null|\|)') + + for line in lines: + line_stripped = line.strip() + + if ignore_re.match(line): + cleaned_lines.append(line) + continue + + pkg_match = pkg_manager_re.match(line) + if pkg_match: + cmd = pkg_flag_re.sub('', line) + cmd = pkg_manager_re.sub(r'\1 \2 -q -y ', cmd) + if not redirected_re.search(cmd): + cmd += ' >&2 2>&1 || true' + cleaned_lines.append(cmd) + continue + + if echo_re.match(line): + if not redirected_re.search(line): + line += ' >&2' + cleaned_lines.append(line) + continue + + if not redirected_re.search(line) and 'exit' not in line_stripped: + line += ' >&2 2>&1 || true' + cleaned_lines.append(line) + + cleaned_script = '\n'.join(cleaned_lines) + cleaned_script = re.sub(r'\n+', '\n', cleaned_script) + cleaned_script = re.sub(r'^\n', '', cleaned_script) + + return cleaned_script async def _cleanup_existing_container(self): """清理可能存在的同名容器""" -- Gitee