schema_tool.py

• Pfad: /var/www/mcp-servers/mcp-db/tools/schema_tool.py
• Namespace: -
• Zeilen: 90 | Größe: 3,010 Bytes
• Geändert: 2025-12-28 13:13:41 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 100

• Dependencies: 100 (25%)
• LOC: 100 (20%)
• Methods: 100 (20%)
• Secrets: 100 (15%)
• Classes: 100 (10%)
• Magic Numbers: 100 (10%)

Keine Issues gefunden.

Dependencies 5

• use datetime.datetime
• use config.Config
• use infrastructure.db_connection.DatabaseConnection
• use shared.domain.LogEntry
• use shared.infrastructure.get_logger

Funktionen 1

• register_schema_tool() Zeile 11

Code

"""Schema Tool - Tabellen-Struktur abfragen"""

from datetime import datetime

from config import Config
from infrastructure.db_connection import DatabaseConnection
from shared.domain import LogEntry
from shared.infrastructure import get_logger


def register_schema_tool(mcp):
    """Registriert db_schema Tool"""

    logger = get_logger("mcp-db", Config)

    @mcp.tool()
    def db_schema(database: str = "ki_protokoll") -> dict:
        """
        Zeigt alle Tabellen in einer Datenbank.

        Args:
            database: Zieldatenbank (ki_protokoll oder ki_system)

        Returns:
            Dict mit tables (List[dict])
        """
        # 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,
                    )
                )
            except Exception:
                pass
            return {"error": error, "tables": []}

        try:
            with DatabaseConnection.get_connection(database) as conn:
                # buffered=True verhindert "Unread result found" Fehler
                cursor = conn.cursor(dictionary=True, buffered=True)
                cursor.execute(
                    """SELECT TABLE_NAME, TABLE_ROWS, CREATE_TIME
                       FROM information_schema.TABLES
                       WHERE TABLE_SCHEMA = %s
                       AND TABLE_TYPE = 'BASE TABLE'
                       ORDER BY TABLE_NAME""",
                    (database,),
                )
                tables = cursor.fetchall()
                cursor.close()

                try:
                    logger.log(
                        LogEntry(
                            timestamp=datetime.now(),
                            client_name="mcp-db",
                            tool_name="db_schema",
                            request=f"db_schema({database})",
                            status="success",
                        )
                    )
                except Exception:
                    pass

                return {"tables": tables, "database": database}

        except Exception as e:
            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],
                    )
                )
            except Exception:
                pass
            return {"error": str(e), "tables": []}