Health Checks¶

from kei_agent import get_health_manager, APIHealthCheck, MemoryHealthCheck

# Health Manager konfigurieren
health = get_health_manager()

# Checks registrieren
health.register_check(APIHealthCheck(
    name="kei_api",
    url="https://api.kei-framework.com/health"
))

health.register_check(MemoryHealthCheck(
    name="system_memory",
    warning_threshold=0.8,
    critical_threshold=0.95
))

# Health Check ausführen
summary = await health.run_all_checks()
print(f"Gesamtstatus: {summary.overall_status}")

for check in summary.checks:
    print(f"  {check.name}: {check.status}")

from kei_agent import BaseHealthCheck, HealthStatus

class DatabaseHealthCheck(BaseHealthCheck):
    def __init__(self, connection_string: str):
        super().__init__("database")
        self.connection_string = connection_string

    async def check(self) -> HealthStatus:
        try:
            # Database-Verbindung testen
            # ... connection logic ...
            return HealthStatus.HEALTHY
        except Exception:
            return HealthStatus.UNHEALTHY

# Registrieren
health.register_check(DatabaseHealthCheck("postgresql://..."))