MCP-Contracts Architektur

Verzeichnisstruktur und Komponenten des MCP-Contracts Servers.

Verzeichnisstruktur

/opt/mcp-servers/mcp-contracts/
├── server.py              # Entry Point (FastMCP)
├── config.py              # Konfiguration
├── .env                   # Credentials (DB_HOST, DB_USER, etc.)
├── requirements.txt       # Dependencies
├── venv/                  # Python Virtual Environment
│
├── domain/
│   ├── __init__.py
│   └── contract.py        # Entities (Contract, ContractHistory, Validation)
│
├── infrastructure/
│   ├── __init__.py
│   ├── db_connection.py   # Datenbankverbindung (ki_protokoll)
│   ├── protokoll_logger.py # Logging nach mcp_log
│   └── contract_repository.py # CRUD-Operationen
│
├── validators/
│   ├── __init__.py
│   └── contract_validator.py # YAML-Validierung, Scope-Prüfung
│
└── tools/
    ├── __init__.py
    └── contract_tools.py  # 9 MCP-Tools

Layer-Architektur

Komponenten

server.py

Entry Point mit FastMCP-Initialisierung:

from mcp.server.fastmcp import FastMCP
from tools.contract_tools import register_contract_tools

mcp = FastMCP("mcp-contracts", instructions="...")
register_contract_tools(mcp)

if __name__ == "__main__":
    mcp.run(transport="stdio")

domain/contract.py

Dataclasses für Contract-Entities:

• Contract - id, name, yaml_content, version, status, scope_description
• ContractHistory - contract_id, old_version, new_version, change_description
• ContractValidation - contract_id, passed, violations, triggered_by
• ContractStatus - Enum (draft, active, deprecated)

infrastructure/contract_repository.py

CRUD-Operationen auf ki_protokoll.contracts:

• find_by_id(), find_by_name()
• find_all() mit Filtern
• create(), update(), deprecate()
• get_history(), get_validations()
• save_validation()
• get_statistics()

validators/contract_validator.py

Validierungslogik:

• validate_yaml_syntax() - YAML-Syntaxprüfung
• validate_scope() - Prüfung gegen definierten Scope
• check_path_existence() - Dateipfade prüfen

Datenbank-Schema

-- ki_protokoll.contracts
CREATE TABLE contracts (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    yaml_content TEXT NOT NULL,
    version VARCHAR(50) DEFAULT '1.0',
    status ENUM('draft','active','deprecated') DEFAULT 'active',
    scope_description TEXT,
    created_by VARCHAR(100),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY (name, version)
);

-- ki_protokoll.contract_history
CREATE TABLE contract_history (
    id INT AUTO_INCREMENT PRIMARY KEY,
    contract_id INT NOT NULL,
    old_version VARCHAR(50),
    new_version VARCHAR(50),
    old_yaml_content TEXT,
    change_description TEXT,
    changed_by VARCHAR(100),
    changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (contract_id) REFERENCES contracts(id)
);

-- ki_protokoll.contract_validations
CREATE TABLE contract_validations (
    id INT AUTO_INCREMENT PRIMARY KEY,
    contract_id INT NOT NULL,
    passed BOOLEAN DEFAULT FALSE,
    violations_count INT DEFAULT 0,
    violations_json TEXT,
    triggered_by VARCHAR(50),
    validated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (contract_id) REFERENCES contracts(id)
);

Datenfluss

Claude Code
    ↓ MCP Tool Call
FastMCP (server.py)
    ↓ contracts_validate()
contract_tools.py
    ↓ Repository + Validator
contract_repository.py + contract_validator.py
    ↓ SQL + Logging
ki_protokoll.contracts + mcp_log