Add fastmcp bootstrapper

README.md

@@ -48,6 +48,7 @@ Those instruments can be bootstrapped for:
 - `fastapi`,
 - `litestar`,
 - or `faststream` service,
+- or `fastmcp` service,
 - or even a service that doesn't use one of these frameworks.
 
 Interested? Let's dive right in ⚡
@@ -78,6 +79,7 @@ Also, you can specify extras during installation for concrete framework:
 - `fastapi`
 - `litestar`
 - `faststream` (ASGI app)
+- `fastmcp`
 
 Also we have `granian` extra that is requires for `create_granian_server`.
 
@@ -198,6 +200,28 @@ settings = YourSettings()
 application: AsgiFastStream = FastStreamBootstrapper(settings).bootstrap()
 ```
 
+### FastMCP
+
+```python
+from fastmcp import FastMCP
+
+from microbootstrap import FastMcpSettings
+from microbootstrap.bootstrappers.fastmcp import FastMcpBootstrapper
+
+
+class YourSettings(FastMcpSettings):
+    service_debug: bool = False
+    service_name: str = "my-awesome-mcp-service"
+    service_description: str = "MCP server for internal tools"
+
+    sentry_dsn: str = "your-sentry-dsn"
+
+
+settings = YourSettings()
+
+application: FastMCP = FastMcpBootstrapper(settings).bootstrap()
+```
+
 ## Settings
 
 The settings object is the core of microbootstrap.

examples/fastmcp_app.py

@@ -0,0 +1,17 @@
+from fastmcp import FastMCP
+
+from microbootstrap import FastMcpSettings
+from microbootstrap.bootstrappers.fastmcp import FastMcpBootstrapper
+
+
+class Settings(FastMcpSettings):
+    service_name: str = "example-mcp"
+    service_description: str = "Example FastMCP service"
+
+
+application: FastMCP = FastMcpBootstrapper(Settings()).bootstrap()
+
+
+@application.tool
+def greet_person(person_name: str) -> str:
+    return f"Hello, {person_name}!"

microbootstrap/__init__.py

@@ -8,6 +8,7 @@
 )
 from microbootstrap.instruments.prometheus_instrument import (
     FastApiPrometheusConfig,
+    FastMcpPrometheusConfig,
     FastStreamPrometheusConfig,
     FastStreamPrometheusMiddlewareProtocol,
     LitestarPrometheusConfig,
@@ -17,6 +18,7 @@
 from microbootstrap.instruments.swagger_instrument import SwaggerConfig
 from microbootstrap.settings import (
     FastApiSettings,
+    FastMcpSettings,
     FastStreamSettings,
     InstrumentsSetupperSettings,
     LitestarSettings,
@@ -27,6 +29,8 @@
     "CorsConfig",
     "FastApiPrometheusConfig",
     "FastApiSettings",
+    "FastMcpPrometheusConfig",
+    "FastMcpSettings",
     "FastStreamOpentelemetryConfig",
     "FastStreamPrometheusConfig",
     "FastStreamPrometheusMiddlewareProtocol",

microbootstrap/bootstrappers/fastmcp.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+import typing
+
+import prometheus_client
+import typing_extensions
+from fastmcp import FastMCP
+from starlette.responses import JSONResponse, Response
+
+from microbootstrap.bootstrappers.base import ApplicationBootstrapper
+from microbootstrap.config.fastmcp import FastMcpConfig
+from microbootstrap.instruments.health_checks_instrument import HealthChecksInstrument, HealthCheckTypedDict
+from microbootstrap.instruments.logging_instrument import LoggingInstrument
+from microbootstrap.instruments.prometheus_instrument import FastMcpPrometheusConfig, PrometheusInstrument
+from microbootstrap.instruments.pyroscope_instrument import PyroscopeInstrument
+from microbootstrap.instruments.sentry_instrument import SentryInstrument
+from microbootstrap.middlewares.fastmcp import FastMcpLoggingMiddleware
+from microbootstrap.settings import FastMcpSettings
+
+
+if typing.TYPE_CHECKING:
+    from starlette.requests import Request
+
+
+class KwargsFastMCP(FastMCP[typing.Any]):
+    def __init__(self, **kwargs: typing.Any) -> None:  # noqa: ANN401
+        super().__init__(**kwargs)
+
+
+class FastMcpBootstrapper(
+    ApplicationBootstrapper[FastMcpSettings, FastMCP[typing.Any], FastMcpConfig],
+):
+    application_config = FastMcpConfig()
+    application_type = KwargsFastMCP
+
+    def bootstrap_before(self: typing_extensions.Self) -> dict[str, typing.Any]:
+        return {
+            "name": self.application_config.name or self.settings.service_name,
+            "instructions": self.application_config.instructions or self.settings.service_description,
+            "version": self.application_config.version or self.settings.service_version,
+        }
+
+    def bootstrap_before_instruments_after_app_created(
+        self,
+        application: FastMCP[typing.Any],
+    ) -> FastMCP[typing.Any]:
+        self.console_writer.print_bootstrap_table()
+        return application
+
+
+FastMcpBootstrapper.use_instrument()(SentryInstrument)
+FastMcpBootstrapper.use_instrument()(PyroscopeInstrument)
+
+
+@FastMcpBootstrapper.use_instrument()
+class FastMcpLoggingInstrument(LoggingInstrument):
+    def bootstrap_after(self, application: FastMCP[typing.Any]) -> FastMCP[typing.Any]:  # type: ignore[override]
+        if not self.instrument_config.logging_turn_off_middleware:
+            application.add_middleware(FastMcpLoggingMiddleware())
+        return application
+
+
+@FastMcpBootstrapper.use_instrument()
+class FastMcpHealthChecksInstrument(HealthChecksInstrument):
+    def bootstrap_after(self, application: FastMCP[typing.Any]) -> FastMCP[typing.Any]:  # type: ignore[override]
+        @application.custom_route(
+            self.instrument_config.health_checks_path,
+            methods=["GET"],
+            name="health_check",
+            include_in_schema=self.instrument_config.health_checks_include_in_schema,
+        )
+        async def health_check_handler(request: Request) -> JSONResponse:  # noqa: ARG001
+            response_data: HealthCheckTypedDict = self.render_health_check_data()
+            return JSONResponse(response_data)
+
+        return application
+
+
+@FastMcpBootstrapper.use_instrument()
+class FastMcpPrometheusInstrument(PrometheusInstrument[FastMcpPrometheusConfig]):
+    def bootstrap_after(self, application: FastMCP[typing.Any]) -> FastMCP[typing.Any]:  # type: ignore[override]
+        if not self.instrument_config.prometheus_register_route:
+            return application
+
+        @application.custom_route(
+            self.instrument_config.prometheus_metrics_path,
+            methods=["GET"],
+            name="metrics",
+            include_in_schema=self.instrument_config.prometheus_metrics_include_in_schema,
+        )
+        async def metrics_handler(request: Request) -> Response:  # noqa: ARG001
+            registry: typing.Final = self.instrument_config.prometheus_registry or prometheus_client.REGISTRY
+            return Response(
+                prometheus_client.generate_latest(registry),
+                headers={"content-type": prometheus_client.CONTENT_TYPE_LATEST},
+            )
+
+        return application
+
+    @classmethod
+    def get_config_type(cls) -> type[FastMcpPrometheusConfig]:
+        return FastMcpPrometheusConfig

microbootstrap/bootstrappers/litestar.py

@@ -160,8 +160,8 @@ def __init__(self, config: OpenTelemetryConfig) -> None:
     def create_open_telemetry_middleware(self, app: ASGIApp) -> OpenTelemetryMiddleware:
         return OpenTelemetryMiddleware(
             app=app,
-            client_request_hook=self.config.client_request_hook_handler,  # type: ignore[arg-type]
-            client_response_hook=self.config.client_response_hook_handler,  # type: ignore[arg-type]
+            client_request_hook=self.config.client_request_hook_handler,
+            client_response_hook=self.config.client_response_hook_handler,
             default_span_details=build_litestar_route_details_from_scope,
             excluded_urls=get_excluded_urls(self.config.exclude_urls_env_key),
             meter=self.config.meter,

microbootstrap/config/fastmcp.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+import dataclasses
+import typing
+
+
+if typing.TYPE_CHECKING:
+    import mcp.types
+    from fastmcp.client.sampling import SamplingHandler
+    from fastmcp.server.auth import AuthProvider
+    from fastmcp.server.lifespan import Lifespan
+    from fastmcp.server.middleware import Middleware as FastMcpMiddleware
+    from fastmcp.server.providers import Provider
+    from fastmcp.server.server import DuplicateBehavior, LifespanCallable
+    from fastmcp.server.transforms import Transform
+    from fastmcp.tools.base import Tool
+    from key_value.aio.protocols import AsyncKeyValue
+
+
+@dataclasses.dataclass
+class FastMcpConfig:
+    name: str | None = None
+    instructions: str | None = None
+    version: str | int | float | None = None
+    website_url: str | None = None
+    icons: list[mcp.types.Icon] | None = None
+    auth: AuthProvider | None = None
+    middleware: typing.Sequence[FastMcpMiddleware] | None = None
+    providers: typing.Sequence[Provider] | None = None
+    transforms: typing.Sequence[Transform] | None = None
+    lifespan: LifespanCallable | Lifespan | None = None
+    tools: typing.Sequence[Tool | typing.Callable[..., typing.Any]] | None = None
+    on_duplicate: DuplicateBehavior | None = None
+    mask_error_details: bool | None = None
+    dereference_schemas: bool = True
+    strict_input_validation: bool | None = None
+    list_page_size: int | None = None
+    tasks: bool | None = None
+    session_state_store: AsyncKeyValue | None = None
+    sampling_handler: SamplingHandler[typing.Any, typing.Any] | None = None
+    sampling_handler_behavior: typing.Literal["always", "fallback"] | None = None
+    client_log_level: mcp.types.LoggingLevel | None = None
+    experimental_capabilities: dict[str, dict[str, typing.Any]] | None = None

microbootstrap/instruments/prometheus_instrument.py

@@ -32,6 +32,11 @@ class FastApiPrometheusConfig(BasePrometheusConfig):
     prometheus_custom_labels: dict[str, typing.Any] = pydantic.Field(default_factory=dict)
 
 
+class FastMcpPrometheusConfig(BasePrometheusConfig):
+    prometheus_registry: typing.Any | None = None
+    prometheus_register_route: bool = True
+
+
 @typing.runtime_checkable
 class FastStreamPrometheusMiddlewareProtocol(typing.Protocol):
     def __init__(

microbootstrap/middlewares/fastmcp.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+import time
+import typing
+
+import structlog
+from fastmcp.server.middleware import Middleware, MiddlewareContext
+
+
+if typing.TYPE_CHECKING:
+    from fastmcp.server.middleware import CallNext
+
+
+fastmcp_access_logger: typing.Final = structlog.get_logger("mcp.access")
+
+
+class FastMcpLoggingMiddleware(Middleware):
+    async def on_message(
+        self,
+        context: MiddlewareContext[typing.Any],
+        call_next: CallNext[typing.Any, typing.Any],
+    ) -> typing.Any:  # noqa: ANN401
+        start_time: typing.Final = time.perf_counter_ns()
+        try:
+            result: typing.Final = await call_next(context)
+        except Exception:
+            fastmcp_access_logger.exception(
+                context.method or "unknown",
+                mcp={
+                    "method": context.method,
+                    "source": context.source,
+                    "type": context.type,
+                },
+                duration=time.perf_counter_ns() - start_time,
+            )
+            raise
+
+        fastmcp_access_logger.info(
+            context.method or "unknown",
+            mcp={
+                "method": context.method,
+                "source": context.source,
+                "type": context.type,
+            },
+            duration=time.perf_counter_ns() - start_time,
+        )
+        return result

microbootstrap/settings.py

@@ -8,6 +8,7 @@
 from microbootstrap import (
     CorsConfig,
     FastApiPrometheusConfig,
+    FastMcpPrometheusConfig,
     FastStreamOpentelemetryConfig,
     FastStreamPrometheusConfig,
     HealthChecksConfig,
@@ -102,6 +103,18 @@ class FastStreamSettings(  # type: ignore[misc]
     asyncapi_path: str | None = "/asyncapi"
 
 
+class FastMcpSettings(  # type: ignore[misc]
+    BaseServiceSettings,
+    ServerConfig,
+    LoggingConfig,
+    SentryConfig,
+    FastMcpPrometheusConfig,
+    HealthChecksConfig,
+    PyroscopeConfig,
+):
+    """Settings for a fastmcp bootstrap."""
+
+
 class InstrumentsSetupperSettings(  # type: ignore[misc]
     BaseServiceSettings,
     LoggingConfig,

pyproject.toml

@@ -73,6 +73,7 @@ litestar = [
 ]
 granian = ["granian[reload]>=1"]
 faststream = ["faststream~=0.6.2", "prometheus-client>=0.20"]
+fastmcp = ["fastmcp>=2,<4", "prometheus-client>=0.20"]
 
 [dependency-groups]
 dev = [