describe_tool.py

• Pfad: /var/www/mcp-servers/mcp-db/tools/describe_tool.py
• Namespace: -
• Zeilen: 136 | Größe: 4,566 Bytes
• Geändert: 2025-12-28 13:24:35 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 88

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

Keine Issues gefunden.

Dependencies 10

• use re
• use time
• use datetime.datetime
• use shared.constants.LOG_ENTRY_MAX_LENGTH
• use shared.constants.LOG_QUERY_MAX_LENGTH
• use shared.constants.MS_PER_SECOND
• use config.Config
• use infrastructure.db_connection.DatabaseConnection
• use shared.domain.LogEntry
• use shared.infrastructure.get_logger

Funktionen 1

• register_describe_tool() Zeile 15

Code

"""Describe Tool - Zeigt Tabellen-Struktur an."""

import re
import time
from datetime import datetime

from shared.constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, MS_PER_SECOND

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


def register_describe_tool(mcp) -> None:
    """Registriert db_describe Tool."""
    logger = get_logger("mcp-db", Config)

    @mcp.tool()
    def db_describe(
        table: str,
        database: str = "ki_dev",
        show_create: bool = False,
    ) -> dict:
        """
        Zeigt Tabellen-Struktur an.

        Args:
            table: Tabellenname
            database: Zieldatenbank (ki_dev oder ki_content)
            show_create: True = SHOW CREATE TABLE, False = DESCRIBE

        Returns:
            Dict mit columns oder create_statement
        """
        start = time.time()

        # Validierung: Tabellenname (SQL-Injection-Schutz)
        if not re.match(r"^[a-zA-Z0-9_]+$", table):
            return {
                "status": "denied",
                "error": "Invalid table name. Only alphanumeric and underscore allowed.",
            }

        # Validierung: Datenbank
        if database not in Config.ALLOWED_DATABASES:
            return {
                "status": "denied",
                "error": f"Database '{database}' not allowed.",
            }

        try:
            with DatabaseConnection.get_connection(database) as conn:
                cursor = conn.cursor(dictionary=True, buffered=True)

                if show_create:
                    cursor.execute(f"SHOW CREATE TABLE `{table}`")
                    row = cursor.fetchone()
                    cursor.close()

                    duration = int((time.time() - start) * MS_PER_SECOND)

                    # Log
                    try:
                        logger.log(
                            LogEntry(
                                timestamp=datetime.now(),
                                client_name="mcp-db",
                                tool_name="db_describe",
                                request=f"SHOW CREATE TABLE {table}",
                                status="success",
                                duration_ms=duration,
                            )
                        )
                    except Exception:
                        pass

                    return {
                        "status": "success",
                        "table": table,
                        "create_statement": row.get("Create Table", "") if row else "",
                        "execution_ms": duration,
                    }
                else:
                    cursor.execute(f"DESCRIBE `{table}`")
                    columns = cursor.fetchall()
                    cursor.close()

                    duration = int((time.time() - start) * MS_PER_SECOND)

                    # Log
                    try:
                        logger.log(
                            LogEntry(
                                timestamp=datetime.now(),
                                client_name="mcp-db",
                                tool_name="db_describe",
                                request=f"DESCRIBE {table}",
                                status="success",
                                duration_ms=duration,
                            )
                        )
                    except Exception:
                        pass

                    return {
                        "status": "success",
                        "table": table,
                        "columns": columns,
                        "execution_ms": duration,
                    }

        except Exception as e:
            duration = int((time.time() - start) * MS_PER_SECOND)

            try:
                logger.log(
                    LogEntry(
                        timestamp=datetime.now(),
                        client_name="mcp-db",
                        tool_name="db_describe",
                        request=f"DESCRIBE {table}",
                        status="error",
                        duration_ms=duration,
                        error_message=str(e)[:LOG_ENTRY_MAX_LENGTH],
                    )
                )
            except Exception:
                pass

            return {
                "status": "error",
                "error": str(e)[:LOG_QUERY_MAX_LENGTH],
                "execution_ms": duration,
            }