karlsCORE KI-Gemeinschaft KI-Wissen Systemisch Denken Erstgespräch Anmelden

Code-Referenz: Die komplette Dokumentation

Hier findest Du die vollständige Referenz aller Module, Funktionen und Tools. Sozusagen das Nachschlagewerk, wenn Du mal wieder vergessen hast, welche Parameter wohin gehören. Wenn man sich eine ganze Weile mit Coding beschäftigt, sammelt sich die Erkenntnis an: Man kann sich nicht alles merken, und das ist völlig in Ordnung.

Modul-Übersicht

Projektstruktur:

content-metadata-server/
├── server.py           # Hauptserver mit FastMCP
├── requirements.txt    # Dependencies
├── config/            # Konfigurationsdateien
│   ├── credentials.json
│   ├── database.json
│   ├── security.json
│   └── logging.json
├── modules/           # Kern-Module
│   ├── __init__.py
│   ├── database.py    # Datenbank-Operationen
│   ├── security.py    # Sicherheits-Validierung
│   └── logging.py     # Logging-System
├── tests/            # Test-Suite
│   ├── test_database.py
│   ├── test_security.py
│   └── test_integration.py
└── logs/             # Log-Dateien
    ├── operations.log
    ├── security.log
    └── queries.log

server.py - Hauptserver

Initialisierung

mcp = FastMCP("Content Metadata Server")

# Globale Manager
config_dir: str = os.environ.get('DATENBANK_CONFIG_DIR', './config')
log_manager: LogManager = LogManager(config_dir)
security: SecurityValidator = SecurityValidator(config_dir)
db_manager: Optional[DatabaseManager] = None

Helper-Funktionen

def get_db_connection() -> DatabaseManager:
    """
    Lazy-Loading für Datenbankverbindung.

    Returns:
        DatabaseManager: Aktive Datenbank-Instanz

    Raises:
        ConnectionError: Wenn Verbindung fehlschlägt
    """
    global db_manager
    if db_manager is None:
        db_manager = DatabaseManager(config_dir)
        db_manager.connect()
    return db_manager

MCP Tools API

list_tables()

@mcp.tool()
async def list_tables() -> Dict[str, Any]:
    """
    Listet alle verfügbaren Tabellen der Datenbank auf.

    Returns:
        Dict mit:
        - success: bool
        - tables: List[str] (bei Erfolg)
        - count: int (bei Erfolg)
        - error: str (bei Fehler)

    Example Response:
        {
            "success": true,
            "tables": ["content_metadata", "pages", "users"],
            "count": 3
        }
    """

describe_table()

@mcp.tool()
async def describe_table(
    table: str = None
) -> Dict[str, Any]:
    """
    Zeigt die Struktur einer Tabelle.

    Args:
        table: Tabellenname (optional, default: content_metadata)

    Returns:
        Dict mit:
        - success: bool
        - table: str
        - structure: List[Dict] mit Spalten-Infos
        - error: str (bei Fehler)

    Example Response:
        {
            "success": true,
            "table": "content_metadata",
            "structure": [
                {
                    "Field": "id",
                    "Type": "int(11)",
                    "Null": "NO",
                    "Key": "PRI",
                    "Default": null,
                    "Extra": "auto_increment"
                }
            ]
        }
    """

select_data()

@mcp.tool()
async def select_data(
    columns: Optional[List[str]] = None,
    where: Optional[str] = None,
    params: Optional[List] = None,
    order_by: Optional[str] = None,
    limit: Optional[int] = None
) -> Dict[str, Any]:
    """
    Führt eine SELECT-Query auf content_metadata aus.

    Args:
        columns: Liste der Spalten (None = alle)
        where: WHERE-Klausel ohne WHERE keyword
        params: Parameter für prepared statements
        order_by: ORDER BY Klausel
        limit: Max Ergebnisse (max 100)

    Returns:
        Dict mit:
        - success: bool
        - data: List[Dict] (bei Erfolg)
        - count: int (bei Erfolg)
        - error: str (bei Fehler)

    Example:
        result = await select_data(
            columns=["id", "title", "status"],
            where="status = ? AND template = ?",
            params=["published", "100-template"],
            order_by="created_at DESC",
            limit=10
        )
    """

insert_data()

@mcp.tool()
async def insert_data(
    data: Dict[str, Any]
) -> Dict[str, Any]:
    """
    Fügt einen neuen Datensatz ein.

    Args:
        data: Dictionary mit Spalten und Werten
              Pflichtfeld: title
              Defaults: status='draft', template='100-template'

    Returns:
        Dict mit:
        - success: bool
        - insert_id: int (bei Erfolg)
        - message: str (bei Erfolg)
        - error: str (bei Fehler)

    Example:
        result = await insert_data({
            "title": "Neue Seite",
            "meta_description": "SEO Beschreibung",
            "status": "draft",
            "template": "100-template"
        })
    """

update_data()

@mcp.tool()
async def update_data(
    data: Dict[str, Any],
    where: str,
    params: Optional[List] = None
) -> Dict[str, Any]:
    """
    Aktualisiert bestehende Datensätze.

    Args:
        data: Dictionary mit zu ändernden Spalten
        where: WHERE-Klausel ohne WHERE keyword
        params: Parameter für WHERE-Klausel

    Returns:
        Dict mit:
        - success: bool
        - affected_rows: int (bei Erfolg)
        - message: str (bei Erfolg)
        - error: str (bei Fehler)

    Example:
        result = await update_data(
            data={"status": "published"},
            where="id = ?",
            params=[42]
        )
    """

delete_data()

@mcp.tool()
async def delete_data(
    where: str,
    params: Optional[List] = None
) -> Dict[str, Any]:
    """
    Löscht Datensätze (VORSICHT!).

    Args:
        where: WHERE-Klausel ohne WHERE keyword
        params: Parameter für WHERE-Klausel

    Returns:
        Dict mit:
        - success: bool
        - affected_rows: int (bei Erfolg)
        - message: str (bei Erfolg)
        - error: str (bei Fehler)

    Security:
        - Blockiert unsichere WHERE-Klauseln
        - Loggt Mass-Deletes als Security-Event

    Example:
        result = await delete_data(
            where="status = ? AND created_at < ?",
            params=["draft", "2024-01-01"]
        )
    """

count_records()

@mcp.tool()
async def count_records(
    where: Optional[str] = None,
    params: Optional[List] = None
) -> Dict[str, Any]:
    """
    Zählt Datensätze in content_metadata.

    Args:
        where: Optionale WHERE-Klausel
        params: Parameter für WHERE-Klausel

    Returns:
        Dict mit:
        - success: bool
        - count: int (bei Erfolg)
        - error: str (bei Fehler)

    Example:
        result = await count_records(
            where="status = ?",
            params=["published"]
        )
    """

get_log_stats()

@mcp.tool()
async def get_log_stats(
    start_time: Optional[str] = None,
    end_time: Optional[str] = None
) -> Dict[str, Any]:
    """
    Gibt Server-Nutzungsstatistiken zurück.

    Args:
        start_time: Start-Zeitpunkt (ISO format)
        end_time: End-Zeitpunkt (ISO format)

    Returns:
        Dict mit:
        - success: bool
        - stats: Dict mit Statistiken
        - error: str (bei Fehler)

    Example:
        result = await get_log_stats(
            start_time="2024-01-01T00:00:00",
            end_time="2024-01-31T23:59:59"
        )
    """

DatabaseManager Klasse

Konstruktor und Verbindung

class DatabaseManager:
    def __init__(self, config_dir: str):
        """
        Args:
            config_dir: Pfad zum Config-Verzeichnis
        """

    def connect(self, user_type: str = 'content_reader') -> bool:
        """
        Stellt Datenbankverbindung her.

        Args:
            user_type: 'content_reader' oder 'content_writer'

        Returns:
            bool: True bei Erfolg

        Raises:
            ConnectionError: Bei Verbindungsfehler
        """

    def disconnect(self):
        """Schließt Datenbankverbindung."""

CRUD-Operationen

def select(
    self,
    columns: Optional[List[str]] = None,
    where: Optional[str] = None,
    params: Optional[List] = None,
    order_by: Optional[str] = None,
    limit: Optional[int] = None
) -> Tuple[bool, Union[List[Dict], str]]:
    """
    SELECT-Operation.

    Returns:
        Tuple: (success: bool, result: List[Dict] oder error: str)
    """

def insert(
    self,
    data: Dict[str, Any]
) -> Tuple[bool, Union[Dict, str]]:
    """
    INSERT-Operation.

    Returns:
        Tuple: (success: bool, result: Dict oder error: str)
    """

def update(
    self,
    data: Dict[str, Any],
    where: str,
    params: Optional[List] = None
) -> Tuple[bool, Union[Dict, str]]:
    """
    UPDATE-Operation.

    Returns:
        Tuple: (success: bool, result: Dict oder error: str)
    """

def delete(
    self,
    where: str,
    params: Optional[List] = None
) -> Tuple[bool, Union[Dict, str]]:
    """
    DELETE-Operation.

    Returns:
        Tuple: (success: bool, result: Dict oder error: str)
    """

Transaktionen

def begin_transaction(self):
    """Startet eine Transaktion."""

def commit_transaction(self):
    """Committed die aktuelle Transaktion."""

def rollback_transaction(self):
    """Macht die aktuelle Transaktion rückgängig."""

SecurityValidator Klasse

Validierungs-Methoden

class SecurityValidator:
    def validate_query(
        self,
        query: str,
        params: Optional[List] = None
    ) -> Tuple[bool, str]:
        """
        Prüft Query auf SQL-Injection.

        Returns:
            Tuple: (valid: bool, message: str)
        """

    def validate_table_name(
        self,
        table: str
    ) -> bool:
        """
        Validiert Tabellennamen.

        Returns:
            bool: True wenn valid
        """

    def check_rate_limit(
        self,
        user: str
    ) -> Tuple[bool, str]:
        """
        Prüft Rate-Limiting.

        Returns:
            Tuple: (allowed: bool, message: str)
        """

    def check_operation_permission(
        self,
        user_type: str,
        operation: str,
        database: str
    ) -> bool:
        """
        Prüft Berechtigungen.

        Returns:
            bool: True wenn erlaubt
        """

LogManager Klasse

Logging-Methoden

class LogManager:
    def get_logger(
        self,
        name: str
    ) -> logging.Logger:
        """
        Gibt benannten Logger zurück.

        Args:
            name: Logger-Name (z.B. 'database', 'security')

        Returns:
            Logger-Instanz
        """

    def log_operation(
        self,
        operation: str,
        details: Dict[str, Any],
        level: str = 'INFO',
        logger_name: str = 'operations'
    ):
        """Loggt eine Operation strukturiert."""

    def log_query(
        self,
        query_type: str,
        query: str,
        params: Optional[List] = None,
        execution_time: Optional[float] = None,
        affected_rows: Optional[int] = None,
        error: Optional[str] = None
    ):
        """Loggt Datenbank-Queries."""

    def log_security_event(
        self,
        event_type: str,
        details: Dict[str, Any],
        severity: str = 'WARNING'
    ):
        """Loggt Security-Events."""

    def log_performance(
        self,
        operation: str,
        duration_ms: float,
        threshold_ms: float = 1000
    ):
        """Loggt Performance-Metriken."""

Konfigurationsdateien

credentials.json

{
  "database_users": {
    "content_reader": {
      "username": "content_reader",
      "password": "SECURE_PASSWORD_HERE"
    },
    "content_writer": {
      "username": "content_writer",
      "password": "SECURE_PASSWORD_HERE"
    }
  }
}

database.json

{
  "connection": {
    "host": "localhost",
    "port": 3306,
    "database": "karlkratz_de",
    "charset": "utf8mb4",
    "autocommit": false,
    "connect_timeout": 10,
    "read_timeout": 30,
    "write_timeout": 30
  },
  "limits": {
    "max_select_limit": 100,
    "max_batch_size": 50,
    "max_query_length": 10000
  }
}

security.json

{
  "rate_limiting": {
    "enabled": true,
    "queries_per_minute": 50,
    "burst_size": 20
  },
  "sql_injection_protection": {
    "enabled": true,
    "blocked_keywords": [
      "DROP", "TRUNCATE", "DELETE FROM",
      "UPDATE", "INSERT INTO", "ALTER",
      "CREATE", "REPLACE", "GRANT", "REVOKE"
    ]
  },
  "permissions": {
    "content_reader": {
      "allowed_operations": ["SELECT", "DESCRIBE"],
      "allowed_databases": ["karlkratz_de"]
    },
    "content_writer": {
      "allowed_operations": ["SELECT", "INSERT", "UPDATE", "DELETE"],
      "allowed_databases": ["karlkratz_de"]
    }
  }
}

logging.json

{
  "level": "INFO",
  "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
  "handlers": {
    "file": {
      "enabled": true,
      "path": "logs/operations.log",
      "max_size_mb": 10,
      "backup_count": 5
    }
  },
  "logger_levels": {
    "database": "DEBUG",
    "security": "INFO",
    "performance": "WARNING"
  },
  "log_queries": {
    "enabled": true,
    "log_parameters": true,
    "log_execution_time": true,
    "log_affected_rows": true
  }
}

Fehler-Codes

Standard-Fehlercodes:

# Database Errors
DB001: Connection failed
DB002: Query execution failed
DB003: Transaction rollback
DB004: Table not found
DB005: Column not found

# Security Errors
SEC001: SQL injection detected
SEC002: Rate limit exceeded
SEC003: Permission denied
SEC004: Invalid table name
SEC005: Unsafe WHERE clause

# Validation Errors
VAL001: Missing required field
VAL002: Invalid data type
VAL003: Value out of range
VAL004: Pattern mismatch
VAL005: Constraint violation

# System Errors
SYS001: Config file not found
SYS002: Logger initialization failed
SYS003: Module import error
SYS004: Environment variable missing
SYS005: File permission error

Performance-Metriken

# Erwartete Performance-Werte
SELECT (einfach):     < 50ms
SELECT (komplex):     < 200ms
INSERT:               < 100ms
UPDATE:               < 100ms
DELETE:               < 100ms
COUNT:                < 30ms

# Rate Limits
Standard User:        50 queries/minute
Privileged User:      200 queries/minute
Burst erlaubt:        20 queries

# Connection Pool
Min Connections:      2
Max Connections:      10
Idle Timeout:         3600s
Connection Timeout:   10s

Testing-Befehle

# Unit Tests ausführen
python -m pytest tests/unit/ -v

# Integration Tests
python -m pytest tests/integration/ -v

# Coverage Report
coverage run -m pytest tests/
coverage report -m
coverage html

# Performance Tests
python tests/test_performance.py

# Security Tests
python tests/test_security.py

# Einzelnen Test ausführen
python -m pytest tests/test_database.py::TestDatabaseOperations::test_insert -v

Deployment-Befehle

# Development starten
python server.py

# Production mit Systemd
sudo systemctl start mcp-content-metadata
sudo systemctl status mcp-content-metadata
sudo systemctl restart mcp-content-metadata
sudo systemctl stop mcp-content-metadata

# Logs anzeigen
journalctl -u mcp-content-metadata -f
tail -f logs/operations.log

# Health Check
curl http://localhost:8080/health

# Metriken abrufen
curl http://localhost:9090/metrics

Debugging-Befehle

# Python Debugger starten
python -m pdb server.py

# Memory Profiling
python -m memory_profiler server.py

# CPU Profiling
python -m cProfile -o profile.stats server.py
python -m pstats profile.stats

# Trace Ausführung
python -m trace --trace server.py > trace.log

# Strace für System Calls
strace -o strace.log python server.py

# MySQL Prozesse anzeigen
mysql -e "SHOW PROCESSLIST;" | grep content_

# Offene Connections zählen
netstat -an | grep 3306 | wc -l

Umgebungsvariablen

# Pflicht
DATENBANK_CONFIG_DIR=/pfad/zu/config

# Optional
MCP_ENV=production|staging|development
LOG_LEVEL=DEBUG|INFO|WARNING|ERROR|CRITICAL
DB_HOST=localhost
DB_PORT=3306
PYTHONPATH=/pfad/zu/server

Quick Reference Card

Die wichtigsten Befehle auf einen Blick:

# Server testen
cd /var/www/mcp/content-metadata-server
source venv/bin/activate
python test_server.py

# Claude Config prüfen
cat ~/.claude/mcp.json

# Logs checken
tail -f logs/operations.log | grep ERROR

# Database testen
mysql -u content_reader -p -e "SELECT COUNT(*) FROM karlkratz_de.content_metadata;"

# Permissions fixen
chmod 755 server.py
chmod 600 config/*.json
chown -R www-data:www-data .

# Service neu starten
sudo systemctl restart mcp-content-metadata

Was ich gelernt habe

Nach Jahren der API-Entwicklung:

Diese Referenz ist Dein Spickzettel. Bookmark sie, drucke sie aus, häng sie an die Wand. Du wirst sie brauchen.

Das war die vollständige technische Referenz. Wenn Du nochmal einen Überblick über das gesamte MCP Server Projekt brauchst oder von vorne beginnen möchtest, findest Du hier die Übersichtsseite mit allen Kapiteln zur MCP Server Entwicklung.