MCP-DB Contracts

Typisierte Datenstrukturen für Request, Response und Logging.

QueryRequest

Immutable Request-Objekt für Datenbankabfragen.

from dataclasses import dataclass
from typing import Optional

@dataclass(frozen=True)
class QueryRequest:
    """Immutable Query Request - SRP: Nur Query-Daten"""
    query: str
    params: Optional[tuple] = None
    database: str = "ki_protokoll"
    max_rows: int = 100

Feld	Typ	Default	Beschreibung
query	str	-	SQL SELECT Statement
params	Optional[tuple]	None	Prepared Statement Parameter
database	str	"ki_protokoll"	Zieldatenbank
max_rows	int	100	Maximale Ergebniszeilen

QueryStatus

Enum für Query-Ergebnisse.

from enum import Enum

class QueryStatus(Enum):
    SUCCESS = "success"
    ERROR = "error"
    DENIED = "denied"

Status	Bedeutung
SUCCESS	Query erfolgreich ausgeführt
ERROR	Technischer Fehler bei Ausführung
DENIED	Query durch Validator abgelehnt

QueryResponse

Strukturierte Antwort auf Datenbankabfragen.

@dataclass
class QueryResponse:
    """Structured Response - SRP: Nur Response-Daten"""
    status: QueryStatus
    data: Optional[List[dict]] = None
    row_count: int = 0
    error: Optional[str] = None
    execution_ms: float = 0.0

Feld	Typ	Beschreibung
status	QueryStatus	Ergebnis-Status
data	Optional[List[dict]]	Ergebniszeilen
row_count	int	Anzahl Zeilen
error	Optional[str]	Fehlermeldung (gekürzt)
execution_ms	float	Ausführungszeit in ms

LogEntry

Struktur für Protokoll-Einträge in mcp_log.

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class LogEntry:
    """Protokoll-Eintrag für ki_protokoll.mcp_log"""
    client_name: str = "mcp-db"
    request: str = ""
    status: str = "success"
    duration_ms: int = 0
    error_message: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)

Feld	Typ	Default	Beschreibung
client_name	str	"mcp-db"	Client-Identifikation
request	str	""	Query-Text (gekürzt auf 200 Zeichen)
status	str	"success"	success, error, denied
duration_ms	int	0	Ausführungsdauer
error_message	Optional[str]	None	Fehlerdetails
timestamp	datetime	now()	Zeitstempel

Beispiel-Nutzung

# Request erstellen
request = QueryRequest(
    query="SELECT * FROM mcp_log WHERE status = %s",
    params=("success",),
    database="ki_protokoll",
    max_rows=50
)

# Response auswerten
if response.status == QueryStatus.SUCCESS:
    for row in response.data:
        print(row)
else:
    print(f"Fehler: {response.error}")