{
"tool_response": {
"filePath": "\/var\/www\/mcp-servers\/mcp-tasks\/tools\/quality_tools.py",
"oldString": "timeout=180,",
"newString": "timeout=SEMGREP_TIMEOUT,",
"originalFile": "\"\"\"Quality Tools für MCP-Tasks Server\"\"\"\nimport sys\nimport time\nimport json\nimport subprocess\nfrom typing import Optional\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-tasks\")\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/shared\")\n\nfrom constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, LLM_TIMEOUT, MS_PER_SECOND, SEMGREP_TIMEOUT\n\nfrom config import Config\nfrom domain.contracts import LogEntry, QualityCheckResult\nfrom infrastructure.protokoll_logger import get_logger\n\n\ndef register_quality_tools(mcp):\n \"\"\"Registriert alle Quality-Tools\"\"\"\n\n logger = get_logger()\n\n # ==================== quality_check ====================\n @mcp.tool()\n def quality_check(\n path: str = \"\/var\/www\/dev.campus.systemische-tools.de\",\n checks: str = \"all\",\n fix: bool = False,\n ) -> dict:\n \"\"\"\n Führt PHP-Quality-Prüfungen aus.\n\n Args:\n path: Zu prüfender Pfad\n checks: Welche Checks (phpstan, cs-fixer, semgrep, all)\n fix: Style-Probleme automatisch korrigieren (nur cs-fixer)\n\n Returns:\n Ergebnis aller Checks mit Details\n \"\"\"\n start = time.time()\n request_str = json.dumps({\"path\": path, \"checks\": checks, \"fix\": fix})\n\n try:\n results = {}\n overall_passed = True\n\n check_list = [\"phpstan\", \"cs-fixer\", \"semgrep\"] if checks == \"all\" else [checks]\n\n for check_name in check_list:\n check_result = QualityCheckResult(check_name=check_name)\n\n if check_name == \"phpstan\":\n try:\n result = subprocess.run(\n [Config.PHPSTAN_BIN, \"analyse\", path, \"--level=5\", \"--no-progress\", \"--error-format=json\"],\n capture_output=True,\n text=True,\n timeout=LLM_TIMEOUT,\n )\n if result.returncode == 0:\n check_result.passed = True\n check_result.issues = 0\n else:\n check_result.passed = False\n try:\n phpstan_output = json.loads(result.stdout)\n check_result.issues = phpstan_output.get(\"totals\", {}).get(\"errors\", 0)\n except Exception:\n check_result.issues = 1\n check_result.details = result.stdout[:500]\n except subprocess.TimeoutExpired:\n check_result.passed = False\n check_result.details = \"PHPStan timeout\"\n except Exception as e:\n check_result.passed = False\n check_result.details = str(e)[:200]\n\n elif check_name == \"cs-fixer\":\n try:\n cmd = [Config.CS_FIXER_BIN, \"fix\" if fix else \"check\", path, \"--dry-run\" if not fix else \"\"]\n cmd = [c for c in cmd if c]\n result = subprocess.run(\n cmd,\n capture_output=True,\n text=True,\n timeout=LLM_TIMEOUT,\n )\n check_result.passed = result.returncode == 0\n check_result.issues = result.stdout.count(\"CHANGED\") if not fix else 0\n check_result.fixed = result.stdout.count(\"CHANGED\") if fix else 0\n except Exception as e:\n check_result.passed = False\n check_result.details = str(e)[:200]\n\n elif check_name == \"semgrep\":\n try:\n result = subprocess.run(\n [Config.SEMGREP_BIN, \"--config=auto\", \"--json\", path],\n capture_output=True,\n text=True,\n timeout=180,\n )\n try:\n semgrep_output = json.loads(result.stdout)\n issues = len(semgrep_output.get(\"results\", []))\n check_result.passed = issues == 0\n check_result.issues = issues\n except Exception:\n check_result.passed = result.returncode == 0\n except subprocess.TimeoutExpired:\n check_result.passed = False\n check_result.details = \"Semgrep timeout\"\n except Exception as e:\n check_result.passed = False\n check_result.details = str(e)[:200]\n\n results[check_name] = check_result.to_dict()\n if not check_result.passed:\n overall_passed = False\n\n duration = int((time.time() - start) * MS_PER_SECOND)\n logger.log(LogEntry(\n tool_name=\"quality_check\",\n request=request_str,\n status=\"success\" if overall_passed else \"error\",\n duration_ms=duration,\n ))\n\n return {\n \"success\": True,\n \"passed\": overall_passed,\n \"path\": path,\n \"results\": results,\n }\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n logger.log(LogEntry(\n tool_name=\"quality_check\",\n request=request_str,\n status=\"error\",\n duration_ms=duration,\n error_message=str(e)[:200],\n ))\n return {\"success\": False, \"error\": str(e)}\n\n # ==================== quality_report ====================\n @mcp.tool()\n def quality_report(\n scope: str = \"full\",\n format: str = \"json\",\n ) -> dict:\n \"\"\"\n Erstellt einen vollständigen Qualitätsbericht.\n\n Args:\n scope: full oder changes_only\n format: json oder markdown\n\n Returns:\n Vollständiger Report über alle Prüfungen\n \"\"\"\n start = time.time()\n request_str = json.dumps({\"scope\": scope, \"format\": format})\n\n try:\n report = {\n \"generated_at\": time.strftime(\"%Y-%m-%d %H:%M:%S\"),\n \"scope\": scope,\n \"dev_path\": \"\/var\/www\/dev.campus.systemische-tools.de\",\n }\n\n # Quality-Checks ausführen\n quality_result = quality_check(\n path=\"\/var\/www\/dev.campus.systemische-tools.de\/src\",\n checks=\"all\",\n )\n report[\"quality_checks\"] = quality_result.get(\"results\", {})\n report[\"quality_passed\"] = quality_result.get(\"passed\", False)\n\n # Zusammenfassung\n report[\"all_passed\"] = report[\"quality_passed\"]\n\n duration = int((time.time() - start) * MS_PER_SECOND)\n logger.log(LogEntry(\n tool_name=\"quality_report\",\n request=request_str,\n status=\"success\",\n duration_ms=duration,\n ))\n\n # Markdown-Format\n if format == \"markdown\":\n md = \"# Quality Report\\n\\n\"\n md += f\"Generated: {report['generated_at']}\\n\\n\"\n md += f\"## Overall Status: {'PASSED' if report['all_passed'] else 'FAILED'}\\n\\n\"\n\n md += \"## Quality Checks\\n\\n\"\n for check_name, check_data in report[\"quality_checks\"].items():\n status = \"PASS\" if check_data.get(\"passed\") else \"FAIL\"\n md += f\"- {status} {check_name}: {check_data.get('issues', 0)} issues\\n\"\n\n return {\n \"success\": True,\n \"report\": md,\n \"format\": \"markdown\",\n }\n\n return {\n \"success\": True,\n \"report\": report,\n \"format\": \"json\",\n }\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n logger.log(LogEntry(\n tool_name=\"quality_report\",\n request=request_str,\n status=\"error\",\n duration_ms=duration,\n error_message=str(e)[:200],\n ))\n return {\"success\": False, \"error\": str(e)}\n",
"structuredPatch": [
{
"oldStart": 99,
"oldLines": 7,
"newStart": 99,
"newLines": 7,
"lines": [
" [Config.SEMGREP_BIN, \"--config=auto\", \"--json\", path],",
" capture_output=True,",
" text=True,",
"- timeout=180,",
"+ timeout=SEMGREP_TIMEOUT,",
" )",
" try:",
" semgrep_output = json.loads(result.stdout)"
]
}
],
"userModified": false,
"replaceAll": false
}
}