Health Module

forge.health — Kubernetes-compatible /health and /ready endpoints with custom check registration.

Overview

The Health module provides standardized health and readiness checking that works with Kubernetes liveness and readiness probes out of the box. Every registered module automatically contributes its health_check() method. Custom checks can be added with a simple decorator.

Installation

pip install forge-kaif

Quick Start

from forge.health import HealthModule, health_router

# Mount in your FastAPI app
app.mount("/health", health_router)

Built-in behavior:

Endpoint	Purpose	Failure Behavior
/health (liveness)	Is the process alive?	Kubernetes restarts the pod
/ready (readiness)	Can the process serve traffic?	Kubernetes stops sending traffic

Key Features

Custom Health Checks

from forge.health import check, HealthResult

@check("database")
async def check_database():
    if await db.is_connected():
        return HealthResult.ok()
    return HealthResult.error("Database connection lost")

@check("cache")
async def check_cache():
    if cache.latency_ms > 100:
        return HealthResult.degraded("Cache latency high")
    return HealthResult.ok()

Module-Level Health

Every ForgeModule can implement a health_check() method that is automatically registered:

class MyModule(ForgeModule):
    name = "my_module"

    async def health_check(self) -> HealthResult:
        if self.is_healthy():
            return HealthResult.ok()
        return HealthResult.error("Module unhealthy")

Check Timeout

Each health check runs with a configurable timeout to prevent hanging probes:

config = HealthConfig(check_timeout_seconds=5)

API Reference

Health check module — standardized /health and /ready endpoints.

Provides a health check registry, built-in checks for module dependencies, and FastAPI routers for /health (liveness) and /ready (readiness) endpoints compatible with Kubernetes probes.

Classes

HealthModule

Manages health and readiness checks for the forge runtime.

Integrates with K8s probes and coordinates module-level checks.

Attributes

check_timeout property

check_timeout: float

Get the health check execution timeout.

health_path property

health_path: str

Get the configured liveness check path.

include_details property

include_details: bool

Dynamically determine if check details should be included in responses.

ready_path property

ready_path: str

Get the configured readiness check path.

registry property

registry: HealthRegistry

Get the health registry instance.

Methods:

check

check(name: str, critical: bool = False) -> Callable[[HealthCheckFn], HealthCheckFn]

Decorator to register a custom health check function.

check_all async

check_all() -> dict[str, dict[str, Any]]

Run all registered checks and return results.

Returns dict of check_name -> {status, message, latency_ms}

is_ready

is_ready(check_results: dict[str, dict[str, Any]]) -> bool

Determine overall readiness.

Returns False if any critical check fails (status is error).

register

register(name: str, check: HealthCheckFn, critical: bool = False) -> None

Register a custom health check.

setup async

setup(runtime: ForgeRuntime) -> None

Initialize the health module and configure settings.

teardown async

teardown() -> None

Teardown the health module.

HealthRegistry

Registry managing custom registered health check functions.

Methods:

register

register(name: str, check_fn: HealthCheckFn, critical: bool = False) -> None

Register a health check function.

run_all async

run_all(timeout: float = 5.0) -> dict[str, HealthResult]

Run all registered checks concurrently.

run_critical async

run_critical(timeout: float = 5.0) -> dict[str, HealthResult]

Run only critical registered checks concurrently.

unregister

unregister(name: str) -> None

Unregister a health check function by name.

HealthResult dataclass

Result of a single health check.

Methods:

degraded classmethod

degraded(message: str) -> HealthResult

Create a degraded health check result.

error classmethod

error(message: str) -> HealthResult

Create a failing health check result.

ok classmethod

ok(message: str | None = None) -> HealthResult

Create a successful health check result.

Functions:

check

check(name: str, critical: bool = False) -> Callable[[HealthCheckFn], HealthCheckFn]

Decorator to register a custom health check.