{
"tool_response": {
"type": "text",
"file": {
"filePath": "\/var\/www\/mcp-servers\/mcp_db\/config.py",
"content": "\"\"\"Zentrale Konfiguration für MCP-DB Server mit Hot-Reload aus JSON\"\"\"\nimport json\nimport os\nfrom pathlib import Path\nfrom typing import List\n\n# Pfad zur JSON-Config\nCONFIG_PATH = Path(\"\/etc\/mcp_db\/config.json\")\n\n\ndef _load_json_config() -> dict:\n \"\"\"Lädt JSON-Config bei jedem Aufruf (Hot-Reload).\"\"\"\n if CONFIG_PATH.exists():\n try:\n with open(CONFIG_PATH, \"r\") as f:\n return json.load(f)\n except (json.JSONDecodeError, IOError):\n pass\n return {}\n\n\nclass _ConfigMeta(type):\n \"\"\"Metaclass für dynamische Property-Zugriffe auf Config.\"\"\"\n\n @property\n def ALLOWED_DATABASES(cls) -> List[str]:\n \"\"\"Erlaubte Datenbanken - aus JSON geladen.\"\"\"\n cfg = _load_json_config()\n return cfg.get(\"allowed_databases\", [\"ki_dev\", \"ki_content\"])\n\n @property\n def ALLOWED_TABLES(cls) -> List[str]:\n \"\"\"Erlaubte Tabellen - flache Liste aus allen DBs.\"\"\"\n cfg = _load_json_config()\n tables_dict = cfg.get(\"allowed_tables\", {})\n # Flatten: alle Tabellen aus allen DBs\n all_tables = []\n for db_tables in tables_dict.values():\n if isinstance(db_tables, list):\n all_tables.extend(db_tables)\n return all_tables if all_tables else [\n \"mcp_log\", \"protokoll\", \"tasks\", \"contracts\",\n \"dokumentation\", \"prompts\", \"critics\"\n ]\n\n @property\n def BLOCKED_KEYWORDS(cls) -> List[str]:\n \"\"\"Blockierte SQL-Keywords.\"\"\"\n cfg = _load_json_config()\n return cfg.get(\"blocked_keywords\", [\n \"DROP\", \"DELETE\", \"INSERT\", \"UPDATE\", \"TRUNCATE\",\n \"ALTER\", \"CREATE\", \"RENAME\", \"GRANT\", \"REVOKE\",\n \"LOAD_FILE\", \"INTO OUTFILE\", \"INTO DUMPFILE\",\n \"BENCHMARK\", \"SLEEP\"\n ])\n\n @property\n def MAX_QUERY_LENGTH(cls) -> int:\n \"\"\"Maximale Query-Länge.\"\"\"\n cfg = _load_json_config()\n limits = cfg.get(\"limits\", {})\n return limits.get(\"max_query_length\", 2000)\n\n @property\n def MAX_ROWS(cls) -> int:\n \"\"\"Maximale Ergebniszeilen.\"\"\"\n cfg = _load_json_config()\n limits = cfg.get(\"limits\", {})\n return limits.get(\"max_rows\", 100)\n\n @property\n def QUERY_TIMEOUT_SEC(cls) -> int:\n \"\"\"Query-Timeout in Sekunden.\"\"\"\n cfg = _load_json_config()\n limits = cfg.get(\"limits\", {})\n return limits.get(\"query_timeout_sec\", 30)\n\n\nclass Config(metaclass=_ConfigMeta):\n \"\"\"\n Konfiguration mit Hot-Reload aus \/etc\/mcp_db\/config.json.\n\n Änderungen an der JSON-Datei werden sofort wirksam,\n ohne den MCP-Server neu starten zu müssen.\n\n Statische Werte aus Environment Variables:\n - DB_HOST, DB_USER, DB_PASSWORD (Query-User)\n - LOG_DB_HOST, LOG_DB_NAME, LOG_DB_USER, LOG_DB_PASSWORD (Logger)\n\n Dynamische Werte aus JSON:\n - ALLOWED_DATABASES, ALLOWED_TABLES, BLOCKED_KEYWORDS\n - MAX_QUERY_LENGTH, MAX_ROWS, QUERY_TIMEOUT_SEC\n \"\"\"\n\n # Statische Werte aus Environment (nur bei Start geladen)\n DB_HOST: str = os.getenv(\"DB_HOST\", \"localhost\")\n DB_USER: str = os.getenv(\"DB_USER\", \"mcp_readonly\")\n DB_PASSWORD: str = os.getenv(\"DB_PASSWORD\", \"\")\n\n LOG_DB_HOST: str = os.getenv(\"LOG_DB_HOST\", \"localhost\")\n LOG_DB_NAME: str = os.getenv(\"LOG_DB_NAME\", \"ki_dev\")\n LOG_DB_USER: str = os.getenv(\"LOG_DB_USER\", \"mcp_logger\")\n LOG_DB_PASSWORD: str = os.getenv(\"LOG_DB_PASSWORD\", \"\")\n",
"numLines": 104,
"startLine": 1,
"totalLines": 104
}
}
}