{
"event": "PreToolUse",
"tool_name": "Read",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp_db\/tools\/databases_tool.py"
}
}
{
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/mcp-servers\/mcp_db\/tools\/schema_tool.py",
"content": "\"\"\"Schema Tool - Tabellen-Struktur abfragen\"\"\"\nimport sys\n\nsys.path.insert(0, \"\/var\/www\/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_schema_tool(mcp):\n \"\"\"Registriert db_schema Tool\"\"\"\n\n logger = ProtokollLogger()\n\n @mcp.tool()\n def db_schema(database: str = \"ki_protokoll\") -> dict:\n \"\"\"\n Zeigt alle Tabellen in einer Datenbank.\n\n Args:\n database: Zieldatenbank (ki_protokoll oder ki_system)\n\n Returns:\n Dict mit tables (List[dict])\n \"\"\"\n # Database Allowlist prüfen\n if database not in Config.ALLOWED_DATABASES:\n error = f\"Database '{database}' not allowed\"\n try:\n logger.log(\n LogEntry(\n request=f\"db_schema({database})\",\n status=\"denied\",\n error_message=error,\n )\n )\n except Exception:\n pass\n return {\"error\": error, \"tables\": []}\n\n try:\n with DatabaseConnection.get_connection(database) as conn:\n # buffered=True verhindert \"Unread result found\" Fehler\n cursor = conn.cursor(dictionary=True, buffered=True)\n cursor.execute(\n \"\"\"SELECT TABLE_NAME, TABLE_ROWS, CREATE_TIME\n FROM information_schema.TABLES\n WHERE TABLE_SCHEMA = %s\n AND TABLE_TYPE = 'BASE TABLE'\n ORDER BY TABLE_NAME\"\"\",\n (database,),\n )\n tables = cursor.fetchall()\n cursor.close()\n\n try:\n logger.log(\n LogEntry(\n request=f\"db_schema({database})\",\n status=\"success\",\n )\n )\n except Exception:\n pass\n\n return {\"tables\": tables, \"database\": database}\n\n except Exception as e:\n try:\n logger.log(\n LogEntry(\n request=f\"db_schema({database})\",\n status=\"error\",\n error_message=str(e)[:200],\n )\n )\n except Exception:\n pass\n return {\"error\": str(e), \"tables\": []}\n",
"numLines": 82,
"startLine": 1,
"totalLines": 82
}
}
}