create endpoint for cloud health check

cognee/api/client.py

@@ -215,6 +215,19 @@ async def detailed_health_check():
         )
 
 
+@app.get("/health/cloud")
+async def cloud_health_check():
+    """
+    Health check endpoint for cloud connection.
+    """
+    try:
+        health_status = await health_checker.check_cloud_connection()
+        status_code = 503 if health_status.status == HealthStatus.UNHEALTHY else 200
+
+        return JSONResponse(status_code=status_code, content=health_status.model_dump())
+    except Exception as e:
+        return JSONResponse(status_code=503, content={"status": "unhealthy", "error": f"Cloud health check failed: {str(e)}"})
+
 app.include_router(get_auth_router(), prefix="/api/v1/auth", tags=["auth"])
 
 app.include_router(

cognee/api/health.py

@@ -324,6 +324,30 @@ class HealthChecker:
             components=components,
         )
 
+    async def check_cloud_connection(self) -> ComponentHealth:
+        """Check cloud connection health."""
+        start_time = time.time()
+        try:
+            from cognee.modules.cloud.operations import check_api_key
+            # TODO: consider moving this to a more appropriate place
+            from cognee.api.v1.sync.sync import _get_cloud_auth_token
+            await check_api_key(_get_cloud_auth_token())
+            response_time = int((time.time() - start_time) * 1000)
+            return ComponentHealth(
+                status=HealthStatus.HEALTHY,
+                provider="cloud",
+                response_time_ms=response_time,
+                details="Connection successful",
+            )
+        except Exception as e:
+            response_time = int((time.time() - start_time) * 1000)
+            logger.error(f"Cloud connection health check failed: {str(e)}")
+            return ComponentHealth(
+                status=HealthStatus.UNHEALTHY,
+                provider="cloud",
+                response_time_ms=response_time,
+                details=f"Connection failed: {str(e)}",
+            )
 
 # Global health checker instance
 health_checker = HealthChecker()

cognee/api/v1/sync/sync.py

@@ -249,7 +249,7 @@ async def _sync_to_cognee_cloud(
     try:
         # Get cloud configuration
         cloud_base_url = await _get_cloud_base_url()
-        cloud_auth_token = await _get_cloud_auth_token(user)
+        cloud_auth_token = _get_cloud_auth_token()
 
         # Step 1: Sync files for all datasets concurrently
         sync_files_tasks = [
@@ -559,7 +559,7 @@ async def _get_cloud_base_url() -> str:
     return os.getenv("COGNEE_CLOUD_API_URL", "http://localhost:8001")
 
 
-async def _get_cloud_auth_token(user: User) -> str:
+def _get_cloud_auth_token() -> str:
     """Get authentication token for Cognee Cloud API."""
     return os.getenv("COGNEE_CLOUD_AUTH_TOKEN", "your-auth-token")