bvandeusenandClaude Opus 5.5
1d9ac2b46a
#5015 "Built-in host-agent playbooks make an MCP caller supply steward_url" - run_playbook, create_schedule and set_alert_action fill a steward-builtin playbook's steward_url when the caller leaves it out or empty, from Settings → General → Public base URL, else the scheme and host the MCP request arrived on — the same source the web actions use. - run_playbook's reply names what it filled (filled_by_steward), and get_playbook shows that value as the variable's default. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
303 lines
13 KiB
Python
303 lines
13 KiB
Python
"""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] = {}
|
|
|
|
# ── 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)`,
|
|
and its optional `mcp_overview(db)` as a get_overview findings source.
|
|
|
|
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
|
|
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
|