MCP-DB Contracts

Typisierte Datenstrukturen für Request und Response.

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_dev"
    max_rows: int = 100

Feld	Typ	Default	Beschreibung
query	str	-	SQL SELECT Statement
params	Optional[tuple]	None	Prepared Statement Parameter
database	str	"ki_dev"	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

ExecuteContract

Contract für DDL-Operationen (db_execute Tool).

@dataclass(frozen=True)
class ExecuteRequest:
    """Request für DDL-Statements"""
    statement: str
    database: str = "ki_dev"
    params: Optional[tuple] = None

@dataclass
class ExecuteResponse:
    """Response für DDL-Statements"""
    status: str  # success, error, denied
    affected_rows: int = 0
    error: Optional[str] = None
    execution_ms: float = 0.0

Beispiel-Nutzung

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

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

Logging

Alle Operationen werden direkt in ki_dev.mcp_log protokolliert. Die Logging-Struktur entspricht der mcp_log Tabelle:

Feld	Typ	Beschreibung
client_name	VARCHAR(50)	Client-Identifikation (mcp-db)
request	TEXT	Query-Text (gekürzt)
status	ENUM	success, error, denied
duration_ms	INT	Ausführungsdauer
error_message	TEXT	Fehlerdetails (optional)