"""Server Lifecycle Management. This module handles starting, running, and stopping the MCP Hangar server. It manages signal handling for graceful shutdown. The lifecycle flow: 1. Setup logging based on CLI config 2. Bootstrap application 3. Start background components 4. Run appropriate server mode (stdio or HTTP) 5. Handle shutdown on exit/signal """ import asyncio from collections.abc import Callable from dataclasses import dataclass import ipaddress from pathlib import Path import signal import sys import threading from typing import Any import yaml from ..errors import bounded_error_type from ..logging_config import get_logger, setup_logging from .api.middleware import create_auth_enforced_app from .bootstrap import ApplicationContext, bootstrap from .cli.cli_compat import CLIConfig from .config import http_graceful_shutdown_timeout, load_config_from_file from .bootstrap.coordination import get_event_tailer, get_lease_keeper from .bootstrap.workers import start_background_workers from .catalogue_readiness import CatalogueRetry from .state import get_discovery_orchestrator, get_runtime_mcp_servers logger = get_logger(__name__) def start_discovery_loop(orchestrator: Any) -> tuple[asyncio.AbstractEventLoop, threading.Thread]: """Start *orchestrator* on its own long-lived event loop, in its own thread. The orchestrator's cycle is a task on the loop it starts on, so that loop has to outlive the call that starts it: not a transport's loop, and not a `run_until_complete` that returns. `ServerLifecycle` and the `Hangar` facade both run discovery this way. Args: orchestrator: The `DiscoveryOrchestrator` bootstrap built. Returns: The loop and the thread running it, for `stop_discovery_loop`. Raises: Exception: Whatever `orchestrator.start()` raised; the loop is stopped and its thread joined first. """ loop = asyncio.new_event_loop() def run_loop() -> None: asyncio.set_event_loop(loop) loop.run_forever() loop.close() thread = threading.Thread(target=run_loop, name="mcp-hangar-discovery", daemon=True) thread.start() try: asyncio.run_coroutine_threadsafe(orchestrator.start(), loop).result() except Exception: loop.call_soon_threadsafe(loop.stop) thread.join() raise logger.info("discovery_started", sources_count=orchestrator.get_stats()["sources_count"]) return loop, thread def stop_discovery_loop(orchestrator: Any, loop: asyncio.AbstractEventLoop, thread: threading.Thread) -> None: """Await *orchestrator*'s cleanup on *loop*, then stop the loop and join *thread*. A failed cleanup is logged and the loop is stopped anyway: shutdown has to finish, and a retained loop kept the process alive after it returned. The sources hear of the stop first, from this thread: one can be blocked on the loop's own thread, and nothing scheduled on the loop runs until it returns. """ try: orchestrator.request_stop() asyncio.run_coroutine_threadsafe(orchestrator.stop(), loop).result() except Exception as e: # noqa: BLE001 -- shutdown must continue after discovery cleanup failure logger.warning("discovery_orchestrator_stop_failed", error=str(e)) finally: loop.call_soon_threadsafe(loop.stop) thread.join() def build_readiness_report(repository: Any) -> tuple[dict[str, Any], int]: """Return the ``/health/ready`` body and HTTP status. Readiness answers one question: **can this gateway accept and route a call?** It deliberately does NOT require a warm backend. Hangar starts backends lazily and shuts them down on ``idle_ttl_s``, so "every backend cold" is the normal steady state of an idle gateway -- and a cold backend starts on the next call. Gating readiness on a warm backend deadlocked Kubernetes deployments: the last backend goes idle -> 503 -> the pod leaves the Service endpoints -> no call can arrive -> nothing ever warms a backend again. The chart wires ``readinessProbe`` to this endpoint, so a pod could sit NotReady indefinitely while the process was perfectly healthy (#599). Backend state is still reported in the body for observability, and the always-on health framework already treats backend availability as *degraded*, not unhealthy (``create_mcp_server_health_check`` is ``critical=False``) -- alerting, not the traffic gate, is where it belongs. What DOES fail readiness: the event store silently degrading to a non-durable in-memory store while a durable driver was configured. That one is a real "do not send me writes I cannot audit" condition. And, on a front door with ``tool_access.required_catalogue``, a replica whose boot warm-up has not yet projected every server on that list (#1446): the ``catalogue`` field counts what is missing. This endpoint answers without authentication, so it carries counts only, and the ids are logged (`catalogue_readiness`). That is not a warm-backend rule, and it does not bring #599 back. It asks whether a server was projected **once**, and a projection outlives the server's stop, so an idle or failed backend never makes a ready replica not ready again. And it is bounded: once ``retry_for_s`` has passed since the configuration was first applied, readiness stops looking at the catalogue and this is today's rule again. With no list, or in ``egress``, the body has no ``catalogue`` field and nothing here changed. Extracted from the endpoint closure so the decision is unit-testable; the bug lived in a closure nothing could reach. """ from ..observability.health import get_event_store_durability_status from .catalogue_readiness import catalogue_readiness ready_count = sum(1 for p in repository.get_all().values() if p.state.value == "ready") total_count = repository.count() durability = get_event_store_durability_status() event_store_ok = durability is None or not durability.degraded catalogue = catalogue_readiness(repository) catalogue_ok = catalogue is None or not catalogue["holds_readiness"] body: dict[str, Any] = { "status": "healthy" if event_store_ok and catalogue_ok else "unhealthy", "ready_mcp_servers": ready_count, "total_mcp_servers": total_count, } if not event_store_ok and durability is not None: body["event_store"] = { "status": "unhealthy", "configured_driver": durability.configured_driver, "durable": durability.durable, "detail": durability.detail, } if catalogue is not None: body["catalogue"] = catalogue return body, (200 if event_store_ok and catalogue_ok else 503) def warm_the_front_door_catalogue(runtime: Any) -> None: """Start every configured mcp_server so this replica can answer ``tools/list``. In ``front_door`` the flat projection **is** ``tools/list``, and the projection is built from ``McpServerStarted``. A replica that has started nothing has discovered nothing, so after every restart it serves an empty catalogue to a perfectly valid tenant -- and no client can change that. The meta-API is not projected for an ordinary tenant, so there is no ``hangar_warm`` to call; a tool name the client already knows resolves against the same empty map (``Tool 'add' not found``); and health checks skip cold servers by construction (``gc.py``, ``state_str in ("cold", "initializing")``). The only remedy was an operator, per replica, over the REST admin API (#878, #885). Two replicas that warmed different servers then answer the same tenant differently -- 18 tools from one, 0 from the other, alternating through the Service (#886). Every replica starting every configured server is what makes the catalogue a property of the configuration again instead of a readout of one replica's warm-up history. **Only in ``front_door``.** In ``egress`` the ``hangar_*`` meta-API is the surface, lazy start on first use is the documented behaviour that ``idle_ttl_s`` is designed around, and starting every backend at boot would change what every existing deployment costs to run. Through the command bus, not ``ensure_ready()``: the aggregate only *records* ``McpServerStarted``, and the command handler is what drains and publishes it. Called directly, this would start the fleet and leave the projection empty until the GC worker's next sweep happened to publish -- which is exactly how group members come to be projected today, up to 30s late and by accident. A backend that fails here stays unprojected: the state it would have been in anyway, logged per server, and reported by the empty-projection metric (#887). This warm-up is not retried. A server named in ``tool_access.required_catalogue`` is, by `catalogue_readiness.CatalogueRetry`, which runs after this on the same thread and within the bounds that module sets out (#1446); every other server that is down at boot stays down until a call starts it or the fleet is warmed again on the next restart. Args: runtime: The runtime holding the fleet and the command bus. """ from ..application.commands import StartMcpServerCommand from ..domain.services.tool_access_resolver import is_front_door from ..fastmcp_server import catalogue_warmup if not is_front_door(): return warmed = failed = skipped = 0 # A listing that arrives before this finishes would be answered with an # empty catalogue the client then caches forever (#1231). It waits instead, # briefly and only when the answer would otherwise be knowably wrong. catalogue_warmup.warmup_started() try: for mcp_server_id in runtime.repository.get_all_ids(): server = runtime.repository.get(mcp_server_id) if server is not None and server.state.value == "dead": # Restored DEAD from the event log. A warm-up of everything is not # a deliberate start of each server, and only a deliberate start # revives a dead one (#1361). skipped += 1 continue try: runtime.command_bus.send(StartMcpServerCommand(mcp_server_id=mcp_server_id)) warmed += 1 except Exception as e: # noqa: BLE001 -- fault-barrier: one dead backend must not cost the others their projection failed += 1 # The type only: a start failure's text can carry what the upstream printed. logger.warning( "front_door_warmup_failed", mcp_server_id=mcp_server_id, error_type=bounded_error_type(type(e).__qualname__), ) finally: # In `finally`: a warm-up that dies must not leave every later listing # waiting out the full deadline for something that will never finish. catalogue_warmup.warmup_finished() logger.info("front_door_warmup_complete", warmed=warmed, failed=failed, skipped_dead=skipped) def start_front_door_warm_up(runtime: Any, retry: CatalogueRetry) -> threading.Thread: """Warm the front door's catalogue, then run *retry* on the servers it missed, on a thread of its own. One thread for both: the retry starts where the warm-up ends, and each returns at once where it does not apply (``egress``, or no ``tool_access.required_catalogue``, #1446). On a thread because a backend handshake is I/O, and nothing may wait on it: `build_readiness_report` spells out why gating the serving path on a warm backend deadlocks the deployment. The front door serves a short list until this finishes, which is the bounded version of serving an empty one forever (#878, #885, #886). `ServerLifecycle.start` and the `Hangar` facade both start it here (#1465). Args: runtime: The runtime holding the fleet and the command bus. retry: The required-catalogue retry; its ``stop`` ends it. Returns: The started thread. """ def warm_up() -> None: warm_the_front_door_catalogue(runtime) retry.run() thread = threading.Thread(target=warm_up, name="mcp-hangar-front-door-warmup", daemon=True) thread.start() return thread def start_coordination() -> None: """Start the management lease keeper, then the event tailer, where bootstrap built them. `ServerLifecycle.start` and the `Hangar` facade both start them here, first, before the workers and discovery (#1465). Everything those run asks whether this instance holds the lease, and a keeper that has not started yet answers no. Starting it here rather than in bootstrap means a process that is assembled but never run never claims to be the manager. The tailer starts after the handlers are registered -- which bootstrap has done by now -- so a peer's event is not applied to an empty handler table. """ keeper = get_lease_keeper() if keeper is not None: keeper.start() tailer = get_event_tailer() if tailer is not None: tailer.start() def stop_coordination(shut_down: Callable[[], None]) -> None: """Stop the event tailer, run *shut_down*, then release the management lease. The tailer stops after the loops it feeds, and the lease is released last: releasing hands management to a peer in seconds rather than a TTL, so everything this instance was doing under it has to have stopped first. Released even when *shut_down* raises, so a failed shutdown does not cost a peer the wait for the TTL. `ServerLifecycle.shutdown` and the `Hangar` facade both stop them here (#1465). Args: shut_down: The application context's shutdown. """ tailer = get_event_tailer() if tailer is not None: tailer.stop() try: shut_down() finally: keeper = get_lease_keeper() if keeper is not None: keeper.stop() def mcp_app_for_serving(mcp_server: Any) -> Any: """Build the ASGI app ``serve --http`` mounts at ``/mcp``. A function rather than four lines inside ``run_http`` so a test can drive the app the CLI actually serves. Composition wired only where tests cannot reach it is how this codebase has repeatedly shipped a surface that was green in the suite and absent in production -- ``MCPServerFactory`` has no production call site, and four separate features were wired only there (#592, #594, #595, #596). A session defect in particular cannot be seen from one instance, so the seam is what makes the #877 test possible at all. ``transport_security`` is passed explicitly: left to the SDK's default the guard is built from its own bind host, so the endpoint answered 421 to the Service DNS name and every Ingress host while ``MCP_TRUSTED_HOSTS`` listed them (#859). ``stateless_http`` is the fix for #877. A handshake-era session lives in ONE replica's memory -- ``StreamableHTTPSessionManager._server_instances`` maps the id to a live transport running as a task in that process -- so N replicas of one coordinated gateway are N servers to a client, and the Service hands each request to whichever it likes. Measured: a client that initializes against one replica and lists tools against another is told ``Session not found``. The session buys this gateway nothing to weigh against that. It issues no server-to-client requests (no elicitation, no sampling, no roots) and no notifications; ``event_store`` is unset here, so there was never any resumability to lose; authorization is per-request at the route chokepoint rather than bound to a session; and session *suspension* keys on ``CallerIdentity.session_id``, which comes from ``x-session-id`` or the JWT ``sid`` claim and never from ``Mcp-Session-Id``. Handshake-era only, and deliberately so: SEP-2567 removed sessions, so a 2026-07-28 request is era-routed to the SDK's modern entry and never reaches the session table. This makes the older revisions behave the way the current one already does rather than inventing a mode. Accepted cost: ``DELETE /mcp`` answers 405, because there is no session to terminate. The SEP-2243 wrap checks a legacy-era POST's ``Mcp-Method`` / ``Mcp-Name`` against its body instead of trusting it. The 2026-07-28 era needs nothing there -- the SDK enforces header/body agreement itself -- but the legacy era does, and this path served neither before (#560). Args: mcp_server: The server ``build_serving_mcp_server()`` produced. Returns: The wrapped ASGI application. """ from ..fastmcp_server.asgi import mcp_transport_security from ..fastmcp_server.modern_surface import wrap_front_door_routing return wrap_front_door_routing( mcp_server.streamable_http_app( transport_security=mcp_transport_security(), stateless_http=True, ) ) def metrics_endpoint(request: Any) -> Any: """``GET /metrics``: the Prometheus exposition ``serve --http`` answers with. At module level so a test can scrape the gateway through the same endpoint the served process mounts (#1369). A test that read the registry directly could not tell whether the scrape carries the metric. """ from starlette.responses import PlainTextResponse from ..metrics import get_metrics return PlainTextResponse(get_metrics(), media_type="text/plain; version=0.0.4; charset=utf-8") def _is_loopback_host(host: str) -> bool: """Return whether a bind host resolves to loopback-only.""" normalized_host = host.strip().lower() if normalized_host in {"127.0.0.1", "::1", "localhost"}: return True try: return ipaddress.ip_address(normalized_host).is_loopback except ValueError: return False class ServerLifecycle: """Manages server start/stop lifecycle. This class coordinates the startup and shutdown of all server components including background workers, discovery orchestrator, and the MCP server. """ def __init__(self, context: ApplicationContext): """Initialize server lifecycle. Args: context: Fully initialized ApplicationContext from bootstrap. """ self._context = context self._running = False self._shutdown_requested = False self._discovery_loop: asyncio.AbstractEventLoop | None = None self._discovery_thread: threading.Thread | None = None self._catalogue_retry = CatalogueRetry(context.runtime) @property def is_running(self) -> bool: """Check if server is running.""" return self._running def start(self) -> None: """Start all background components. Starts: - Background workers (GC, health check) - Discovery orchestrator (if enabled) Does NOT start the MCP server - that's handled by run_stdio() or run_http(). """ if self._running: logger.warning("server_lifecycle_already_running") return self._running = True logger.info("server_lifecycle_start") # First: everything below asks whether this instance holds the lease. # The `Hangar` facade starts coordination, the workers and the warm-up # through the same functions (#1435, #1465). start_coordination() start_background_workers(self._context.background_workers) self._start_discovery() # Last, and on a thread of its own. start_front_door_warm_up(self._context.runtime, self._catalogue_retry) def _start_discovery(self) -> None: """Start discovery on a dedicated long-lived event loop.""" orchestrator = self._context.discovery_orchestrator if orchestrator is None: return self._discovery_loop, self._discovery_thread = start_discovery_loop(orchestrator) def run_stdio(self) -> None: """Run MCP server in stdio mode. Blocks until exit. This is the standard mode for Claude Desktop, Cursor, and other MCP clients that communicate via stdin/stdout. """ logger.info("starting_stdio_server") try: self._context.mcp_server.run() except KeyboardInterrupt: logger.info("stdio_server_shutdown", reason="keyboard_interrupt") except Exception as e: # noqa: BLE001 -- fault-barrier: fatal server error boundary logger.critical( "fatal_server_error", error=str(e), error_type=type(e).__name__, ) sys.exit(1) def run_http(self, host: str, port: int, unsafe_no_auth: bool = False) -> None: # noqa: C901 -- baseline CC=19; split before extending """Run MCP server in HTTP mode. Blocks until exit. This mode is compatible with LM Studio and other MCP HTTP clients. Endpoints: - /mcp: Streamable HTTP MCP endpoint (POST/GET) Args: host: Host to bind to. port: Port to bind to. """ import uvicorn auth_components = self._context.auth_components auth_enabled = bool(auth_components and auth_components.enabled) if not auth_enabled and not _is_loopback_host(host): message = "Refusing to start HTTP on non-loopback without authentication. Use --unsafe-no-auth to override." if not unsafe_no_auth: logger.error("http_auth_required_for_non_loopback", host=host, port=port, message=message) raise SystemExit(1) logger.warning( "http_auth_disabled_non_loopback_override", host=host, port=port, message=message, ) # Checked at bootstrap, so this cannot raise for a configuration that # booted. None is uvicorn's own default: in-flight requests are waited # for without a bound, and it is logged as null so the bound in force # is visible either way (#1447). graceful_shutdown_timeout_s = http_graceful_shutdown_timeout(self._context.config) logger.info( "starting_http_server", host=host, port=port, graceful_shutdown_timeout_s=graceful_shutdown_timeout_s, ) # Update FastMCP settings for HTTP mode. FastMCP (SDK v1) carries # host/port on .settings; MCPServer (SDK v2) exposes a Settings object # that has no host/port fields. Either way this is vestigial for our # serving path -- the host uvicorn below binds host/port directly -- so # set it only where the fields actually exist. mcp_server = self._context.mcp_server settings = getattr(mcp_server, "settings", None) if settings is not None and hasattr(settings, "host"): settings.host = host settings.port = port mcp_app = mcp_app_for_serving(mcp_server) # Create auxiliary routes for /metrics, /health, /ready import time from starlette.applications import Starlette from starlette.responses import JSONResponse from starlette.routing import Route from .bootstrap.composition import get_runtime _start_time = time.time() _startup_complete = False def liveness_endpoint(request): """Liveness check - is the process alive?""" return JSONResponse({"status": "healthy"}) def readiness_endpoint(request): """Readiness check - can we handle traffic?""" body, status_code = build_readiness_report(get_runtime().repository) return JSONResponse(body, status_code=status_code) def startup_endpoint(request): """Startup check - has initialization completed?""" nonlocal _startup_complete # Mark startup complete after first check (bootstrap is done by this point) _startup_complete = True uptime = time.time() - _start_time return JSONResponse( { "status": "healthy", "startup_complete": _startup_complete, "uptime_seconds": round(uptime, 2), } ) routes = [ Route("/health/live", liveness_endpoint, methods=["GET"]), Route("/health/ready", readiness_endpoint, methods=["GET"]), Route("/health/startup", startup_endpoint, methods=["GET"]), Route("/metrics", metrics_endpoint, methods=["GET"]), ] # Register RFC 9728 Protected Resource Metadata endpoint (unauthenticated discovery). # The endpoint is placed on the aux app (outside auth enforcement) so clients can # reach it before obtaining a token. When OIDC is not configured / no issuer is # set, we return 404 — there is nothing to advertise. _oidc_issuers: list[str] = ( auth_components.oidc_issuers if auth_components and hasattr(auth_components, "oidc_issuers") else [] ) _oidc_resource_uri_cfg = ( auth_components.oidc_resource_uri if auth_components and hasattr(auth_components, "oidc_resource_uri") else "" ) from ..auth.prm import build_prm_response, build_resource_base_url def prm_endpoint(request): """RFC 9728 Protected Resource Metadata (unauthenticated discovery). Returns 404 when no OIDC issuer is configured — nothing to advertise. """ if not _oidc_issuers: return JSONResponse( {"error": "not_found", "message": "No OIDC issuer configured"}, status_code=404, ) resource_base = _oidc_resource_uri_cfg or build_resource_base_url(request.scope) return JSONResponse( build_prm_response(issuers=_oidc_issuers, resource_uri=resource_base), media_type="application/json", ) routes.append(Route("/.well-known/oauth-protected-resource", prm_endpoint, methods=["GET"])) # Create REST API router for /api/* endpoints # Stdio mode: pass auth_components from context for consistency. # Auth enforcement on REST API applies even in stdio mode when auth is configured. from ..server.api import create_api_router api_app = create_api_router(auth_components=getattr(self._context, "auth_components", None)) # create_api_router already wired the component services from the # application context. Overlay this lifecycle's own ApplicationContext # when it carries one, so the two objects cannot disagree: this used to # be the ONLY place app.state.approval_gate_service was set, and it read # a field that was never populated, so /api/approvals 500'd (#678). approval_svc = getattr(self._context, "approval_service", None) if approval_svc is not None: api_app.state.approval_gate_service = approval_svc # Mount health/metrics and REST API together in one Starlette app from starlette.routing import Mount all_routes = routes + [Mount("/api", app=api_app)] aux_app = Starlette(routes=all_routes) _PRM_PATH = "/.well-known/oauth-protected-resource" async def combined_app(scope, receive, send): """Combined ASGI app that routes to aux (health/metrics/api) or MCP.""" if scope["type"] in ("http", "websocket"): path = scope.get("path", "") if ( path.startswith("/health/") or path == "/metrics" or path == _PRM_PATH or path == "/api" or path.startswith("/api/") ): await aux_app(scope, receive, send) return await mcp_app(scope, receive, send) # Apply authentication middleware if enabled if auth_components and auth_components.enabled: starlette_app = self._create_auth_app(combined_app, auth_components) logger.info("http_auth_enabled") else: starlette_app = combined_app # CORS goes OUTSIDE auth, on the app that actually serves traffic. # create_api_router documents CORS as outermost, but only on the # mounted api_app -- the served process wrapped the combined app # (health + /api + /mcp) with auth directly, so a browser preflight # 401'd before any CORS layer could answer, and /mcp never had CORS # headers at all (#993). Wrapping here covers both, for allowed and # refused origins alike; the inner copy on api_app is redundant but # harmless (same config, headers are set, not appended). from starlette.middleware.cors import CORSMiddleware from .api.middleware import get_cors_config starlette_app = CORSMiddleware(app=starlette_app, **get_cors_config()) # Configure uvicorn with log_config=None to disable default uvicorn logging # Our structlog configuration will handle all logging uniformly config = uvicorn.Config( starlette_app, host=host, port=port, log_config=None, # Disable uvicorn's default logging access_log=False, # Disable access logs (we'll handle them via structlog if needed) # Off, so the peer the app sees is the peer that connected. # uvicorn's default rewrote a loopback peer (or any in # FORWARDED_ALLOW_IPS) to its X-Forwarded-For address before Hangar # saw the request -- so a loopback proxy never looked like a proxy # and its x-session-id was ignored (GHSA-fhwh-fmq2-7m5c), and the # forwarded-address trust lived in a variable Hangar never read. # `MCP_TRUSTED_PROXIES` (`TrustedProxyResolver`) is now the one # decision, applied by the auth middleware and the identity bridge. proxy_headers=False, # `http.graceful_shutdown_timeout_s`: how long a stop waits for the # requests in flight before cancelling them. Unset passes None, # which is uvicorn's own default, so nothing changes (#1447). timeout_graceful_shutdown=graceful_shutdown_timeout_s, ) async def run_server(): server = uvicorn.Server(config) logger.info("http_server_started", host=host, port=port, endpoint="/mcp") try: await server.serve() finally: self.shutdown() logger.info("http_server_stopped") try: asyncio.run(run_server()) except KeyboardInterrupt: logger.info("http_server_shutdown", reason="keyboard_interrupt") except asyncio.CancelledError: logger.info("http_server_shutdown", reason="cancelled") except Exception as e: # noqa: BLE001 -- fault-barrier: fatal server error boundary logger.critical( "fatal_server_error", error=str(e), error_type=type(e).__name__, ) sys.exit(1) def shutdown(self) -> None: """Graceful shutdown of all components. Stops: - Runtime (hot-loaded) mcp_servers - Background workers - Discovery orchestrator - All configured mcp_servers This method is safe to call multiple times. """ if self._shutdown_requested: logger.debug("shutdown_already_requested") return self._shutdown_requested = True logger.info("server_lifecycle_shutdown_start") # Before anything below stops a server, for the reason the scheduled # commands are cancelled next: the catalogue retry would start it again. self._catalogue_retry.stop() # First: a retry a saga scheduled would otherwise fire while the servers # below are being stopped, and start one again (#1389). The context # cancels once more after its workers stop, for any armed in between. self._context.cancel_scheduled_commands() self._cleanup_runtime_mcp_servers() self._stop_discovery() # After the loops it gates, and before the process ends: releasing the # lease is what turns "a peer takes over in a TTL" into "a peer takes # over in seconds". Releasing it while discovery was still winding down # would let a peer start converging against a fleet this instance is # still touching. stop_coordination(self._context.shutdown) self._running = False logger.info("server_lifecycle_shutdown_complete") def _stop_discovery(self) -> None: """Await discovery cleanup, then stop and join its dedicated loop.""" orchestrator = self._context.discovery_orchestrator loop = self._discovery_loop thread = self._discovery_thread if orchestrator is None or loop is None or thread is None: return try: stop_discovery_loop(orchestrator, loop, thread) finally: self._discovery_loop = None self._discovery_thread = None def _cleanup_runtime_mcp_servers(self) -> None: """Cleanup all hot-loaded runtime mcp_servers.""" runtime_store = get_runtime_mcp_servers() if runtime_store.count() == 0: return logger.info( "cleaning_up_runtime_mcp_servers", count=runtime_store.count(), ) for mcp_server, metadata in runtime_store.list_all(): try: mcp_server.shutdown() except Exception as e: # noqa: BLE001 -- fault-barrier: one mcp_server shutdown failure must not prevent others logger.warning( "runtime_mcp_server_shutdown_error", mcp_server_id=str(mcp_server.mcp_server_id), error=str(e), ) if metadata.cleanup: try: metadata.cleanup() except Exception as e: # noqa: BLE001 -- fault-barrier: cleanup callback failure must not prevent other cleanups logger.warning( "runtime_mcp_server_cleanup_error", mcp_server_id=str(mcp_server.mcp_server_id), error=str(e), ) runtime_store.clear() logger.info("runtime_mcp_servers_cleaned_up") def _create_auth_app(self, inner_app, auth_components): """Create auth-enabled ASGI app wrapper. Args: inner_app: The inner ASGI app to wrap. auth_components: Auth components with middleware. Returns: ASGI app with authentication. """ return create_auth_enforced_app(inner_app, auth_components) def _setup_signal_handlers(lifecycle: ServerLifecycle) -> None: """Setup graceful shutdown on SIGTERM/SIGINT and reload on SIGHUP. Args: lifecycle: ServerLifecycle instance to shutdown on signal. """ def shutdown_handler(signum, _frame): sig_name = signal.Signals(signum).name logger.info("shutdown_signal_received", signal=sig_name) lifecycle.shutdown() sys.exit(0) def reload_handler(signum, _frame): """Handle SIGHUP for configuration reload.""" sig_name = signal.Signals(signum).name logger.info("reload_signal_received", signal=sig_name) try: from ..application.commands.commands import ReloadConfigurationCommand command = ReloadConfigurationCommand( graceful=True, requested_by="sighup", ) # Access command bus from lifecycle context result = lifecycle._context.runtime.command_bus.send(command) logger.info("config_reload_completed_via_signal", result=result) except Exception as e: # noqa: BLE001 -- fault-barrier: signal handler must not crash process logger.error( "config_reload_failed_via_signal", error=str(e), error_type=type(e).__name__, ) signal.signal(signal.SIGTERM, shutdown_handler) signal.signal(signal.SIGINT, shutdown_handler) # SIGHUP is not available on Windows if hasattr(signal, "SIGHUP"): signal.signal(signal.SIGHUP, reload_handler) logger.debug("sighup_handler_registered") @dataclass(frozen=True) class LoggingSettings: """What `setup_logging` is called with once every source has been read.""" level: str json_format: bool log_file: str | None def resolve_logging_settings( config_path: str | None, *, log_level: str = "INFO", log_file: str | None = None, json_logs: bool = False, ) -> LoggingSettings: """Resolve the logging settings from the command line and the config file. `serve` and `pin` both resolve through here, so one command cannot honour a control the other ignores (#1236). The order, highest first: 1. ``--log-level`` / ``--log-file`` / ``--json-logs``; 2. ``MCP_LOG_LEVEL`` / ``MCP_JSON_LOGS``. Typer folds these into the flag's value, so they arrive here as the flag and outrank the config file; 3. the config file's ``logging`` section (``level``, ``file``, ``json_format``); 4. the defaults: INFO, console format, no file. A level of ``INFO`` from layer 1 or 2 cannot be told apart from the default, so the config file's ``logging.level`` overrides it. Args: config_path: The config file whose ``logging`` section is read, if any. log_level: The level from the flag or its environment variable. log_file: The log file from the flag, if any. json_logs: Whether the flag or its environment variable asked for JSON. """ log_level = log_level.upper() level = log_level file = log_file json_format = json_logs if config_path and Path(config_path).exists(): try: full_config = load_config_from_file(config_path) logging_config = full_config.get("logging", {}) # Config file values are used only if CLI didn't specify if log_level == "INFO": # Default value level = logging_config.get("level", level).upper() if not log_file: file = logging_config.get("file", file) if not json_logs: json_format = logging_config.get("json_format", json_format) except (FileNotFoundError, yaml.YAMLError, ValueError, OSError) as e: # Config loading failed - use CLI values, log will be set up shortly logger.debug("config_preload_failed", error=str(e)) return LoggingSettings(level=level, json_format=json_format, log_file=file) def _setup_logging_from_config(cli_config: CLIConfig) -> None: """Set up logging for `serve` from its CLI config and the config file. The order of precedence is `resolve_logging_settings`'s. Args: cli_config: Parsed CLI configuration. """ settings = resolve_logging_settings( cli_config.config_path, log_level=cli_config.log_level, log_file=cli_config.log_file, json_logs=cli_config.json_logs, ) setup_logging(level=settings.level, json_format=settings.json_format, log_file=settings.log_file) def run_server(cli_config: CLIConfig) -> None: """Main entry point that ties everything together. This function orchestrates: 1. Setup logging based on CLI config 2. Bootstrap application 3. Setup signal handlers 4. Start lifecycle (background workers, discovery) 5. Run appropriate server mode 6. Handle shutdown on exit/signal Args: cli_config: Parsed CLI configuration from parse_args(). """ # Setup logging first _setup_logging_from_config(cli_config) mode_str = "http" if cli_config.http_mode else "stdio" logger.info( "mcp_registry_starting", mode=mode_str, log_file=cli_config.log_file, ) # Bootstrap application. The transport is passed because `auth.stdio.principal` # is read only when this process serves over stdio (ADR-026). context = bootstrap(cli_config.config_path, stdio=not cli_config.http_mode) # Create lifecycle manager lifecycle = ServerLifecycle(context) _setup_signal_handlers(lifecycle) # Start background components and the dedicated discovery lifecycle loop. lifecycle.start() # Log ready state mcp_server_ids = list(context.runtime.repository.get_all_ids()) orchestrator = get_discovery_orchestrator() discovery_status = "enabled" if orchestrator else "disabled" logger.info( "mcp_registry_ready", mcp_servers=mcp_server_ids, discovery=discovery_status, ) # Run server in appropriate mode try: if cli_config.http_mode: lifecycle.run_http(cli_config.http_host, cli_config.http_port, unsafe_no_auth=cli_config.unsafe_no_auth) else: lifecycle.run_stdio() finally: # Ensure cleanup on exit lifecycle.shutdown() __all__ = [ "ServerLifecycle", "build_readiness_report", "mcp_app_for_serving", "run_server", ]