{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp_db\/tools\/tables_tool.py",
"content": "\"\"\"Tables Tool - Listet Tabellen einer Datenbank auf.\"\"\"\n\nimport time\nfrom datetime import datetime\n\nfrom constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, MS_PER_SECOND\n\nfrom config import Config\nfrom infrastructure.db_connection import DatabaseConnection\nfrom shared.domain import LogEntry\nfrom shared.infrastructure import get_logger\n\n\ndef register_tables_tool(mcp) -> None:\n \"\"\"Registriert db_tables Tool.\"\"\"\n logger = get_logger(\"mcp-db\", Config)\n\n @mcp.tool()\n def db_tables(\n database: str = \"ki_dev\",\n include_row_count: bool = False,\n ) -> dict:\n \"\"\"\n Listet alle Tabellen einer Datenbank auf.\n\n Args:\n database: Zieldatenbank (ki_dev oder ki_content)\n include_row_count: Row-Count pro Tabelle inkludieren\n\n Returns:\n Dict mit tables (Liste der Tabellen)\n \"\"\"\n start = time.time()\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 include_row_count:\n cursor.execute(\n \"\"\"SELECT table_name, table_rows\n FROM information_schema.tables\n WHERE table_schema = %s\n ORDER BY table_name\"\"\",\n (database,),\n )\n rows = cursor.fetchall()\n tables = [\n {\"name\": r[\"table_name\"], \"row_count\": r[\"table_rows\"] or 0}\n for r in rows\n ]\n else:\n cursor.execute(f\"SHOW TABLES FROM `{database}`\")\n rows = cursor.fetchall()\n # SHOW TABLES returns dict with key like 'Tables_in_ki_dev'\n tables = [\n {\"name\": list(r.values())[0]}\n for r in rows\n ]\n\n cursor.close()\n\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log\n try:\n logger.log(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_tables\",\n request=f\"SHOW TABLES FROM {database}\",\n status=\"success\",\n duration_ms=duration,\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"success\",\n \"database\": database,\n \"tables\": tables,\n \"table_count\": len(tables),\n \"execution_ms\": duration,\n }\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n try:\n logger.log(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_tables\",\n request=f\"SHOW TABLES FROM {database}\",\n status=\"error\",\n duration_ms=duration,\n error_message=str(e)[:LOG_ENTRY_MAX_LENGTH],\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"error\",\n \"error\": str(e)[:LOG_QUERY_MAX_LENGTH],\n \"execution_ms\": duration,\n }\n"
}
}
{
"tool_response": {
"type": "update",
"filePath": "\/var\/www\/mcp-servers\/mcp_db\/tools\/tables_tool.py",
"content": "\"\"\"Tables Tool - Listet Tabellen einer Datenbank auf.\"\"\"\n\nimport time\nfrom datetime import datetime\n\nfrom constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, MS_PER_SECOND\n\nfrom config import Config\nfrom infrastructure.db_connection import DatabaseConnection\nfrom shared.domain import LogEntry\nfrom shared.infrastructure import get_logger\n\n\ndef register_tables_tool(mcp) -> None:\n \"\"\"Registriert db_tables Tool.\"\"\"\n logger = get_logger(\"mcp-db\", Config)\n\n @mcp.tool()\n def db_tables(\n database: str = \"ki_dev\",\n include_row_count: bool = False,\n ) -> dict:\n \"\"\"\n Listet alle Tabellen einer Datenbank auf.\n\n Args:\n database: Zieldatenbank (ki_dev oder ki_content)\n include_row_count: Row-Count pro Tabelle inkludieren\n\n Returns:\n Dict mit tables (Liste der Tabellen)\n \"\"\"\n start = time.time()\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 include_row_count:\n cursor.execute(\n \"\"\"SELECT table_name, table_rows\n FROM information_schema.tables\n WHERE table_schema = %s\n ORDER BY table_name\"\"\",\n (database,),\n )\n rows = cursor.fetchall()\n tables = [\n {\"name\": r[\"table_name\"], \"row_count\": r[\"table_rows\"] or 0}\n for r in rows\n ]\n else:\n cursor.execute(f\"SHOW TABLES FROM `{database}`\")\n rows = cursor.fetchall()\n # SHOW TABLES returns dict with key like 'Tables_in_ki_dev'\n tables = [\n {\"name\": list(r.values())[0]}\n for r in rows\n ]\n\n cursor.close()\n\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log\n try:\n logger.log(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_tables\",\n request=f\"SHOW TABLES FROM {database}\",\n status=\"success\",\n duration_ms=duration,\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"success\",\n \"database\": database,\n \"tables\": tables,\n \"table_count\": len(tables),\n \"execution_ms\": duration,\n }\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n try:\n logger.log(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_tables\",\n request=f\"SHOW TABLES FROM {database}\",\n status=\"error\",\n duration_ms=duration,\n error_message=str(e)[:LOG_ENTRY_MAX_LENGTH],\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"error\",\n \"error\": str(e)[:LOG_QUERY_MAX_LENGTH],\n \"execution_ms\": duration,\n }\n",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 21,
"newStart": 1,
"newLines": 19,
"lines": [
" \"\"\"Tables Tool - Listet Tabellen einer Datenbank auf.\"\"\"",
"-import sys",
"+",
" import time",
"+from datetime import datetime",
" ",
"-sys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_db\")",
"-sys.path.insert(0, \"\/var\/www\/mcp-servers\/shared\")",
"-",
" from constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, MS_PER_SECOND",
" ",
" from config import Config",
"-from domain.log_contract import LogEntry",
" from infrastructure.db_connection import DatabaseConnection",
"-from infrastructure.protokoll_logger import ProtokollLogger",
"+from shared.domain import LogEntry",
"+from shared.infrastructure import get_logger",
" ",
" ",
" def register_tables_tool(mcp) -> None:",
" \"\"\"Registriert db_tables Tool.\"\"\"",
"- logger = ProtokollLogger()",
"+ logger = get_logger(\"mcp-db\", Config)",
" ",
" @mcp.tool()",
" def db_tables("
]
},
{
"oldStart": 75,
"oldLines": 6,
"newStart": 73,
"newLines": 9,
"lines": [
" try:",
" logger.log(",
" LogEntry(",
"+ timestamp=datetime.now(),",
"+ client_name=\"mcp-db\",",
"+ tool_name=\"db_tables\",",
" request=f\"SHOW TABLES FROM {database}\",",
" status=\"success\",",
" duration_ms=duration,"
]
},
{
"oldStart": 97,
"oldLines": 6,
"newStart": 98,
"newLines": 9,
"lines": [
" try:",
" logger.log(",
" LogEntry(",
"+ timestamp=datetime.now(),",
"+ client_name=\"mcp-db\",",
"+ tool_name=\"db_tables\",",
" request=f\"SHOW TABLES FROM {database}\",",
" status=\"error\",",
" duration_ms=duration,"
]
}
],
"originalFile": "\"\"\"Tables Tool - Listet Tabellen einer Datenbank auf.\"\"\"\nimport sys\nimport time\n\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_db\")\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/shared\")\n\nfrom constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, MS_PER_SECOND\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_tables_tool(mcp) -> None:\n \"\"\"Registriert db_tables Tool.\"\"\"\n logger = ProtokollLogger()\n\n @mcp.tool()\n def db_tables(\n database: str = \"ki_dev\",\n include_row_count: bool = False,\n ) -> dict:\n \"\"\"\n Listet alle Tabellen einer Datenbank auf.\n\n Args:\n database: Zieldatenbank (ki_dev oder ki_content)\n include_row_count: Row-Count pro Tabelle inkludieren\n\n Returns:\n Dict mit tables (Liste der Tabellen)\n \"\"\"\n start = time.time()\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 include_row_count:\n cursor.execute(\n \"\"\"SELECT table_name, table_rows\n FROM information_schema.tables\n WHERE table_schema = %s\n ORDER BY table_name\"\"\",\n (database,),\n )\n rows = cursor.fetchall()\n tables = [\n {\"name\": r[\"table_name\"], \"row_count\": r[\"table_rows\"] or 0}\n for r in rows\n ]\n else:\n cursor.execute(f\"SHOW TABLES FROM `{database}`\")\n rows = cursor.fetchall()\n # SHOW TABLES returns dict with key like 'Tables_in_ki_dev'\n tables = [\n {\"name\": list(r.values())[0]}\n for r in rows\n ]\n\n cursor.close()\n\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log\n try:\n logger.log(\n LogEntry(\n request=f\"SHOW TABLES FROM {database}\",\n status=\"success\",\n duration_ms=duration,\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"success\",\n \"database\": database,\n \"tables\": tables,\n \"table_count\": len(tables),\n \"execution_ms\": duration,\n }\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n try:\n logger.log(\n LogEntry(\n request=f\"SHOW TABLES FROM {database}\",\n status=\"error\",\n duration_ms=duration,\n error_message=str(e)[:LOG_ENTRY_MAX_LENGTH],\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"error\",\n \"error\": str(e)[:LOG_QUERY_MAX_LENGTH],\n \"execution_ms\": duration,\n }\n"
}
}