Health Checks

A health checks manager with concurrent execution. Use it to expose liveness, readiness, and aggregate health endpoints.

HealthChecks: Register check functions with a FastAPI-style decorator, run them concurrently with per-check timeouts and caching.
OpsServer: Serve /livez, /readyz, /healthz and /metrics on a port of their own. The recommended way.
health_router: FastAPI router that serves the same endpoints on the app port.

Quick start

Create a HealthChecks and register checks with the @health.check(name) decorator:

from grelmicro.health import HealthChecks, HealthDetails

# Create the HealthChecks instance and register it with the Grelmicro
# app via `Grelmicro(uses=[health])` so the router can resolve it.
health = HealthChecks()


# Register checks with the @health.check(name) decorator
@health.check("database")
async def check_database() -> HealthDetails | None:
    # Return None on success, raise on failure
    return None


@health.check("redis")
async def check_redis() -> HealthDetails | None:
    # Return a dict to include details
    return {"latency_ms": 1.2}


# Optional dependency: mark non-critical so its failure doesn't
# take the instance out of the load balancer.
@health.check("external-api", critical=False)
async def check_external_api() -> HealthDetails | None:
    return None

Register the instance with a Grelmicro app (see below) so OpsServer or the router can resolve it, or pass it explicitly via health_router(health).

Health Check

A health check is a function returning None (healthy) or a HealthDetails dict (also healthy, with metrics attached). Raising an exception signals failure.

from grelmicro.health import HealthChecks, HealthDetails
from grelmicro.health.errors import HealthError

health = HealthChecks()


# Decorator form: register an async function under a name
@health.check("database")
async def check_database() -> HealthDetails | None:
    # Return None on success (healthy, no details)
    return None


@health.check("redis")
async def check_redis() -> HealthDetails | None:
    # Return a dict to include details (e.g. metrics)
    return {"latency_ms": 1.2, "version": "7.2"}


@health.check("external-api", critical=False)
async def check_external_api() -> HealthDetails | None:
    # Raise HealthError to expose a specific message in the error field.
    # Other exceptions produce an error formatted as "{ExceptionType}: {message}".
    msg = "Connection refused"
    raise HealthError(msg)

Return None: healthy, no details.
Return a HealthDetails dict: healthy, with details. Values can be primitives, datetime, nested dicts, lists, or tuples.
Raise HealthError: unhealthy. The exception message appears in the error field.
Raise any other exception: unhealthy, with an error field formatted as "{ExceptionType}: {message}" (for example "ValueError: bad input"). The traceback is logged server-side to avoid leaking internal information.

HealthDetails is a type alias for dict[str, JSONEncodable]. Both sync and async check functions are supported. Sync functions run in a worker thread via asyncio.to_thread so they never block the event loop.

The check decorator goes on top

@health.check registers the function it is handed and returns that same one. A decorator written below it wraps the module-level name, while the registry keeps holding the original, so it applies to direct calls and never to a probe. Put @health.check on top:

@health.check("database")
@retry(when=TimeoutError, attempts=3)
async def check_db() -> None: ...

Written the other way round, @retry refuses with a TypeError that names the right order.

A check registered as a bound method, with health.add("db", svc.check), guards that object's method and the class function it runs through. Another instance of the same class is left alone. Wrapping a method loses the object it was bound to, so a decorator that copies the name with functools.wraps is not refused, because refusing it would refuse every other instance too. Put the registering decorator on top and the question does not arise.

Grelmicro app integration

Register the HealthChecks with a Grelmicro app to lifecycle it alongside the rest of your modules. Same FastAPI-style explicit registration as Tasks:

fragment

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks

health = HealthChecks()
micro = Grelmicro(uses=[health])

@health.check("redis")
async def redis_alive() -> None:
    ...

async with micro:
    report = await health.run()

Use Grelmicro.use(item) (or uses=) for entry-point components like HealthChecks and Tasks. The caller keeps the reference and uses it directly.

For imperative registration (without a decorator), use health.add(name, func):

from grelmicro.health import HealthChecks, HealthDetails

health = HealthChecks()


async def check_kafka() -> HealthDetails | None:
    return None


health.add("kafka", check_kafka, critical=True, timeout=2.0)

Critical vs Non-Critical

By default, all checks are critical: a failure flips the aggregate to error and returns 503 on /readyz and /healthz. Pass critical=False for optional dependencies:

from grelmicro.health import HealthChecks, HealthDetails

health = HealthChecks()


@health.check("external-api", critical=False)
async def check_external_api() -> HealthDetails | None:
    return None

Scenario	Aggregate status on `/healthz`	`/readyz`	`/healthz`
All critical pass	`ok`	`200`	`200`
Non-critical failed	`ok`	`200`	`200`
Critical failed	`error`	`503`	`503`

Non-critical checks do not pull the instance from the load balancer. They are not run on /readyz (which runs critical only). Their status appears per-check in the /healthz body so operators and dashboards can see degraded dependencies without triggering traffic removal.

Timeout

Checks that exceed their timeout are reported as failed:

from grelmicro.health import HealthChecks, HealthDetails

# Component default: 2s per check (default is 5s)
health = HealthChecks(timeout=2.0)


# Per-check override: tight timeout for a flaky optional dep
@health.check("analytics", critical=False, timeout=0.5)
async def check_analytics() -> HealthDetails | None:
    return None

HealthChecks has a global default timeout (5.0 seconds). Per-check overrides are set on registration:

from grelmicro.health import HealthChecks, HealthDetails

health = HealthChecks()


@health.check("slow-api", critical=False, timeout=0.5)
async def check_slow_api() -> HealthDetails | None:
    return None

A slow non-critical check hits the timeout and is reported with status: "error" in the response body, but the aggregate stays ok and /readyz stays 200.

Timeout detection uses asyncio.timeout. The wrapper distinguishes the configured timeout from a TimeoutError raised inside the check itself (for example a socket timeout).

Caching

HealthChecks caches each check's result for cache_ttl (default 1 second, in whole seconds or as a timedelta) and coalesces concurrent calls via single-flight per check. A given check runs at most once per TTL regardless of how many endpoints or concurrent requests are in flight. This prevents probe traffic from amplifying onto your database.

from grelmicro.health import HealthChecks

# Default: 1-second TTL with single-flight per check
HealthChecks(timeout=5.0, cache_ttl=1)

# Disable caching entirely
HealthChecks(cache_ttl=0)

Returning Details with an Error

Raise HealthError(message, details=...) to attach a diagnostic payload to a failing check. The payload appears under details on the check entry, subject to show_details:

from grelmicro.health import HealthChecks, HealthDetails
from grelmicro.health.errors import HealthError

health = HealthChecks()


@health.check("database")
async def check_database() -> HealthDetails | None:
    # Simulate a failure with a diagnostic payload
    msg = "connection pool exhausted"
    raise HealthError(msg, details={"active": 10, "idle": 0, "max": 10})

Provider Readiness Checks

Every connection provider ships a built-in readiness probe. Redis and Valkey run PING, Postgres and SQLite run SELECT 1. These are cheaper and more honest than a borrowed lock round trip, and you no longer hand-write one check per backend.

Register a provider's check with add_provider. It registers under provider:{short_name}, for example provider:redis:

from grelmicro.health import HealthChecks
from grelmicro.providers.redis import RedisProvider

store = RedisProvider("redis://localhost:6379")
cache = RedisProvider("redis://localhost:6379/1")

health = HealthChecks()

# Register the provider's built-in readiness check as "provider:redis".
health.add_provider(store)

# A degradable dependency: visible in /healthz, never fails /readyz.
health.add_provider(cache, name="cache", critical=False)

The check is critical by default, so an unreachable backend fails /readyz. Pass critical=False for a degradable dependency such as a cache. Pass name= to disambiguate two providers of the same vendor, which registers provider:{name} instead.

A HealthChecks registers a check for every provider on the app at once. Pass provider_checks= to pick the check:

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks
from grelmicro.providers.postgres import PostgresProvider
from grelmicro.providers.redis import RedisProvider

redis = RedisProvider("redis://localhost:6379")
postgres = PostgresProvider("postgresql://localhost:5432/app")

# Registers "provider:redis" and "provider:postgres" critical checks on startup.
micro = Grelmicro(
    uses=[redis, postgres, HealthChecks(provider_checks="require")]
)

On startup this discovers every active provider, including ones a Component borrows without listing, and registers a provider:{short_name} check for each. provider_checks= takes one of three modes:

Mode	What it registers
`"report"` (default)	A non-critical check per provider. It shows in `/healthz` and leaves `/readyz` alone.
`"require"`	A critical check per provider. A down backend fails `/readyz`.
`"off"`	Nothing.

A backend that every replica shares takes every replica out of rotation at once when it fails /readyz, so the default shows it without failing readiness. Pass "require" when the service is useless without the backend.

A provider with no built-in probe is skipped. A provider you already register with add_provider keeps that registration. When two providers share a short_name, the second is skipped with a warning: register it explicitly with add_provider(provider, name=...) to give it a distinct name.

Circuit Breakers

Register a check on a circuit breaker with add_circuitbreaker. It registers under circuitbreaker:{breaker.name}, for example circuitbreaker:inverter:

from grelmicro.health import HealthChecks
from grelmicro.resilience import CircuitBreaker

inverter = CircuitBreaker.consecutive_count("inverter", error_threshold=5)
heater = CircuitBreaker.consecutive_count("heater", error_threshold=5)

health = HealthChecks()

# Registered as "circuitbreaker:inverter", visible in /healthz only.
health.add_circuitbreaker(inverter)

# An open heater breaker fails /readyz.
health.add_circuitbreaker(heater, critical=True)

The check is non-critical by default. An open breaker usually means a shared dependency is down for every replica, so failing readiness would take every replica out of rotation at once. Pass critical=True when the service is useless without that dependency. Pass name= when the breaker name is not a valid check name.

Each run reads the state from the breaker's backend, so a replica that made no call still sees a breaker another replica opened.

State	Check status	Reported
`CLOSED`	`ok`	nothing
`HALF_OPEN`, `FORCED_CLOSED`	`ok`	`{"state": "HALF_OPEN"}` in the details
`OPEN`	`error`	`circuit breaker open, next probe in 12s`, details `{"state": "OPEN", "retry_after": 12}`
`FORCED_OPEN`	`error`	`circuit breaker forced open until reset`, details `{"state": "FORCED_OPEN"}`

Serve the Endpoints

Register an OpsServer next to your app. It serves the probes and /metrics on a port of their own, so they never share the port your users reach:

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks
from grelmicro.http import OpsServer
from grelmicro.metrics import Metrics

micro = Grelmicro(
    uses=[
        OpsServer(),
        HealthChecks(provider_checks="require"),
        Metrics(exporter="prometheus"),
    ]
)

app = FastAPI()
micro.install(app)
# Your routes on the app port (8000 with Uvicorn)
# GET /livez, /readyz, /healthz and /metrics on :8080

It listens on port 8080 by default. Pass port= or set GREL_OPS_PORT to move it. Point the probes and the Prometheus scrape at that port, and keep it off the internet: read Secure the ops endpoints. It works the same next to FastAPI, another ASGI framework, or no web framework at all. Ops Server covers the configuration.

FastAPI Integration

Some platforms probe only the port they route traffic to. There, serve the endpoints from the app itself:

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks, HealthDetails
from grelmicro.integrations.fastapi import health_router

health = HealthChecks()


@health.check("database")
async def check_database() -> HealthDetails | None:
    return None


micro = Grelmicro(uses=[health])

app = FastAPI()
micro.install(app)
app.include_router(health_router())
# Endpoints: GET /livez, GET /readyz, GET /healthz

The app port is the public port

Whatever the router serves, every caller who reaches the app reaches it too. Behind a reverse proxy, that is the internet. /healthz names every check and reports each failing check's error string. Gate it with healthz_dependencies and keep show_details off, as shown in Verbose Details.

This creates three endpoints:

Endpoint	Purpose	Success	Failure	Body
`GET /livez`	Liveness probe. Runs no check per request, and reflects the liveness checks when set.	`200`	`503`, or no response (timeout)	empty
`GET /readyz`	Readiness probe. Runs critical checks only.	`200`	`503`	empty
`GET /healthz`	Aggregate JSON report for humans and dashboards. Runs all checks.	`200`	`503`	JSON `{status, checks}`

/livez never checks a dependency. Kubernetes restarts the container whose liveness probe fails, so a check on a database there would restart every replica when the database goes down. Catch a stuck worker shows what /livez catches and how to catch the rest.

All three also accept HEAD. All responses set Cache-Control: no-store. Probe endpoints return an empty body. The HTTP status code is the entire signal.

Paths follow the z-pages convention (/livez, /readyz, /healthz). The trailing z avoids collisions with application routes like /health.

micro.install(app) mounts no route. Register an OpsServer, or include the router yourself. A registered HealthChecks that nothing serves is reported at startup with the health-not-served warning.

OpenAPI Schema

The three endpoints stay out of the OpenAPI schema. They answer exactly the same, and /openapi.json and the docs pages name neither the probes nor the shape of the report. An orchestrator does not read your schema, and a generated client has no use for a probe.

Pass include_in_schema=True to publish them, with the /healthz response model and its 503 response. /livez and /readyz are documented as returning no body, because that is what they return:

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks
from grelmicro.integrations.fastapi import health_router

health = HealthChecks()
micro = Grelmicro(uses=[health])

app = FastAPI()
micro.install(app)
app.include_router(health_router(include_in_schema=True))
# GET /livez, GET /readyz, GET /healthz, and all three in /openapi.json

This decides what the schema publishes, not what is reachable. To keep /healthz from answering at all, gate it with healthz_dependencies. To close a probe off, leave it off your ingress.

Report Body

/healthz reports the aggregate status and one entry per check. A check that passed carries status and critical alone:

{
  "status": "error",
  "checks": {
    "store": {"status": "ok", "critical": true},
    "analytics": {"status": "ok", "critical": false},
    "database": {"status": "error", "critical": true, "error": "connection refused"}
  }
}

A field that has nothing to report is left out rather than sent as null. error appears only on a failing check, and details only when the check returned some and show_details allows them. Read an absent error as "this check passed".

Using with Docker, Compose, and other Orchestrators

Different orchestrators consume different probes. grelmicro exposes all three endpoints, pick the ones that fit:

Orchestrator	Uses
Kubernetes, OpenShift	`livenessProbe` → `/livez`, `readinessProbe` → `/readyz`, `startupProbe` → `/livez`
Docker (`HEALTHCHECK`), Docker Compose, Docker Swarm	single healthcheck → `/livez` (restart on failure)
Nomad (`check` stanza)	`/livez` for liveness, `/readyz` for routing
systemd (`WatchdogSec`)	`/livez` (restart on failure)
Reverse proxies (Traefik, nginx, HAProxy, Envoy), load balancers (AWS ALB, GCP LB)	`/readyz` for upstream health
Dashboards, uptime monitors (Prometheus, Pingdom)	`/healthz` for full report

Docker Compose example, with the probes on the OpsServer port:

services:
  app:
    image: myapp
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/livez"]
      interval: 10s
      timeout: 2s
      retries: 3

Docker and Docker Swarm only model a single healthcheck per container, so point it at /livez. Route readiness through a reverse proxy that probes /readyz.

Exclude Checks

Both /readyz and /healthz accept an ?exclude query parameter with a comma-separated list of check names. Useful for temporarily muting a known-flaky check without redeploy:

GET /readyz?exclude=analytics,recommendations
GET /healthz?exclude=analytics

Excluded checks are not run and do not appear in the response.

Verbose Details

Each check can return verbose metadata (versions, internal hostnames, pool stats, latencies). Exposing it publicly on /healthz leaks infrastructure fingerprints to anyone who can reach the endpoint, so by default grelmicro strips every details field from the response.

The gates below apply to health_router() on the app port. OpsServer takes show_details=True or False and nothing else, because the network in front of it is what keeps it private.

The show_details parameter controls visibility with three forms:

`show_details`	Who sees details	Use when
`False` (default)	nobody	Safe default for a public `/healthz`
`True`	everyone	`/healthz` is on a private network only
`Depends(fn)`	requests for which `fn()` returns `True`	Public `/healthz`, admin-only details

With show_details=Depends(fn), fn is wired into FastAPI's dependency-injection graph. Return True to include details for that request, False to strip them. Everything FastAPI supports works: Depends sub-dependencies, Security, Request injection, async functions, yield cleanup. Returning False strips details but the endpoint still returns 200/503 with status and check names. This way, uptime monitors without credentials still get actionable aggregate status while admin tools with credentials get the full payload. Raising HTTPException blocks the endpoint, so prefer returning False for a soft strip.

from fastapi import Depends, Request

from grelmicro.integrations.fastapi import health_router
from grelmicro.security import TrustedProxies, resolve_client_address

# Your own proxies. Required: without it a caller's own header is believed.
trusted = TrustedProxies(["10.0.0.0/8"])


def from_private_network(request: Request) -> bool:
    # Resolve rather than reading `request.client.host`. Behind a proxy the
    # raw peer is the proxy's own private address, so checking it directly
    # reports every caller as private and shows details to the public.
    client = resolve_client_address(request.scope, trusted)
    # `forwarded` means a trusted proxy vouched for this address. Without
    # it nobody did, and behind a proxy the address is the proxy's own,
    # which reads as private and shows details to everyone.
    if client is None or not client.forwarded:
        return False
    return client.ip.is_private


# Hide details from everyone (default).
router = health_router()

# Show details to everyone (private /healthz only).
router = health_router(show_details=True)

# Show details when the dependency returns True.
router = health_router(show_details=Depends(from_private_network))

With details enabled, a check that returned some includes a details field:

{
  "status": "ok",
  "checks": {
    "redis": {
      "status": "ok",
      "critical": true,
      "details": {"latency_ms": 1.2, "version": "7.2"}
    }
  }
}

The error field is always returned

show_details strips only the details field. Each failing check still reports its error string, even on a public /healthz. Keep check error messages safe for the endpoint audience: raise HealthError with a short, non-sensitive message and put diagnostics in details, which stays hidden by default. To hide error strings too, gate the whole endpoint with healthz_dependencies.

`show_details` vs `healthz_dependencies`

Two independent gates sit in front of /healthz:

Parameter	Failure effect	Typical use
`healthz_dependencies`	Blocks the entire endpoint (`401`/`403`)	Keep `/healthz` entirely private
`show_details=Depends(fn)`	When `fn()` returns `False`, strips the `details` field. Endpoint still returns `200`/`503`	Keep aggregate status public, hide verbose metadata

For stricter setups, gate the whole /healthz endpoint behind authentication while leaving /livez and /readyz open (most orchestrators and load balancers cannot carry credentials):

from fastapi import Depends, Request

from grelmicro.integrations.fastapi import health_router


def require_admin(request: Request) -> None:
    # Replace with your real auth. Raise HTTPException on failure.
    return None


# Gate /healthz behind auth. /livez and /readyz remain open for
# orchestrators and load balancers.
router = health_router(
    show_details=True,
    healthz_dependencies=[Depends(require_admin)],
)

URL Prefix

Mount the health endpoints under a custom prefix:

from fastapi import FastAPI

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks
from grelmicro.integrations.fastapi import health_router

health = HealthChecks()
micro = Grelmicro(uses=[health])

app = FastAPI()
micro.install(app)
app.include_router(health_router(prefix="/api/v1"))
# Endpoints: GET /api/v1/livez, GET /api/v1/readyz, GET /api/v1/healthz

Catch a stuck worker

A worker is stuck when its event loop is blocked by a long synchronous call, or when coroutines or threads deadlock while the loop still runs. With one worker per container, a blocked loop stops answering /livez, so the probe times out and the container restarts. A deadlock that leaves the loop free still answers 200, and with several workers per container an idle worker answers the probe for a stuck one.

Set Liveness to catch both, in every setup:

import time

from grelmicro.health import HealthChecks, HealthDetails, HealthError, Liveness

MAX_IDLE = 60
"""Seconds the consumer may go without handling a message."""

health = HealthChecks(liveness=Liveness(stall_timeout=30, failure_threshold=3))
last_handled = time.monotonic()


@health.check("consumer-progress", liveness=True)
async def check_consumer_progress() -> HealthDetails | None:
    if time.monotonic() - last_handled > MAX_IDLE:
        msg = f"no message handled in {MAX_IDLE} s"
        raise HealthError(msg)
    return None

The loop watchdog. With stall_timeout, a thread checks that the loop still runs callbacks, four times per stall_timeout and at most once a second. Past stall_timeout, it logs where the loop thread is stuck and exits the process. It pauses while the app opens or closes.
Liveness checks. A check with liveness=True runs every interval seconds (10 by default), never on /readyz or /healthz. /livez answers 503 from the first failed round. After failure_threshold failed rounds in a row, the worker stops itself with SIGTERM. If it has not exited shutdown_timeout seconds later (30 by default), it exits at once. A check with critical=False is logged when it fails, like a readiness check, and never counts as a failed round.

Either way the worker exits, and whatever runs it starts another: Kubernetes restarts the container, Gunicorn replaces the worker, and so does the uvicorn --workers supervisor.

Look inside the process only

A liveness check that reads a database or another service restarts every replica when that service goes down. Check what only this worker knows, such as whether its consumer made progress.

Without a Web Framework

A worker, a consumer, or a scheduler serves no HTTP, and Kubernetes still probes it. OpsServer gives it a port of its own, with no framework and no new dependency:

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks
from grelmicro.http import OpsServer

micro = Grelmicro(
    uses=[HealthChecks(provider_checks="require"), OpsServer(port=8080)]
)

GET /livez, /readyz and /healthz answer on 8080, exactly as the FastAPI router answers them. Read Ops Server.

On another ASGI framework, mount the endpoints in the app you already serve:

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Mount, Route

from grelmicro import Grelmicro
from grelmicro.health import HealthChecks, HealthDetails, health_asgi

health = HealthChecks()
micro = Grelmicro(uses=[health])


@health.check("database")
async def check_database() -> HealthDetails | None:
    return None


async def home(request: object) -> PlainTextResponse:
    return PlainTextResponse("hello")


app = Starlette(
    routes=[
        Route("/", home),
        # Last: mounted at "" it matches every path, so a route after it
        # would never be reached.
        Mount("", app=health_asgi()),
    ]
)
micro.install(app)
# Endpoints: GET /livez, GET /readyz, GET /healthz

health_asgi() is a pure-ASGI app. It mounts in Starlette, Litestar, or anything else that speaks ASGI, and it takes the same prefix and show_details the router takes. Mounted at "" it matches every path, so it goes after the routes of your own. It has no Depends, so gate it with your own middleware where the router would take healthz_dependencies.

You run	Recommended	On the app port instead
FastAPI	`OpsServer()`	`health_router()`, with the OpenAPI schema and `Depends` gates
Another ASGI framework	`OpsServer()`	`health_asgi()`, gated by your own middleware
No web framework	`OpsServer()`	None, there is no app port

Design

Why Three Endpoints

Each endpoint serves a different audience:

Endpoint	Audience	Answers
`/livez`	Orchestrator (Kubernetes, Docker, Nomad, systemd)	Is the process alive? Should it be restarted?
`/readyz`	Load balancer, reverse proxy, service mesh	Can this instance serve traffic?
`/healthz`	Operators, dashboards, uptime monitors	What is the state of each component?

Liveness never checks dependencies. A failing database must never restart your container. Readiness runs all critical checks concurrently. If any fails, the instance is removed from the load balancer. Aggregate also runs non-critical checks and returns a JSON report for humans. Probe bodies stay empty to keep the wire minimal and the signal unambiguous. The HTTP status code is the only thing orchestrators read.

Status Vocabulary

Binary status, used for both components and the aggregate:

Status	Meaning
`ok`	The check passed. At the aggregate level: every critical check passed.
`error`	The check failed. At the aggregate level: at least one critical check failed.

Non-critical failures produce status: "error" on the individual check but do not flip the aggregate. The aggregate only goes to error when at least one critical check fails.

Function-Based API

Checks are plain functions. No base class to inherit from.

from grelmicro.health import HealthChecks, HealthDetails
from grelmicro.health.errors import HealthError

health = HealthChecks()


# Decorator form: register an async function under a name
@health.check("database")
async def check_database() -> HealthDetails | None:
    # Return None on success (healthy, no details)
    return None


@health.check("redis")
async def check_redis() -> HealthDetails | None:
    # Return a dict to include details (e.g. metrics)
    return {"latency_ms": 1.2, "version": "7.2"}


@health.check("external-api", critical=False)
async def check_external_api() -> HealthDetails | None:
    # Raise HealthError to expose a specific message in the error field.
    # Other exceptions produce an error formatted as "{ExceptionType}: {message}".
    msg = "Connection refused"
    raise HealthError(msg)

This mirrors FastAPI's own @app.get("/path") routing style and keeps grelmicro small. Return types are statically checked. JSONEncodable (recursive) includes primitives, datetime, nested dict / list / tuple, and any Mapping subclass. Type checkers (mypy, ty) catch non-serializable returns like bytes or custom objects at authoring time.

Concurrent Execution

All selected checks run in parallel via an asyncio.TaskGroup. A slow check does not block other checks. Each check runs with its own timeout (falling back to the default).