{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp_db\/tools\/schema_tool.py",
"content": "\"\"\"Schema Tool - Tabellen-Struktur abfragen\"\"\"\n\nfrom datetime import datetime\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_schema_tool(mcp):\n \"\"\"Registriert db_schema Tool\"\"\"\n\n logger = get_logger(\"mcp-db\", Config)\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 pruefen\n if database not in Config.ALLOWED_DATABASES:\n error = f\"Database '{database}' not allowed\"\n try:\n logger.log(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_schema\",\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 timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_schema\",\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 timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_schema\",\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"
}
}
{
"tool_response": {
"type": "update",
"filePath": "\/var\/www\/mcp-servers\/mcp_db\/tools\/schema_tool.py",
"content": "\"\"\"Schema Tool - Tabellen-Struktur abfragen\"\"\"\n\nfrom datetime import datetime\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_schema_tool(mcp):\n \"\"\"Registriert db_schema Tool\"\"\"\n\n logger = get_logger(\"mcp-db\", Config)\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 pruefen\n if database not in Config.ALLOWED_DATABASES:\n error = f\"Database '{database}' not allowed\"\n try:\n logger.log(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_schema\",\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 timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_schema\",\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 timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_schema\",\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",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 18,
"newStart": 1,
"newLines": 17,
"lines": [
" \"\"\"Schema Tool - Tabellen-Struktur abfragen\"\"\"",
"-import sys",
" ",
"-sys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_db\")",
"+from datetime import datetime",
" ",
" 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_schema_tool(mcp):",
" \"\"\"Registriert db_schema Tool\"\"\"",
" ",
"- logger = ProtokollLogger()",
"+ logger = get_logger(\"mcp-db\", Config)",
" ",
" @mcp.tool()",
" def db_schema(database: str = \"ki_protokoll\") -> dict:"
]
},
{
"oldStart": 25,
"oldLines": 12,
"newStart": 24,
"newLines": 15,
"lines": [
" Returns:",
" Dict mit tables (List[dict])",
" \"\"\"",
"- # Database Allowlist prüfen",
"+ # Database Allowlist pruefen",
" if database not in Config.ALLOWED_DATABASES:",
" error = f\"Database '{database}' not allowed\"",
" try:",
" logger.log(",
" LogEntry(",
"+ timestamp=datetime.now(),",
"+ client_name=\"mcp-db\",",
"+ tool_name=\"db_schema\",",
" request=f\"db_schema({database})\",",
" status=\"denied\",",
" error_message=error,"
]
},
{
"oldStart": 58,
"oldLines": 6,
"newStart": 60,
"newLines": 9,
"lines": [
" try:",
" logger.log(",
" LogEntry(",
"+ timestamp=datetime.now(),",
"+ client_name=\"mcp-db\",",
"+ tool_name=\"db_schema\",",
" request=f\"db_schema({database})\",",
" status=\"success\",",
" )"
]
},
{
"oldStart": 71,
"oldLines": 6,
"newStart": 76,
"newLines": 9,
"lines": [
" try:",
" logger.log(",
" LogEntry(",
"+ timestamp=datetime.now(),",
"+ client_name=\"mcp-db\",",
"+ tool_name=\"db_schema\",",
" request=f\"db_schema({database})\",",
" status=\"error\",",
" error_message=str(e)[:200],"
]
}
],
"originalFile": "\"\"\"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"
}
}