"""Steward's MCP server: the StewardMCPServer instance and its mount at /mcp. The /mcp path is dispatched to the MCP SDK's streamable-HTTP ASGI app BEFORE Quart sees it (wrapping `app.asgi_app`), so Quart's session, before_request hooks and error pages do not apply. This module supplies the equivalents: - bearer auth: a personal API token resolves to a Caller (user + effective role), set in a ContextVar for the tools — 401 otherwise; - the database gate: the same ride-out-a-restart probe the web pipeline uses, answering 503 rather than failing deep in a tool; - role gating, default-deny: every tool is registered with a minimum role and a tool the caller's role doesn't reach is neither listed nor callable. """ from __future__ import annotations import difflib import functools import json import logging from mcp.server.mcpserver import MCPServer from mcp.server.mcpserver.exceptions import ToolError, UnexpectedToolError from mcp.server.transport_security import TransportSecuritySettings from mcp.types import ToolAnnotations from sqlalchemy.exc import SQLAlchemyError from steward.auth.middleware import role_meets from steward.models.users import UserRole from . import _context logger = logging.getLogger(__name__) MCP_PATH = "/mcp" # Claude Code injects only the first ~2,048 characters of a server's # instructions; keep this well under (test_instructions_fit). The workflow # across tools lives here; each tool's own contract lives in its docstring. INSTRUCTIONS = """\ Steward is a self-hosted monitoring and infrastructure hub: hosts, uptime monitors, host-agent metrics, Docker containers and logs, alerts, and Ansible automation. You act as the user who created your token, at that token's role. Start broad, then narrow: get_overview says what is wrong right now; follow its ids into get_host / get_monitor / get_container, then metrics and logs. List tools are bounded: a response with `truncated: true` says which argument fetches more. Times are UTC ISO-8601; units are in field names (_pct, _ms, _bytes). Changes are audited under the user's name. run_playbook is a dry run (Ansible check mode) unless you pass check=false; dry-run first and show the user the diff before applying. If a tool refuses for lack of role, whoami says what this token can do. """ _DB_UNAVAILABLE = ( "Steward is running but can't reach its database right now, so nothing was " "read or changed. This is usually a database restart; retry in a minute." ) def _is_db_failure(exc: BaseException | None) -> bool: seen: set[int] = set() while exc is not None and id(exc) not in seen: seen.add(id(exc)) if isinstance(exc, (SQLAlchemyError, OSError, TimeoutError)): return True exc = exc.__cause__ or exc.__context__ return False class StewardMCPServer(MCPServer): """MCPServer with per-tool minimum roles, plugin-scoped tools and strict args.""" def __init__(self, app, **kwargs): super().__init__("steward", instructions=INSTRUCTIONS.strip(), **kwargs) self._steward_app = app self.tool_roles: dict[str, UserRole] = {} self.tool_plugins: dict[str, str] = {} # plugin name -> its `mcp_overview(db)`: findings it adds to get_overview. self.overview_sources: dict[str, object] = {} # plugin name -> its `mcp_events(db, host_id, host_name, since, until)`: # what it knows happened on a host, for get_anomaly. self.event_sources: dict[str, object] = {} # plugin name -> its `mcp_resource_events(db, source, resource, since, # until)`: what it knows happened to a resource with no host. self.resource_event_sources: dict[str, object] = {} # ── registration ───────────────────────────────────────────────────────── def register_tool(self, fn, *, role: UserRole, read_only: bool, plugin: str | None = None) -> None: """The only way a tool reaches the server: name, minimum role, effect. `role` and `read_only` are required — there is no default for a mutating tool to inherit by omission. """ name = fn.__name__ if name in self.tool_roles: raise ValueError(f"MCP tool {name!r} is already registered") if not isinstance(role, UserRole): raise TypeError(f"MCP tool {name!r}: role must be a UserRole, got {role!r}") self.add_tool( fn, name=name, annotations=ToolAnnotations(read_only_hint=read_only, destructive_hint=None if read_only else True), ) self.tool_roles[name] = role if plugin: self.tool_plugins[name] = plugin def register_plugin(self, name: str, module) -> None: """Add a plugin's tools via its optional `register_mcp_tools(register)`, its optional `mcp_overview(db)` as a get_overview findings source, and its optional `mcp_events(db, host_id, host_name, since, until)` as a source of what happened on a host (get_anomaly), and its optional `mcp_resource_events(db, source, resource, since, until)` for what happened to a resource with no host. Idempotent per plugin (hot-reload drops its previous tools first) and tolerant: a raising hook is logged and leaves the plugin with no tools, never a broken server. """ self._drop_plugin_tools(name) self.overview_sources.pop(name, None) overview = getattr(module, "mcp_overview", None) if callable(overview): self.overview_sources[name] = overview self.event_sources.pop(name, None) events = getattr(module, "mcp_events", None) if callable(events): self.event_sources[name] = events self.resource_event_sources.pop(name, None) resource_events = getattr(module, "mcp_resource_events", None) if callable(resource_events): self.resource_event_sources[name] = resource_events hook = getattr(module, "register_mcp_tools", None) if hook is None: return try: hook(functools.partial(self.register_tool, plugin=name)) except Exception: # All or nothing: a hook that fails halfway leaves no partial set. self._drop_plugin_tools(name) logger.exception("Plugin %r: register_mcp_tools() raised; its MCP tools are unavailable", name) def _drop_plugin_tools(self, name: str) -> None: for tool_name in [t for t, p in self.tool_plugins.items() if p == name]: self.remove_tool(tool_name) self.tool_roles.pop(tool_name, None) self.tool_plugins.pop(tool_name, None) # ── gating ─────────────────────────────────────────────────────────────── def plugin_enabled(self, plugin: str) -> bool: cfg = (self._steward_app.config.get("PLUGINS") or {}).get(plugin) return isinstance(cfg, dict) and bool(cfg.get("enabled")) def refusal(self, name: str, caller) -> str | None: """Why `caller` may not use tool `name`, or None if they may.""" required = self.tool_roles.get(name) if required is None: return f"There is no Steward tool named {name!r}." plugin = self.tool_plugins.get(name) if plugin and not self.plugin_enabled(plugin): return (f"{name} belongs to the {plugin} plugin, which is disabled in this " "Steward. An admin can enable it under Settings → Plugins.") if caller is None: return "This request is not authenticated." if not role_meets(caller.role, required): return (f"{name} needs the {required.value} role; this connection acts as " f"{caller.role.value}. A token's role is capped at its user's role — " "create a token with a higher role on Steward's AI access page if " "your account allows it.") return None async def list_tools(self): caller = _context.current_caller_or_none() tools = await super().list_tools() return [t for t in tools if self.refusal(t.name, caller) is None] async def call_tool(self, name, arguments, context=None): refusal = self.refusal(name, _context.current_caller_or_none()) if refusal: raise ToolError(refusal) # Reject undeclared arguments: pydantic would silently drop a misnamed # one and run the tool with that field's default. tool = self._tool_manager.get_tool(name) declared = set((tool.parameters or {}).get("properties", {})) if tool else set() unknown = sorted(set(arguments or {}) - declared) if unknown: hints = [] for arg in unknown: close = difflib.get_close_matches(arg, sorted(declared), n=1) hints.append(f"'{arg}'" + (f" (did you mean '{close[0]}'?)" if close else "")) raise ToolError( f"{name} does not accept {', '.join(hints)}. It accepts: " f"{', '.join(sorted(declared)) or 'no arguments'}. Nothing was run." ) try: return await super().call_tool(name, arguments, context) except UnexpectedToolError as exc: cause = exc.__cause__ # ValueError is how a Steward tool refuses: what, why, what instead. if isinstance(cause, ValueError): raise ToolError(f"{name}: {cause}") from cause if _is_db_failure(cause): logger.warning("MCP tool %s hit a database failure: %r", name, cause) raise ToolError(_DB_UNAVAILABLE) from cause raise def build_mcp_server(app) -> StewardMCPServer: from .tools import register_all mcp = StewardMCPServer(app) register_all(mcp.register_tool) return mcp async def _send_json(send, status: int, payload: dict, extra_headers=()) -> None: await send({ "type": "http.response.start", "status": status, "headers": [(b"content-type", b"application/json"), *extra_headers], }) await send({"type": "http.response.body", "body": json.dumps(payload).encode()}) def mount_mcp(app) -> StewardMCPServer: """Mount the MCP endpoint at /mcp and return the server. stateless_http: every request stands alone on its bearer token. The stateful default strands Claude Code after a redeploy — it reconnects with a session id the new process never issued and won't re-initialize on the 404 — so stateless is what makes a restart invisible to clients. DNS-rebinding protection is off: its default allow-list is localhost only, so any deployment behind a reverse proxy would get 421s. That protection guards a browser rebinding DNS onto a localhost server; here the bearer token is the security boundary. """ # Looked up through their modules per request (not bound here) so a test # can stand in for the token store or the database probe. from steward import database from steward.core import api_tokens from steward.core.plugin_manager import get_loaded_plugin_modules mcp = build_mcp_server(app) for name, module in sorted(get_loaded_plugin_modules().items()): mcp.register_plugin(name, module) mcp_asgi = mcp.streamable_http_app( streamable_http_path=MCP_PATH, stateless_http=True, transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False), ) app.mcp_server = mcp # The SDK's session manager owns a task group that must be running before # it serves; host it in Quart's serving lifecycle. @app.before_serving async def _start_mcp() -> None: cm = mcp.session_manager.run() await cm.__aenter__() app._mcp_session_cm = cm @app.after_serving async def _stop_mcp() -> None: cm = getattr(app, "_mcp_session_cm", None) if cm is not None: await cm.__aexit__(None, None, None) async def handle_mcp(scope, receive, send): headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope.get("headers", [])} auth = headers.get("authorization", "") raw = auth[7:].strip() if auth[:7].lower() == "bearer " else "" engine = getattr(app, "_db_engine", None) try: if engine is not None: await database.ensure_database_reachable(engine) caller = await api_tokens.resolve_token(app, raw) except (database.DatabaseUnavailable, SQLAlchemyError, OSError) as exc: logger.error("MCP request refused, database unreachable: %r", exc) await _send_json(send, 503, {"error": "database_unavailable", "message": _DB_UNAVAILABLE}) return if caller is None: await _send_json( send, 401, {"error": "unauthorized", "message": "Send a Steward API token as 'Authorization: Bearer stw_…'. " "Create one on Steward's AI access page."}, extra_headers=[(b"www-authenticate", b'Bearer realm="steward-mcp"')], ) return caller_token = _context._caller_ctx.set(caller) app_token = _context._app_ctx.set(app) host = headers.get("host", "") url_token = _context._base_url_ctx.set(f"{scope.get('scheme', 'http')}://{host}" if host else "") try: await mcp_asgi(scope, receive, send) finally: _context._caller_ctx.reset(caller_token) _context._app_ctx.reset(app_token) _context._base_url_ctx.reset(url_token) original_asgi = app.asgi_app async def dispatch(scope, receive, send): if scope["type"] == "http": path = scope.get("path", "") if path == MCP_PATH or path.startswith(MCP_PATH + "/"): # Path passes through untouched: the SDK app routes on /mcp itself. return await handle_mcp(scope, receive, send) return await original_asgi(scope, receive, send) app.asgi_app = dispatch return mcp