Custom Actuator Endpoints Guide¶

PyFly's actuator module is fully extensible. You can create custom management endpoints that are auto-discovered from the DI container and exposed alongside the built-in health, beans, env, info, loggers, metrics, prometheus, and threaddump endpoints.

Table of Contents¶

ActuatorEndpoint Protocol
Creating a Custom Endpoint
Auto-Discovery
Write Operations
Routes Beyond the Actuator
Per-Endpoint Configuration
ActuatorRegistry
Built-in Endpoints Reference
Health
Beans
Environment
Info
Loggers
Metrics
Prometheus
Thread Dump
Index Endpoint
Complete Example

ActuatorEndpoint Protocol¶

Every actuator endpoint implements the ActuatorEndpoint protocol from pyfly.actuator.ports:

from typing import Any, Protocol, runtime_checkable

@runtime_checkable
class ActuatorEndpoint(Protocol):
    @property
    def endpoint_id(self) -> str:
        """URL path suffix: /actuator/{endpoint_id}."""
        ...

    @property
    def enabled(self) -> bool:
        """Default enable state. Can be overridden via config."""
        ...

    async def handle(self, context: Any = None) -> dict[str, Any]:
        """Handle a request and return a JSON-serializable dict."""
        ...

Property/Method	Description
`endpoint_id`	URL suffix — the endpoint is served at `/actuator/{endpoint_id}`
`enabled`	Default enable state. Can be overridden per-endpoint in config.
`handle(context)`	Returns a JSON-serializable dict for the response body.

Source: src/pyfly/actuator/ports.py

Creating a Custom Endpoint¶

Implement the ActuatorEndpoint protocol and decorate with @component for auto-discovery:

from pyfly.container import component


@component
class GitInfoEndpoint:
    """Exposes Git commit info at /actuator/git."""

    @property
    def endpoint_id(self) -> str:
        return "git"

    @property
    def enabled(self) -> bool:
        return True

    async def handle(self, context=None) -> dict:
        return {
            "branch": "main",
            "commit": {
                "id": "abc1234",
                "message": "feat: add git info endpoint",
                "time": "2026-02-15T10:30:00Z",
            },
        }

This endpoint will be available at GET /actuator/git and will appear in the /actuator index endpoint's _links.

Auto-Discovery¶

The ActuatorRegistry.discover_from_context(context) method scans all beans in the DI container for instances implementing the ActuatorEndpoint protocol:

# In create_app() when actuator_enabled=True:
registry = ActuatorRegistry(config=config)

# Register built-in endpoints
registry.register(HealthEndpoint(agg))
registry.register(BeansEndpoint(context))
# ...

# Auto-discover custom ActuatorEndpoint beans
registry.discover_from_context(context)

Any @component (or @service, @repository, etc.) that satisfies the ActuatorEndpoint protocol is automatically registered. No additional configuration is needed.

Write Operations¶

A custom endpoint is GET by default. To accept a command with a document, mark one method with @write_operation — the Spring @WriteOperation equivalent — and it is mounted as POST /actuator/{endpoint_id} (and POST /actuator/{endpoint_id}/{selector} when the endpoint sets supports_selector = True):

from typing import Any

from pyfly.actuator import write_operation
from pyfly.container import component


@component
class LevelEndpoint:
    @property
    def endpoint_id(self) -> str:
        return "level"

    @property
    def enabled(self) -> bool:
        return True

    async def handle(self, context: Any = None) -> dict[str, Any]:
        return {"level": self.level}

    @write_operation
    async def change(self, body: dict[str, Any], context: dict[str, Any]) -> dict[str, Any] | None:
        if body.get("level") not in ("debug", "info"):
            return {"error": f"unknown level {body.get('level')!r}"}   # -> 400
        self.level = body["level"]
        return None                                                     # -> 204

body is the request's JSON object; a missing body, a body that is not JSON or JSON that is not an object is refused with 400 before the method runs. context is the same {"query": ..., "selector": ...} the read operation receives. Return None for 204, a dict for 200 with that document, or a dict carrying "error" for 400. An endpoint declares at most one write operation.

Routes Beyond the Actuator¶

When a management operation does not fit the endpoint model — it needs its own request shape, its own authentication, a response that is not a JSON object — implement ManagementRoutesContributor from pyfly.web.ports:

from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from pyfly.container import component


@component
class GuardrailDryRunRoutes:
    def management_routes(self) -> list[Route]:
        async def dry_run(request: Request) -> JSONResponse:
            ...
        return [Route("/internal/guardrails/test", dry_run, methods=["POST"])]

The routes are mounted on the management surface: on the management app under its base path when pyfly.management.server.port selects a separate listener, on the main app when the management surface is shared (the default), and nowhere when management is disabled — exactly like the actuator. A contributor is an ordinary bean, so it is instantiated by ApplicationContext.start(); under the generated main.py that happens inside the app's lifespan, after create_app returned, and the post-start rescan mounts the routes then (the same rescan that finds late health indicators and @bean-produced controllers). A context started before create_app is served at build time. Either way management_routes() is called once per bean. The management port carries none of the application's security filters unless pyfly.management.security.enabled is on, so a contributed route that must not be public authenticates itself.

Per-Endpoint Configuration¶

Each endpoint's enable state can be overridden in pyfly.yaml:

pyfly:
  actuator:
    endpoints:
      health:
        enabled: true       # Keep health enabled (default)
      loggers:
        enabled: false      # Disable loggers in production
      git:
        enabled: true       # Enable custom git endpoint
      metrics:
        enabled: true       # Auto-enabled when prometheus_client is installed

The config key pattern is: pyfly.actuator.endpoints.{endpoint_id}.enabled

Priority order: 1. Config override (highest priority) 2. Endpoint's own enabled property (default)

ActuatorRegistry¶

The ActuatorRegistry (pyfly.actuator.registry) manages all endpoint instances:

from pyfly.actuator import ActuatorRegistry

registry = ActuatorRegistry(config=config)

# Register endpoints
registry.register(my_endpoint)

# Get only enabled endpoints
enabled = registry.get_enabled_endpoints()
# Returns: dict[str, ActuatorEndpoint]

# Auto-discover from DI context
registry.discover_from_context(application_context)

Method	Description
`register(endpoint)`	Register an `ActuatorEndpoint` instance.
`get_enabled_endpoints()`	Return all endpoints whose enable state is `True`.
`discover_from_context(context)`	Scan DI container for `ActuatorEndpoint` beans.

Source: src/pyfly/actuator/registry.py

Built-in Endpoints Reference¶

Health¶

Property	Value
Path	`/actuator/health`
Methods	GET
Default	Enabled
Special	Returns 200 (UP) or 503 (DOWN)

Aggregates all HealthIndicator beans. See the Actuator Guide for details.

Beans¶

Property	Value
Path	`/actuator/beans`
Methods	GET
Default	Enabled

Lists all registered beans with type, scope, and stereotype.

Environment¶

Property	Value
Path	`/actuator/env`
Methods	GET
Default	Enabled

Returns active configuration profiles.

Info¶

Property	Value
Path	`/actuator/info`
Methods	GET
Default	Enabled

Returns application metadata from pyfly.app.* config.

Loggers¶

Property	Value
Path	`/actuator/loggers`
Methods	GET, POST
Default	Enabled

GET: Lists all loggers with configured and effective levels, plus built-in logger groups. Levels use Spring Boot's vocabulary.

{
    "loggers": {
        "ROOT": {"configuredLevel": "INFO", "effectiveLevel": "INFO"},
        "pyfly.web": {"configuredLevel": null, "effectiveLevel": "INFO"}
    },
    "levels": ["OFF", "ERROR", "WARN", "INFO", "DEBUG", "TRACE"],
    "groups": {
        "web": {"configuredLevel": null, "members": ["pyfly.web", "uvicorn", "starlette"]},
        "sql": {"configuredLevel": null, "members": ["sqlalchemy", "pyfly.data"]}
    }
}

POST /actuator/loggers/{name}: Changes a logger's level at runtime. The logger name is a path parameter; the body carries configuredLevel (send null to reset to inherited). Returns 204 No Content on success.

curl -X POST http://localhost:8080/actuator/loggers/pyfly.web \
  -H "Content-Type: application/json" \
  -d '{"configuredLevel": "DEBUG"}'
# (204 No Content)

Source: src/pyfly/actuator/endpoints/loggers_endpoint.py

Metrics¶

Property	Value
Path	`/actuator/metrics` and `/actuator/metrics/{name}`
Methods	GET
Default	Enabled when `prometheus_client` installed

Exposes Prometheus registry data in Micrometer-compatible JSON (dot-case names, COUNT/TOTAL_TIME/MAX/VALUE statistics). Auto-configured by MetricsActuatorAutoConfiguration when prometheus_client is importable.

Source: src/pyfly/actuator/endpoints/metrics_endpoint.py

Prometheus¶

Property	Value
Path	`/actuator/prometheus`
Methods	GET
Default	Enabled when `prometheus_client` installed

Prometheus text exposition format (text/plain; version=0.0.4) scrape target. Auto-configured alongside the metrics endpoint.

Source: src/pyfly/actuator/endpoints/prometheus_endpoint.py

Thread Dump¶

Property	Value
Path	`/actuator/threaddump`
Methods	GET
Default	Enabled

Returns all live threads with their stack traces. className is the Python module name; methodName uses co_qualname (Python 3.11+) to include the enclosing class, falling back to co_name.

Source: src/pyfly/actuator/endpoints/threaddump_endpoint.py

Index Endpoint¶

GET /actuator returns a HAL-style index listing all enabled endpoints:

{
    "_links": {
        "self": {"href": "/actuator"},
        "health": {"href": "/actuator/health"},
        "beans": {"href": "/actuator/beans"},
        "env": {"href": "/actuator/env"},
        "info": {"href": "/actuator/info"},
        "loggers": {"href": "/actuator/loggers"},
        "git": {"href": "/actuator/git"}
    }
}

This index is automatically generated from the registry's enabled endpoints.

Complete Example¶

"""app.py — Application with custom actuator endpoints."""

from pyfly.container import component
from pyfly.core import pyfly_application, PyFlyApplication
from pyfly.web.adapters.starlette import create_app
from pyfly.actuator import HealthStatus


# ── Custom Actuator Endpoint ──────────────────────────────

@component
class CacheStatsEndpoint:
    """Exposes cache statistics at /actuator/cache-stats."""

    @property
    def endpoint_id(self) -> str:
        return "cache-stats"

    @property
    def enabled(self) -> bool:
        return True

    async def handle(self, context=None) -> dict:
        return {
            "hits": 1234,
            "misses": 56,
            "hit_rate": 0.956,
            "evictions": 12,
        }


@component
class FeatureFlagsEndpoint:
    """Exposes feature flag state at /actuator/features."""

    @property
    def endpoint_id(self) -> str:
        return "features"

    @property
    def enabled(self) -> bool:
        return True

    async def handle(self, context=None) -> dict:
        return {
            "flags": {
                "dark-mode": True,
                "beta-checkout": False,
                "new-search": True,
            }
        }


# ── Custom Health Indicator ───────────────────────────────

@component
class DatabaseHealthIndicator:
    async def health(self) -> HealthStatus:
        return HealthStatus(status="UP", details={"type": "postgresql"})


# ── Application ───────────────────────────────────────────

@pyfly_application(
    name="my-service",
    version="1.0.0",
    scan_packages=["app"],
)
class Application:
    pass


async def main():
    pyfly_app = PyFlyApplication(Application)
    await pyfly_app.startup()

    app = create_app(
        title="My Service",
        version="1.0.0",
        context=pyfly_app.context,
        actuator_enabled=True,
    )

    # Available endpoints:
    # GET /actuator              — index with _links
    # GET /actuator/health       — aggregated health
    # GET /actuator/beans        — bean registry
    # GET /actuator/env          — active profiles
    # GET /actuator/info         — app metadata
    # GET /actuator/loggers      — logger config
    # GET /actuator/cache-stats  — custom: cache statistics
    # GET /actuator/features     — custom: feature flags

    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

Configuration to disable a custom endpoint:

pyfly:
  actuator:
    endpoints:
      cache-stats:
        enabled: false  # Disable cache stats in production