health endpoint

Author: Danelegend

code-interpreter/app/main.py

@@ -11,6 +11,8 @@
 
 from app.api.routes import router as api_router
 from app.app_configs import EXECUTOR_BACKEND, HOST, PORT, PYTHON_EXECUTOR_DOCKER_IMAGE
+from app.models.schemas import HealthResponse
+from app.services.executor_factory import get_executor
 
 # Configure logging
 logging.basicConfig(
@@ -101,8 +103,11 @@ def create_app() -> FastAPI:
     )
 
     @app.get("/health")
-    def health() -> dict[str, str]:  # sync + strictly typed
-        return {"status": "ok"}
+    def health() -> HealthResponse:
+        """Health check that verifies the executor backend is operational."""
+        result = get_executor().check_health()
+        return HealthResponse(status=result.status, message=result.message)
+
 
     app.include_router(api_router, prefix="/v1")
     return app

code-interpreter/app/models/schemas.py

@@ -115,3 +115,11 @@ class ListFilesResponse(BaseModel):
         default_factory=list,
         description="List of all stored files with their metadata.",
     )
+
+
+# ── Health check models ──────────────────────────────────────────────
+
+
+class HealthResponse(BaseModel):
+    status: Literal["ok", "error"]
+    message: StrictStr | None = None

code-interpreter/app/services/executor_base.py

@@ -98,6 +98,14 @@ class StreamResult:
 StreamEvent = StreamChunk | StreamResult
 
 
+@dataclass(frozen=True, slots=True)
+class HealthCheck:
+    """Result of an executor health check."""
+
+    status: Literal["ok", "error"]
+    message: str | None = None
+
+
 class ExecutorProtocol(Protocol):
     def execute_python(
         self,
@@ -114,6 +122,13 @@ def execute_python(
 
 
 class BaseExecutor(ABC):
+    def check_health(self) -> HealthCheck:
+        """Check if the executor backend is operational.
+
+        Default implementation returns ok. Override for backend-specific checks.
+        """
+        return HealthCheck(status="ok")
+
     @abstractmethod
     def execute_python(
         self,

code-interpreter/app/services/executor_docker.py

@@ -24,6 +24,7 @@
     BaseExecutor,
     EntryKind,
     ExecutionResult,
+    HealthCheck,
     StreamChunk,
     StreamEvent,
     StreamResult,
@@ -49,6 +50,51 @@ def __init__(self) -> None:
         self.image = PYTHON_EXECUTOR_DOCKER_IMAGE
         self.run_args = PYTHON_EXECUTOR_DOCKER_RUN_ARGS
 
+    def check_health(self) -> HealthCheck:
+        """Verify Docker daemon is reachable and the executor image is available."""
+        # Check Docker daemon connectivity
+        try:
+            result = subprocess.run(
+                [self.docker_binary, "version", "--format", "{{.Server.Version}}"],
+                capture_output=True,
+                timeout=5,
+                check=False,
+            )
+        except FileNotFoundError:
+            return HealthCheck(status="error", message="Docker binary not found")
+        except subprocess.TimeoutExpired:
+            return HealthCheck(status="error", message="Docker daemon not responding")
+
+        if result.returncode != 0:
+            stderr = result.stderr.decode("utf-8", errors="replace").strip()
+            return HealthCheck(
+                status="error",
+                message=f"Docker daemon not reachable: {stderr}",
+            )
+
+        # Check executor image is available locally
+        image_with_tag = f"{self.image}:latest"
+        try:
+            img_result = subprocess.run(
+                [self.docker_binary, "image", "inspect", image_with_tag],
+                capture_output=True,
+                timeout=5,
+                check=False,
+            )
+        except subprocess.TimeoutExpired:
+            return HealthCheck(
+                status="error",
+                message=f"Timeout checking image {image_with_tag}",
+            )
+
+        if img_result.returncode != 0:
+            return HealthCheck(
+                status="error",
+                message=f"Executor image {image_with_tag} not available locally",
+            )
+
+        return HealthCheck(status="ok")
+
     def _resolve_docker_binary(self) -> str:
         candidate = PYTHON_EXECUTOR_DOCKER_BIN
         docker_path = which(candidate)

code-interpreter/app/services/executor_kubernetes.py

@@ -31,6 +31,7 @@
     BaseExecutor,
     EntryKind,
     ExecutionResult,
+    HealthCheck,
     WorkspaceEntry,
     wrap_last_line_interactive,
 )
@@ -76,6 +77,22 @@ def __init__(self) -> None:
         self.image = KUBERNETES_EXECUTOR_IMAGE
         self.service_account = KUBERNETES_EXECUTOR_SERVICE_ACCOUNT
 
+    def check_health(self) -> HealthCheck:
+        """Verify Kubernetes API is reachable and the namespace is accessible."""
+        try:
+            self.v1.read_namespace(name=self.namespace)
+        except ApiException as e:
+            return HealthCheck(
+                status="error",
+                message=f"Kubernetes API error (namespace={self.namespace}): {e.reason}",
+            )
+        except Exception as e:
+            return HealthCheck(
+                status="error",
+                message=f"Kubernetes API not reachable: {e}",
+            )
+        return HealthCheck(status="ok")
+
     def _create_pod_manifest(
         self,
         pod_name: str,

code-interpreter/tests/e2e/test_basic_flow.py

@@ -18,7 +18,7 @@ def test_execute_endpoint_basic_flow() -> None:
             pytest.fail(f"Failed to reach Code Interpreter service at {BASE_URL}: {exc!s}")
 
         assert health_response.status_code == 200, health_response.text
-        assert health_response.json() == {"status": "ok"}
+        assert health_response.json()["status"] == "ok"
 
         execute_payload: dict[str, Any] = {
             "code": "print('hello from e2e')",