feat: record FastMCP HTTP request and tool call metrics

FastMcpPrometheusInstrument runs prometheus-fastapi-instrumentator on every ASGI
application http_app() builds, labelled by route and excluding the metrics route,
and a FastMCP middleware counts and times tool calls in fastmcp_tool_calls_total
and fastmcp_tool_call_duration_seconds. The fastmcp-metrics extra now requires
prometheus-fastapi-instrumentator.

Author: lesnik512

docs/integrations/fastmcp.md

@@ -64,6 +64,13 @@ nanoseconds. A message that raises is logged at exception level and the exceptio
 
 Set `health_checks_enabled=False` to omit the health route.
 
+## Metrics
+
+With the `fastmcp-metrics` extra installed, the registry is served at `prometheus_metrics_path`,
+every application `application.http_app()` builds counts its requests by route, and every tool call
+is counted and timed. See [the configuration reference](../introduction/configuration.md#prometheus-fastmcp)
+for the metric names and parameters.
+
 ## Tracing
 
 With `opentelemetry_endpoint` set (and the `fastmcp-otl` extra installed), every ASGI application

docs/introduction/configuration.md

@@ -95,6 +95,24 @@ Additional parameters for FastAPI integration:
 - `prometheus_instrument_params` - passed to `method Instrumentator(...).instrument`
 - `prometheus_expose_params` - passed to `method Instrumentator(...).expose`.
 
+### Prometheus FastMCP
+
+Prometheus's integration for FastMCP uses the `prometheus_client` and `prometheus_fastapi_instrumentator`
+packages, and serves `prometheus_client.REGISTRY` at `prometheus_metrics_path`.
+
+Every ASGI application `application.http_app()` builds counts its requests in `http_requests_total`,
+`http_request_duration_seconds` and the instrumentator's other default metrics, labelled by route
+(`handler="/mcp"`); requests to unknown paths are grouped into `handler="none"`, and the metrics
+route itself is not counted. Tool calls are counted in `fastmcp_tool_calls_total{tool, status}`, with
+`status` either `success` or `error`, and timed in `fastmcp_tool_call_duration_seconds{tool}`.
+
+Additional parameters for FastMCP integration:
+
+- `prometheus_instrumentator_params` - passed to `prometheus_fastapi_instrumentator.Instrumentator`;
+  `excluded_handlers` defaults to the metrics route.
+- `prometheus_instrument_params` - passed to `method Instrumentator(...).instrument`.
+- `fastmcp_prometheus_tool_metrics_enabled` - count and time tool calls (default: `True`).
+
 
 ## Opentelemetry
 

lite_bootstrap/bootstrappers/fastmcp_bootstrapper.py

@@ -1,6 +1,7 @@
 import contextlib
 import dataclasses
 import functools
+import inspect
 import re
 import time
 import typing
@@ -18,13 +19,13 @@
 
 if import_checker.is_fastmcp_installed:
     from fastmcp import FastMCP
+    from fastmcp.server.http import StarletteWithLifespan
     from fastmcp.server.middleware import Middleware, MiddlewareContext
     from fastmcp.server.providers import Provider
     from starlette.requests import Request
     from starlette.responses import JSONResponse, Response
 
 if import_checker.is_fastmcp_opentelemetry_installed:
-    from fastmcp.server.http import StarletteWithLifespan
     from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
     from opentelemetry.metrics import get_meter_provider
     from opentelemetry.trace import get_tracer_provider
@@ -40,10 +41,19 @@
 if import_checker.is_prometheus_client_installed:
     import prometheus_client
 
+if import_checker.is_prometheus_fastapi_instrumentator_installed:
+    from prometheus_fastapi_instrumentator import Instrumentator
+    from prometheus_fastapi_instrumentator import metrics as instrumentator_metrics
+
+    _DEFAULT_METRICS_PARAMETERS: typing.Final = frozenset(inspect.signature(instrumentator_metrics.default).parameters)
+
 
 # OpenTelemetryMiddleware matches its patterns against a full URL, not a bare path.
 _EXCLUDED_URL_SCHEME_AND_HOST: typing.Final = r"^\w+://[^/]*"
 
+_TOOL_CALL_STATUS_SUCCESS: typing.Final = "success"
+_TOOL_CALL_STATUS_ERROR: typing.Final = "error"
+
 # Set by StarletteInstrumentor too, so an application is never traced twice
 _OPENTELEMETRY_INSTRUMENTED_MARKER: typing.Final = "_is_instrumented_by_opentelemetry"
 
@@ -78,6 +88,24 @@ def restore() -> None:
     return restore
 
 
+def _build_tool_call_metrics() -> tuple["prometheus_client.Counter", "prometheus_client.Histogram"]:
+    # The global registry rejects a second collector of the same name, so a later bootstrap reuses the first
+    registered_collectors: typing.Final = prometheus_client.REGISTRY._names_to_collectors  # noqa: SLF001
+    if "fastmcp_tool_calls_total" in registered_collectors:
+        return (
+            typing.cast("prometheus_client.Counter", registered_collectors["fastmcp_tool_calls_total"]),
+            typing.cast("prometheus_client.Histogram", registered_collectors["fastmcp_tool_call_duration_seconds"]),
+        )
+    return (
+        prometheus_client.Counter(
+            "fastmcp_tool_calls_total", "Number of MCP tool calls by tool and outcome.", ["tool", "status"]
+        ),
+        prometheus_client.Histogram(
+            "fastmcp_tool_call_duration_seconds", "Duration of MCP tool calls by tool.", ["tool"]
+        ),
+    )
+
+
 def build_fastmcp_route_details_from_scope(
     scope: "Scope",
     routes: "typing.Iterable[BaseRoute]",
@@ -106,6 +134,32 @@ async def lifespan(self) -> AsyncGenerator[None]:
             finally:
                 self._teardown()
 
+    class FastMcpPrometheusMiddleware(Middleware):
+        def __init__(
+            self,
+            tool_calls_total: "prometheus_client.Counter",
+            tool_call_duration_seconds: "prometheus_client.Histogram",
+        ) -> None:
+            self.tool_calls_total = tool_calls_total
+            self.tool_call_duration_seconds = tool_call_duration_seconds
+
+        async def on_call_tool(
+            self,
+            context: "MiddlewareContext[typing.Any]",
+            call_next: "typing.Callable[[MiddlewareContext[typing.Any]], typing.Awaitable[typing.Any]]",
+        ) -> typing.Any:  # noqa: ANN401
+            tool_name: typing.Final = context.message.name
+            start_time: typing.Final = time.perf_counter()
+            try:
+                result = await call_next(context)
+            except Exception:
+                self.tool_calls_total.labels(tool=tool_name, status=_TOOL_CALL_STATUS_ERROR).inc()
+                raise
+            finally:
+                self.tool_call_duration_seconds.labels(tool=tool_name).observe(time.perf_counter() - start_time)
+            self.tool_calls_total.labels(tool=tool_name, status=_TOOL_CALL_STATUS_SUCCESS).inc()
+            return result
+
     class FastMcpLoggingMiddleware(Middleware):
         async def on_message(
             self,
@@ -142,6 +196,9 @@ class FastMcpConfig(
 ):
     application: "FastMCP[typing.Any]" = dataclasses.field(default_factory=_make_fastmcp)
     fastmcp_logging_middleware_enabled: bool = False
+    fastmcp_prometheus_tool_metrics_enabled: bool = True
+    prometheus_instrumentator_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict)
+    prometheus_instrument_params: dict[str, typing.Any] = dataclasses.field(default_factory=dict)
 
 
 @dataclasses.dataclass(kw_only=True)
@@ -221,14 +278,46 @@ def teardown(self) -> None:
 @dataclasses.dataclass(kw_only=True)
 class FastMcpPrometheusInstrument(PrometheusInstrument):
     bootstrap_config: FastMcpConfig
-    missing_dependency_message = "prometheus_client is not installed"
+    missing_dependency_message = "prometheus_client or prometheus_fastapi_instrumentator is not installed"
+    _restore_http_app: typing.Callable[[], None] | None = dataclasses.field(
+        default=None, init=False, repr=False, compare=False
+    )
+    _default_metrics: typing.Callable[..., typing.Any] | None = dataclasses.field(
+        default=None, init=False, repr=False, compare=False
+    )
 
     @staticmethod
     def dependencies_installed() -> bool:
-        return import_checker.is_prometheus_client_installed
+        return (
+            import_checker.is_prometheus_client_installed
+            and import_checker.is_prometheus_fastapi_instrumentator_installed
+        )
+
+    def _instrument_http_app(self, http_application: "StarletteWithLifespan") -> "StarletteWithLifespan":
+        config: typing.Final = self.bootstrap_config
+        instrumentator_params: typing.Final[dict[str, typing.Any]] = {
+            "excluded_handlers": [f"^{re.escape(config.prometheus_metrics_path)}$"],
+            **config.prometheus_instrumentator_params,
+        }
+        # A second http_app() would register the default metrics again, which the instrumentator
+        # treats as a duplicate and silently skips, so every application shares the first set.
+        if self._default_metrics is None:
+            default_metrics_params: typing.Final[dict[str, typing.Any]] = {
+                name: value
+                for name, value in {**instrumentator_params, **config.prometheus_instrument_params}.items()
+                if name in _DEFAULT_METRICS_PARAMETERS
+            }
+            self._default_metrics = instrumentator_metrics.default(**default_metrics_params)
+        Instrumentator(**instrumentator_params).add(self._default_metrics).instrument(
+            http_application, **config.prometheus_instrument_params
+        )
+        return http_application
 
     def bootstrap(self) -> None:
         config = self.bootstrap_config
+        self._restore_http_app = _postprocess_http_apps(config.application, self._instrument_http_app)
+        if config.fastmcp_prometheus_tool_metrics_enabled:
+            config.application.add_middleware(FastMcpPrometheusMiddleware(*_build_tool_call_metrics()))
 
         @config.application.custom_route(
             config.prometheus_metrics_path,
@@ -242,6 +331,11 @@ async def metrics_handler(_: "Request") -> "Response":
                 headers={"content-type": prometheus_client.CONTENT_TYPE_LATEST},
             )
 
+    def teardown(self) -> None:
+        if self._restore_http_app is not None:
+            self._restore_http_app()
+            self._restore_http_app = None
+
 
 @dataclasses.dataclass(kw_only=True)
 class FastMcpLoggingInstrument(LoggingInstrument):

pyproject.toml

@@ -203,6 +203,8 @@ fastmcp-otl = [
 fastmcp-metrics = [
     "lite-bootstrap[fastmcp]",
     "prometheus-client>=0.20",
+    # >=8.0.1 to share the fastapi-metrics floor; it requires starlette>=1.0, which fastmcp 3.0 already pulls.
+    "prometheus-fastapi-instrumentator>=8.0.1",
 ]
 fastmcp-all = [
     "lite-bootstrap[fastmcp,fastmcp-otl,fastmcp-metrics,sentry,logging,pyroscope]",

tests/test_fastmcp_bootstrap.py

@@ -7,7 +7,8 @@
 
 import prometheus_client
 import pytest
-from fastmcp import FastMCP
+from fastmcp import Client, FastMCP
+from fastmcp.exceptions import ToolError
 from fastmcp.server.http import StarletteWithLifespan
 from fastmcp.server.middleware import MiddlewareContext
 from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
@@ -25,6 +26,7 @@
 from lite_bootstrap.bootstrappers.fastmcp_bootstrapper import (
     FastMcpLoggingMiddleware,
     FastMcpOpenTelemetryInstrument,
+    FastMcpPrometheusMiddleware,
     _postprocess_http_apps,
 )
 from lite_bootstrap.exceptions import ConfigurationError
@@ -446,7 +448,6 @@ def test_fastmcp_otel_leaves_http_app_alone_when_not_configured() -> None:
     bootstrapper = FastMcpBootstrapper(bootstrap_config=_make_test_config())
     application = bootstrapper.bootstrap()
     try:
-        assert "http_app" not in vars(application)
         assert _count_opentelemetry_middlewares(application.http_app()) == 0
     finally:
         bootstrapper.teardown()
@@ -462,6 +463,140 @@ def test_fastmcp_otel_is_skipped_without_asgi_instrumentation() -> None:
                 bootstrap_config=_make_test_config(opentelemetry_log_traces=True)
             )
         application = bootstrapper.bootstrap()
+        try:
+            assert _count_opentelemetry_middlewares(application.http_app()) == 0
+        finally:
+            bootstrapper.teardown()
+
+
+@pytest.fixture(autouse=True)
+def _unregister_request_metrics() -> None:
+    """Give each test unregistered request metrics: a second registration is skipped, not reused."""
+    registered_collectors = dict(prometheus_client.REGISTRY._collector_to_names)  # noqa: SLF001
+    for collector, names in registered_collectors.items():
+        if any(name.startswith(("http_request", "http_response")) for name in names):
+            prometheus_client.REGISTRY.unregister(collector)
+
+
+def _sample_value(name: str, labels: dict[str, str]) -> float:
+    return prometheus_client.REGISTRY.get_sample_value(name, labels) or 0.0
+
+
+def _http_requests_total(handler: str, method: str = "GET", status_group: str = "2xx") -> float:
+    return _sample_value("http_requests_total", {"handler": handler, "method": method, "status": status_group})
+
+
+def test_fastmcp_prometheus_counts_http_requests_by_route_template() -> None:
+    health_checks_path = f"/health-{uuid.uuid4().hex}/"
+    config = _make_test_config(health_checks_path=health_checks_path)
+    bootstrapper = FastMcpBootstrapper(bootstrap_config=config)
+    application = bootstrapper.bootstrap()
+    try:
+        with TestClient(application.http_app()) as client:
+            client.get(health_checks_path)
+            client.get(health_checks_path)
+            client.get("/missing")
+            client.get(config.prometheus_metrics_path)
+    finally:
+        bootstrapper.teardown()
+
+    expected_health_requests = 2
+    assert _http_requests_total(health_checks_path) == expected_health_requests
+    assert _http_requests_total("none", status_group="4xx") == 1
+    assert _http_requests_total(config.prometheus_metrics_path) == 0
+
+
+def test_fastmcp_prometheus_counts_requests_of_every_http_app() -> None:
+    health_checks_path = f"/health-{uuid.uuid4().hex}/"
+    bootstrapper = FastMcpBootstrapper(bootstrap_config=_make_test_config(health_checks_path=health_checks_path))
+    application = bootstrapper.bootstrap()
+    try:
+        for http_application in (application.http_app(), application.http_app(path="/other")):
+            with TestClient(http_application) as client:
+                client.get(health_checks_path)
+    finally:
+        bootstrapper.teardown()
+
+    expected_requests = 2
+    assert _http_requests_total(health_checks_path) == expected_requests
+
+
+def test_fastmcp_prometheus_instrumentator_params_are_passed() -> None:
+    health_checks_path = f"/health-{uuid.uuid4().hex}/"
+    config = _make_test_config(
+        health_checks_path=health_checks_path,
+        prometheus_instrumentator_params={"excluded_handlers": [health_checks_path]},
+    )
+    bootstrapper = FastMcpBootstrapper(bootstrap_config=config)
+    application = bootstrapper.bootstrap()
+    try:
+        with TestClient(application.http_app()) as client:
+            client.get(health_checks_path)
+    finally:
+        bootstrapper.teardown()
+
+    assert _http_requests_total(health_checks_path) == 0
+
+
+def test_fastmcp_prometheus_teardown_restores_http_app() -> None:
+    bootstrapper = FastMcpBootstrapper(bootstrap_config=_make_test_config())
+    application = bootstrapper.bootstrap()
+    assert "http_app" in vars(application)
+
+    bootstrapper.teardown()
+
+    assert "http_app" not in vars(application)
+
+
+async def test_fastmcp_prometheus_counts_tool_calls() -> None:
+    bootstrapper = FastMcpBootstrapper(bootstrap_config=_make_test_config())
+    application = bootstrapper.bootstrap()
+    echo_name = f"echo_{uuid.uuid4().hex}"
+    failing_name = f"failing_{uuid.uuid4().hex}"
+
+    def echo(text: str) -> str:
+        return text
+
+    def failing() -> str:
+        message = "boom"
+        raise ValueError(message)
+
+    application.tool(echo, name=echo_name)
+    application.tool(failing, name=failing_name)
+    try:
+        async with Client(application) as client:
+            await client.call_tool(echo_name, {"text": "hi"})
+            await client.call_tool(echo_name, {"text": "hi"})
+            with pytest.raises(ToolError):
+                await client.call_tool(failing_name, {})
+    finally:
+        bootstrapper.teardown()
+
+    expected_successes = 2
+    assert _sample_value("fastmcp_tool_calls_total", {"tool": echo_name, "status": "success"}) == expected_successes
+    assert _sample_value("fastmcp_tool_calls_total", {"tool": failing_name, "status": "error"}) == 1
+    assert _sample_value("fastmcp_tool_call_duration_seconds_count", {"tool": echo_name}) == expected_successes
+
+
+def test_fastmcp_prometheus_tool_metrics_can_be_disabled() -> None:
+    bootstrapper = FastMcpBootstrapper(
+        bootstrap_config=_make_test_config(fastmcp_prometheus_tool_metrics_enabled=False)
+    )
+    application = bootstrapper.bootstrap()
+    try:
+        assert not [one for one in application.middleware if isinstance(one, FastMcpPrometheusMiddleware)]
+    finally:
+        bootstrapper.teardown()
+
+
+def test_fastmcp_prometheus_is_skipped_without_instrumentator() -> None:
+    with emulate_package_missing_with_module_reload(
+        "prometheus_fastapi_instrumentator",
+        ["lite_bootstrap.bootstrappers.fastmcp_bootstrapper"],
+    ):
+        with pytest.warns(UserWarning, match="prometheus_fastapi_instrumentator"):
+            bootstrapper = fastmcp_bootstrapper.FastMcpBootstrapper(bootstrap_config=_make_test_config())
+        application = bootstrapper.bootstrap()
         try:
             assert "http_app" not in vars(application)
         finally: