Writing Custom Health Checks in Python

Every serious product eventually has a health question that no off-the-shelf check can answer. "Are last night's settlement files non-empty and reconciled?" "Did the nightly export actually write rows for every active tenant?" "Is the feature-flag service returning the flag values we expect for our canary account?" These are business-logic questions, and answering them is exactly what custom health checks in Python are for. This guide is written for the engineer who has to build one: the anatomy of a check, how to return structured results, how to handle failure, and how these run safely inside a sandbox.

If you are still deciding what to monitor rather than how, start with what is product health monitoring. This post assumes you already know the signal you want to watch and need to express it in code.

Why built-in health checks aren't always enough

Managed check types (HTTP status, latency, TLS expiry, ping) cover the generic surface of any service. They are the right default and you should use them liberally. But they only know how to ask generic questions. They cannot know that your product considers itself unhealthy when the ratio of settled to pending transactions drops below a threshold, or when a specific downstream record goes stale.

That product-specific logic lives in your team's head, and a built-in check has no way to express it. A custom health check closes the gap. It is a small function, written by you, that encodes exactly what "healthy" means for one narrow slice of your product. It runs on the same schedule and status model as everything else. Python is the natural language for these because it is readable, ubiquitous on engineering teams, and has a client library for essentially everything you might need to inspect.

What makes a good custom health check

Before any code, the principles. A good custom health check is:

• Narrow. One check answers one question. "Is checkout healthy?" is three checks, not one. Narrow checks give precise status and precise alerts.
• Deterministic. Given the same product state, it returns the same status. Avoid checks that depend on wall-clock timing races or flaky external calls without retries.
• Fast and bounded. A health check is not a batch job. It should complete in seconds and never hang; it runs every few minutes, forever.
• Read-only. A check observes; it does not mutate product state. A check that "fixes" things is a cron job wearing a costume, and it will eventually cause an incident.
• Explicit about failure. It distinguishes "the thing I'm checking is unhealthy" from "I couldn't check," the healthy/degraded/unhealthy/unknown distinction from defining service health status.

The anatomy of a custom health check in Python

Concretely, a health check is a function with a known entrypoint that receives context and returns a structured result. A minimal, framework-agnostic shape looks like this:

# health check entrypoint. `context` is injected by the runner and carries
# the product's configuration and a scoped HTTP client. The function returns a
# dict describing the status; it never mutates anything.
def check(context):
    try:
        response = context.http.get(
            context.config["health_url"],
            timeout=5,
        )
        response.raise_for_status()
        return {"status": "healthy", "message": "endpoint responded 200"}
    except Exception as exc:
        # We could not complete the check; that is UNKNOWN, not unhealthy.
        return {"status": "unknown", "message": f"check failed to run: {exc}"}

Three things are worth naming here. The entrypoint is a single well-known function (check) so the runner knows what to call. The context is injected, so the check does not hard-code URLs or reach for ambient credentials. It receives what it needs. And the return value is a structured object, not a boolean, because a boolean cannot express degraded or unknown.

Example 1: validating an API response payload

The most common custom check inspects the content of a response, not just its status code. An API can return a perfect 200 with a body that proves the product is broken. Here we check that a pricing API returns a plausible, non-empty catalog with sane prices:

def check(context):
    try:
        resp = context.http.get(context.config["catalog_url"], timeout=5)
        resp.raise_for_status()
        catalog = resp.json()
    except Exception as exc:
        return {"status": "unknown", "message": f"could not fetch catalog: {exc}"}

    items = catalog.get("items", [])
    if not items:
        # 200 OK, but the product is empty; that is a real outage.
        return {"status": "unhealthy", "message": "catalog returned zero items"}

    # Business rule: no active item may have a non-positive price.
    bad = [i["sku"] for i in items if i.get("active") and i.get("price", 0) <= 0]
    if bad:
        return {
            "status": "unhealthy",
            "message": f"{len(bad)} active items priced <= 0: {bad[:5]}",
        }

    if len(items) < context.config.get("min_expected_items", 10):
        return {
            "status": "degraded",
            "message": f"catalog unexpectedly small: {len(items)} items",
        }

    return {"status": "healthy", "message": f"catalog OK ({len(items)} items)"}

Notice the graduated result: an empty catalog is unhealthy, a suspiciously small one is degraded, and a failure to fetch is unknown. The check encodes a business rule ("active items must have a positive price") that no generic HTTP check could ever know to enforce.

Example 2: checking a database record-count threshold

Another workhorse pattern: assert that a background process is keeping a table current. Suppose a nightly job should insert one summary row per active tenant. If it silently fails, the API stays up but the data rots. A custom check catches it:

from datetime import datetime, timedelta, timezone

def check(context):
    since = datetime.now(timezone.utc) - timedelta(hours=26)
    try:
        with context.db.connect() as conn:
            fresh = conn.scalar(
                "SELECT count(*) FROM daily_summary WHERE created_at >= %s",
                (since,),
            )
            active_tenants = conn.scalar(
                "SELECT count(*) FROM tenants WHERE status = 'active'"
            )
    except Exception as exc:
        return {"status": "unknown", "message": f"db query failed: {exc}"}

    if active_tenants == 0:
        return {"status": "unknown", "message": "no active tenants to compare"}

    coverage = fresh / active_tenants
    if coverage < 0.90:
        return {
            "status": "unhealthy",
            "message": f"summary coverage {coverage:.0%} ({fresh}/{active_tenants})",
        }
    if coverage < 1.0:
        return {
            "status": "degraded",
            "message": f"summary coverage {coverage:.0%}, some tenants missing",
        }
    return {"status": "healthy", "message": f"all {active_tenants} tenants covered"}

This is the kind of check that saves a team from discovering, three weeks later, that reporting has been quietly wrong. It is read-only, bounded, and returns a graded status keyed to a real threshold.

Returning structured status objects

Across all three examples the return contract is the same: a dict with a status drawn from a fixed vocabulary and a human-readable message. Keeping that contract consistent is what lets the platform roll many checks up into one product status and route alerts sensibly. A tiny helper keeps checks tidy and the vocabulary honest:

def result(status, message, **details):
    assert status in {"healthy", "degraded", "unhealthy", "unknown"}
    return {"status": status, "message": message, "details": details}

# usage: return result("degraded", "cache cold", hit_rate=0.42)

The details field is where you attach the numbers that made the decision: the coverage ratio, the offending SKUs, the measured latency. When someone opens the incident, those details are the difference between "it's degraded" and "it's degraded because coverage is 87% and these four tenants are missing."

Sandbox security considerations

Running arbitrary code on a schedule is a security decision, not just a feature. A custom health check runs your code, but the platform running it has to assume any code could be hostile or buggy, and contain it accordingly. If you are building or evaluating a system that executes custom checks, these are the controls that matter:

• Isolation. Each execution runs in its own sandbox with no access to other tenants' data or to the host. A check should never be able to see anything outside its own product context.
• Controlled egress. Network access is restricted to what the check legitimately needs. Unrestricted outbound is how a check becomes an exfiltration or SSRF vector.
• No ambient credentials. The check receives a scoped, short-lived credential for the product it monitors, never a broad key it can misuse. Secrets are injected per execution, not sitting in the environment.
• Resource and time limits. CPU, memory, and wall-clock caps prevent a runaway check from starving the system, and enforce the "fast and bounded" rule for you.

Testing a health check before you ship it

A health check is production code that runs unattended forever, so it deserves the same testing rigor as anything else on the critical path, if not more, because a broken check fails silently. A well-structured check is trivially testable: because context is injected rather than reached for, you can pass a fake context and assert on the returned status. No network, no database, no mocking frameworks required for the core logic.

import types
from mychecks.catalog import check

def fake_context(payload, min_items=10):
    ctx = types.SimpleNamespace()
    ctx.config = {"catalog_url": "http://x", "min_expected_items": min_items}
    ctx.http = types.SimpleNamespace(
        get=lambda url, timeout=5: types.SimpleNamespace(
            json=lambda: payload,
            raise_for_status=lambda: None,
        )
    )
    return ctx

def test_empty_catalog_is_unhealthy():
    r = check(fake_context({"items": []}))
    assert r["status"] == "unhealthy"

def test_negative_price_is_unhealthy():
    payload = {"items": [{"sku": "A", "active": True, "price": 0}]}
    assert check(fake_context(payload))["status"] == "unhealthy"

def test_small_catalog_is_degraded():
    payload = {"items": [{"sku": "A", "active": True, "price": 5}]}
    assert check(fake_context(payload, min_items=10))["status"] == "degraded"

Each branch of the status logic gets a test, and you can run them in your normal CI before the check ever touches production. This is also where you encode the tricky cases you thought of while writing the check (the empty result, the boundary threshold, the malformed payload) so that a future edit can't silently regress them. A check you have tested is a check you can trust to page you. An untested one is just another thing that might be lying.

How Sentrock runs custom Python health rules

Sentrock's custom Python health rules implement exactly this model. You write a check function, and Sentrock executes it in an isolated sandbox with restricted egress and per-execution resource limits. It injects the product's context (configuration and a scoped credential resolved just-in-time) so your code never handles raw secrets. The structured result you return maps to the same healthy, degraded, unhealthy, and unknown statuses as every built-in check. Your custom logic shows up on the same dashboard, on the same schedule, and with the same alerting as the rest of your monitoring. There is no separate system to babysit. A custom rule is just a health rule you happened to write in Python.

That uniformity is the point. The value of custom checks is not that they are exotic. It is that they let product-specific truth participate in your ordinary reliability practice instead of living in a forgotten cron job. Because the runner handles scheduling, isolation, credential resolution, status roll-up, and alerting, the only thing you own is the logic that is unique to your product. Everything a health check normally requires as scaffolding (where to run it, how to secure it, how to route its results) is provided. A fifteen-line function is a complete, production-grade check rather than the start of an infrastructure project.

Takeaways

Writing effective custom health checks in Python comes down to a few durable habits:

• Keep each check narrow, deterministic, fast, and strictly read-only.
• Return a structured status from a fixed vocabulary, never a bare boolean.
• Catch your own errors and return unknown so a broken check is never mistaken for a healthy product.
• Attach the deciding numbers in a details field for whoever responds to the alert.
• Run checks with least privilege, in isolation, with bounded resources.

The reward is monitoring that finally speaks your product's language. For how custom checks fit into a broader practice, read SaaS reliability monitoring best practices. Or write your first custom health rule and put one of these examples to work.

Related Articles

Monitoring

Defining Service Health Status

The healthy/degraded/unhealthy/unknown model your check's return values should map to.

Read Article →

Best Practices

SaaS Reliability Monitoring: Best Practices

Where custom checks fit into a complete, evergreen reliability monitoring practice.

Read Article →