{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/opt\/mcp-servers\/mcp-db\/tools\/describe_tool.py",
"content": "\"\"\"Describe Tool - Zeigt Tabellen-Struktur an.\"\"\"\nimport re\nimport sys\nimport time\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-db\")\n\nfrom config import Config\nfrom domain.log_contract import LogEntry\nfrom infrastructure.db_connection import DatabaseConnection\nfrom infrastructure.protokoll_logger import ProtokollLogger\n\n\ndef register_describe_tool(mcp) -> None:\n \"\"\"Registriert db_describe Tool.\"\"\"\n logger = ProtokollLogger()\n\n @mcp.tool()\n def db_describe(\n table: str,\n database: str = \"ki_dev\",\n show_create: bool = False,\n ) -> dict:\n \"\"\"\n Zeigt Tabellen-Struktur an.\n\n Args:\n table: Tabellenname\n database: Zieldatenbank (ki_dev oder ki_content)\n show_create: True = SHOW CREATE TABLE, False = DESCRIBE\n\n Returns:\n Dict mit columns oder create_statement\n \"\"\"\n start = time.time()\n\n # Validierung: Tabellenname (SQL-Injection-Schutz)\n if not re.match(r\"^[a-zA-Z0-9_]+$\", table):\n return {\n \"status\": \"denied\",\n \"error\": \"Invalid table name. Only alphanumeric and underscore allowed.\",\n }\n\n # Validierung: Datenbank\n if database not in Config.ALLOWED_DATABASES:\n return {\n \"status\": \"denied\",\n \"error\": f\"Database '{database}' not allowed.\",\n }\n\n try:\n with DatabaseConnection.get_connection(database) as conn:\n cursor = conn.cursor(dictionary=True, buffered=True)\n\n if show_create:\n cursor.execute(f\"SHOW CREATE TABLE `{table}`\")\n row = cursor.fetchone()\n cursor.close()\n\n duration = int((time.time() - start) * 1000)\n\n # Log\n try:\n logger.log(\n LogEntry(\n request=f\"SHOW CREATE TABLE {table}\",\n status=\"success\",\n duration_ms=duration,\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"success\",\n \"table\": table,\n \"create_statement\": row.get(\"Create Table\", \"\") if row else \"\",\n \"execution_ms\": duration,\n }\n else:\n cursor.execute(f\"DESCRIBE `{table}`\")\n columns = cursor.fetchall()\n cursor.close()\n\n duration = int((time.time() - start) * 1000)\n\n # Log\n try:\n logger.log(\n LogEntry(\n request=f\"DESCRIBE {table}\",\n status=\"success\",\n duration_ms=duration,\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"success\",\n \"table\": table,\n \"columns\": columns,\n \"execution_ms\": duration,\n }\n\n except Exception as e:\n duration = int((time.time() - start) * 1000)\n\n try:\n logger.log(\n LogEntry(\n request=f\"DESCRIBE {table}\",\n status=\"error\",\n duration_ms=duration,\n error_message=str(e)[:100],\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"error\",\n \"error\": str(e)[:200],\n \"execution_ms\": duration,\n }\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/opt\/mcp-servers\/mcp-db\/tools\/describe_tool.py",
"content": "\"\"\"Describe Tool - Zeigt Tabellen-Struktur an.\"\"\"\nimport re\nimport sys\nimport time\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-db\")\n\nfrom config import Config\nfrom domain.log_contract import LogEntry\nfrom infrastructure.db_connection import DatabaseConnection\nfrom infrastructure.protokoll_logger import ProtokollLogger\n\n\ndef register_describe_tool(mcp) -> None:\n \"\"\"Registriert db_describe Tool.\"\"\"\n logger = ProtokollLogger()\n\n @mcp.tool()\n def db_describe(\n table: str,\n database: str = \"ki_dev\",\n show_create: bool = False,\n ) -> dict:\n \"\"\"\n Zeigt Tabellen-Struktur an.\n\n Args:\n table: Tabellenname\n database: Zieldatenbank (ki_dev oder ki_content)\n show_create: True = SHOW CREATE TABLE, False = DESCRIBE\n\n Returns:\n Dict mit columns oder create_statement\n \"\"\"\n start = time.time()\n\n # Validierung: Tabellenname (SQL-Injection-Schutz)\n if not re.match(r\"^[a-zA-Z0-9_]+$\", table):\n return {\n \"status\": \"denied\",\n \"error\": \"Invalid table name. Only alphanumeric and underscore allowed.\",\n }\n\n # Validierung: Datenbank\n if database not in Config.ALLOWED_DATABASES:\n return {\n \"status\": \"denied\",\n \"error\": f\"Database '{database}' not allowed.\",\n }\n\n try:\n with DatabaseConnection.get_connection(database) as conn:\n cursor = conn.cursor(dictionary=True, buffered=True)\n\n if show_create:\n cursor.execute(f\"SHOW CREATE TABLE `{table}`\")\n row = cursor.fetchone()\n cursor.close()\n\n duration = int((time.time() - start) * 1000)\n\n # Log\n try:\n logger.log(\n LogEntry(\n request=f\"SHOW CREATE TABLE {table}\",\n status=\"success\",\n duration_ms=duration,\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"success\",\n \"table\": table,\n \"create_statement\": row.get(\"Create Table\", \"\") if row else \"\",\n \"execution_ms\": duration,\n }\n else:\n cursor.execute(f\"DESCRIBE `{table}`\")\n columns = cursor.fetchall()\n cursor.close()\n\n duration = int((time.time() - start) * 1000)\n\n # Log\n try:\n logger.log(\n LogEntry(\n request=f\"DESCRIBE {table}\",\n status=\"success\",\n duration_ms=duration,\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"success\",\n \"table\": table,\n \"columns\": columns,\n \"execution_ms\": duration,\n }\n\n except Exception as e:\n duration = int((time.time() - start) * 1000)\n\n try:\n logger.log(\n LogEntry(\n request=f\"DESCRIBE {table}\",\n status=\"error\",\n duration_ms=duration,\n error_message=str(e)[:100],\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"error\",\n \"error\": str(e)[:200],\n \"execution_ms\": duration,\n }\n",
"structuredPatch": [],
"originalFile": null
}
}