Fix: Use FastAPI/uvicorn for healthcheck like worker implementation

- Create proper healthcheck server module using FastAPI/uvicorn
- Follow exact pattern from worker healthcheck implementation
- Manager process now handles healthcheck in background mode
- Returns JSON response like worker healthcheck
- Tests updated to expect JSON response

Author: zzstoatzz

src/prefect/cli/server.py

@@ -583,28 +583,14 @@ def run_manager_process(
     # Start healthcheck server if requested
     if with_healthcheck:
         import threading
-        from http.server import BaseHTTPRequestHandler, HTTPServer
-
-        class HealthHandler(BaseHTTPRequestHandler):
-            def do_GET(self):
-                if self.path == "/health":
-                    self.send_response(200)
-                    self.send_header("Content-type", "text/plain")
-                    self.end_headers()
-                    self.wfile.write(b"OK")
-                else:
-                    self.send_response(404)
-                    self.end_headers()
-
-            def log_message(self, format, *args):
-                pass  # Suppress request logging
 
-        def run_healthcheck_server():
-            server = HTTPServer(("0.0.0.0", 8080), HealthHandler)
-            server.serve_forever()
+        from prefect.server.services.healthcheck import build_healthcheck_server
 
+        healthcheck_server = build_healthcheck_server()
         healthcheck_thread = threading.Thread(
-            target=run_healthcheck_server, daemon=True
+            name="healthcheck-server-thread",
+            target=healthcheck_server.run,
+            daemon=True,
         )
         healthcheck_thread.start()
         logger.info("Healthcheck server started on port 8080")
@@ -673,32 +659,21 @@ def start_services(
     if not background:
         app.console.print("\n[blue]Starting services... Press CTRL+C to stop[/]\n")
 
-        # Start healthcheck server if requested
+        # Import at runtime to avoid circular imports
+        import threading
+
+        healthcheck_server = None
         healthcheck_thread = None
+
         if with_healthcheck:
-            import threading
-            from http.server import BaseHTTPRequestHandler, HTTPServer
-
-            class HealthHandler(BaseHTTPRequestHandler):
-                def do_GET(self):
-                    if self.path == "/health":
-                        self.send_response(200)
-                        self.send_header("Content-type", "text/plain")
-                        self.end_headers()
-                        self.wfile.write(b"OK")
-                    else:
-                        self.send_response(404)
-                        self.end_headers()
-
-                def log_message(self, format, *args):
-                    pass  # Suppress request logging
-
-            def run_healthcheck_server():
-                server = HTTPServer(("0.0.0.0", 8080), HealthHandler)
-                server.serve_forever()
+            from prefect.server.services.healthcheck import build_healthcheck_server
 
+            # Create and start the healthcheck server in a separate thread
+            healthcheck_server = build_healthcheck_server()
             healthcheck_thread = threading.Thread(
-                target=run_healthcheck_server, daemon=True
+                name="healthcheck-server-thread",
+                target=healthcheck_server.run,
+                daemon=True,
             )
             healthcheck_thread.start()
             app.console.print("[green]Healthcheck server started on port 8080[/]")
@@ -707,6 +682,13 @@ def run_healthcheck_server():
             asyncio.run(Service.run_services())
         except KeyboardInterrupt:
             pass
+        finally:
+            if healthcheck_server and healthcheck_thread:
+                logger.debug("Stopping healthcheck server...")
+                healthcheck_server.should_exit = True
+                healthcheck_thread.join()
+                logger.debug("Healthcheck server stopped.")
+
         app.console.print("\n[green]All services stopped.[/]")
         return
 

src/prefect/server/services/healthcheck.py

@@ -0,0 +1,39 @@
+"""
+Health check server for Prefect services, following the same pattern as the worker health check.
+"""
+
+import uvicorn
+from fastapi import APIRouter, FastAPI, status
+from fastapi.responses import JSONResponse
+
+
+def build_healthcheck_server(log_level: str = "error") -> uvicorn.Server:
+    """
+    Build a healthcheck FastAPI server for services.
+
+    Unlike the worker healthcheck, this is a simple liveness check that
+    returns 200 OK if the services are running.
+
+    Args:
+        log_level (str): the log level to use for the server
+
+    Returns:
+        uvicorn.Server: the configured server ready to run
+    """
+    app = FastAPI()
+    router = APIRouter()
+
+    def perform_health_check():
+        # Simple liveness check - if we can respond, services are running
+        return JSONResponse(status_code=status.HTTP_200_OK, content={"message": "OK"})
+
+    router.add_api_route("/health", perform_health_check, methods=["GET"])
+    app.include_router(router)
+
+    config = uvicorn.Config(
+        app=app,
+        host="0.0.0.0",
+        port=8080,
+        log_level=log_level,
+    )
+    return uvicorn.Server(config=config)

tests/cli/test_server_services.py

@@ -157,7 +157,7 @@ def test_start_services_with_healthcheck(self, pid_file: Path):
         try:
             response = requests.get("http://localhost:8080/health", timeout=5)
             assert response.status_code == 200
-            assert response.text == "OK"
+            assert response.json() == {"message": "OK"}
         finally:
             # Always clean up
             invoke_and_assert(