diff --git a/docs/plugins/writing-a-plugin.md b/docs/plugins/writing-a-plugin.md index 4128f08..7e93f1f 100644 --- a/docs/plugins/writing-a-plugin.md +++ b/docs/plugins/writing-a-plugin.md @@ -323,6 +323,42 @@ Role hierarchy: `admin > operator > viewer`. Requiring `viewer` grants access to --- +## Agent Tools (MCP) + +Steward serves an MCP endpoint at `/mcp` that AI clients reach with a personal API token created on the AI access page. A plugin adds tools to it by exporting an optional `register_mcp_tools(register)` from `__init__.py`: + +```python +# plugins/myplugin/__init__.py +def register_mcp_tools(register) -> None: + from .mcp_tools import list_targets, rescan_target + from steward.models.users import UserRole + register(list_targets, role=UserRole.viewer, read_only=True) + register(rescan_target, role=UserRole.operator, read_only=False) +``` + +```python +# plugins/myplugin/mcp_tools.py +from steward.mcp._context import current_caller, mcp_app + + +async def list_targets(limit: int = 50) -> dict: + """What this tool is for, when to use it over its siblings, what each + argument's empty value means, and what comes back. This docstring is the + tool's description: it is all the agent knows about it.""" + app = mcp_app() # Quart's current_app does not exist here + async with app.db_sessionmaker() as db: + ... + return {"targets": [...], "total": n, "truncated": n > limit} +``` + +- `role` and `read_only` are required. A tool is listed and callable only for callers whose role reaches `role`, and only while the plugin is enabled. +- Tools run outside Quart's request pipeline: use `mcp_app()` and `current_caller()` from `steward.mcp._context`, never `current_app` or `session`. +- Refuse by raising `ValueError("what was wrong, and what to do instead")`. The agent reads that text. Any other exception reaches it only as a generic failure. +- Tool names are global across core and every plugin; a name already taken fails registration and the plugin contributes no tools (logged). +- Hot-reloading the plugin replaces its tools. + +--- + ## Publishing to the Catalog The official plugin catalog is hosted at `https://git.fabledsword.com/bvandeusen/Steward-plugins`. Anyone can submit a plugin by opening a pull request — first-party and third-party plugins are treated identically by the catalog system. @@ -413,6 +449,7 @@ Hot-reload works reliably for brand-new plugin installs. Updates to already-acti - [ ] Revision IDs prefixed with plugin name - [ ] `record_metric()` called inside `session.begin()` - [ ] Routes use `@require_role` decorator +- [ ] MCP tools (if any) registered via `register_mcp_tools(register)` with a role and `read_only` - [ ] Templates namespaced under `templates/myplugin/` - [ ] Plugin enabled in app settings - [ ] `index.yaml` entry added with `download_url` and `checksum_sha256` diff --git a/pyproject.toml b/pyproject.toml index c10b66a..de5781d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,9 @@ dependencies = [ "click>=8.0", "httpx>=0.27", "cryptography>=42", + # Floor, not a cap: MCPServer (mcp.server.mcpserver) and the mcp.types + # mirror that steward/mcp/ is written against first ship in SDK 2.2. + "mcp>=2.2", ] [project.optional-dependencies] diff --git a/steward/app.py b/steward/app.py index e66e696..9b302d4 100644 --- a/steward/app.py +++ b/steward/app.py @@ -282,6 +282,12 @@ def create_app( async def server_error(_): return await render_template("errors/500.html"), 500 + # ── MCP endpoint (/mcp) ─────────────────────────────────────────────────── + # Mounted after plugins load so their register_mcp_tools() hooks are seen. + # It wraps app.asgi_app: /mcp never enters Quart's pipeline (see mcp/server.py). + from .mcp import mount_mcp + mount_mcp(app) + # ── 11. Startup hook ─────────────────────────────────────────────────────── @app.before_serving async def startup(): diff --git a/steward/core/plugin_manager.py b/steward/core/plugin_manager.py index b78d8e9..3f68510 100644 --- a/steward/core/plugin_manager.py +++ b/steward/core/plugin_manager.py @@ -29,6 +29,11 @@ _FAILED_PLUGINS: dict[str, str] = {} # {"plugin": , "label": str, "href": str, "section": str} _PLUGIN_NAV: list[dict] = [] +# Loaded plugin modules, by name, for the MCP server: a plugin contributes +# agent tools through an optional register_mcp_tools(register) export, which +# steward/mcp/server.py calls when it mounts (and on hot-reload, below). +_PLUGIN_MODULES: dict[str, object] = {} + def get_plugin_failures() -> dict[str, str]: """Return a copy of the failed-plugin registry (name → error message).""" @@ -70,6 +75,23 @@ def _collect_plugin_nav(name: str, module) -> None: }) +def get_loaded_plugin_modules() -> dict[str, object]: + """Loaded plugin modules by name (a copy), for hooks consumed after loading.""" + return dict(_PLUGIN_MODULES) + + +def _collect_plugin_module(app: "Quart", name: str, module) -> None: + """Remember a loaded plugin module; hand it to a live MCP server if mounted. + + At boot the MCP server mounts after plugins load and reads them via + get_loaded_plugin_modules(); a plugin hot-loaded later is registered here. + """ + _PLUGIN_MODULES[name] = module + mcp_server = getattr(app, "mcp_server", None) + if mcp_server is not None: + mcp_server.register_plugin(name, module) + + def resolve_plugin_path(plugin_dirs: list[Path], name: str) -> Path | None: """Return the first plugin root that contains `name`, else None. @@ -264,6 +286,7 @@ def load_plugins(app: "Quart") -> None: _FAILED_PLUGINS.pop(name, None) # clear any stale failure from a previous reload attempt _LOADED_PLUGINS.add(name) _collect_plugin_nav(name, module) + _collect_plugin_module(app, name, module) logger.info("Plugin %r loaded (v%s)", name, meta.get("version", "?")) @@ -483,6 +506,7 @@ def hot_reload_plugin(app: "Quart", name: str) -> tuple[bool, str]: _LOADED_PLUGINS.add(name) _collect_plugin_nav(name, module) + _collect_plugin_module(app, name, module) logger.info("Plugin %r hot-reloaded (v%s)", name, meta.get("version", "?")) return True, f"Plugin activated (v{meta.get('version', '?')})" diff --git a/steward/mcp/__init__.py b/steward/mcp/__init__.py new file mode 100644 index 0000000..2a290eb --- /dev/null +++ b/steward/mcp/__init__.py @@ -0,0 +1,4 @@ +"""The MCP endpoint (/mcp): Steward as a server an LLM client can inspect and drive.""" +from .server import mount_mcp + +__all__ = ["mount_mcp"] diff --git a/steward/mcp/_context.py b/steward/mcp/_context.py new file mode 100644 index 0000000..60a78d3 --- /dev/null +++ b/steward/mcp/_context.py @@ -0,0 +1,38 @@ +"""Per-request MCP context: who is calling, and which app they are calling. + +The /mcp transport bypasses Quart's request pipeline, so neither `session` +nor `current_app` exists inside a tool. The ASGI auth wrapper in server.py +sets these from the resolved bearer token before dispatching; tools read them +through `current_caller()` and `mcp_app()`. + +Neither falls back to a default: a tool reached without an authenticated +caller is a bug in the transport, and it raises rather than acting for nobody. +""" +from __future__ import annotations +from contextvars import ContextVar +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from quart import Quart + from steward.core.api_tokens import Caller + +_caller_ctx: ContextVar["Caller | None"] = ContextVar("steward_mcp_caller", default=None) +_app_ctx: ContextVar["Quart | None"] = ContextVar("steward_mcp_app", default=None) + + +def current_caller() -> "Caller": + caller = _caller_ctx.get() + if caller is None: + raise RuntimeError("no MCP caller in context: the request was not bearer-authenticated") + return caller + + +def current_caller_or_none() -> "Caller | None": + return _caller_ctx.get() + + +def mcp_app() -> "Quart": + app = _app_ctx.get() + if app is None: + raise RuntimeError("no Steward app in MCP context") + return app diff --git a/steward/mcp/server.py b/steward/mcp/server.py new file mode 100644 index 0000000..9c8c2b4 --- /dev/null +++ b/steward/mcp/server.py @@ -0,0 +1,292 @@ +"""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] = {} + + # ── 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)`. + + 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) + 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) + try: + await mcp_asgi(scope, receive, send) + finally: + _context._caller_ctx.reset(caller_token) + _context._app_ctx.reset(app_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 diff --git a/steward/mcp/tools/__init__.py b/steward/mcp/tools/__init__.py new file mode 100644 index 0000000..8f8005b --- /dev/null +++ b/steward/mcp/tools/__init__.py @@ -0,0 +1,22 @@ +"""Core MCP tool modules. + +Each module defines plain `async def` tools at module level (tests call them +directly) and a `register(register_tool)` that lists them with their minimum +role. `register_tool(fn, role=..., read_only=...)` takes both as required +keywords: a tool with no declared role is unreachable, and there is no default +read_only for a mutating tool to inherit by forgetting. + +A tool's docstring IS its contract with the agent — it ships as the tool's +description. Say what it is for, when to reach for it over its siblings, what +each argument's empty value means, and what comes back. +""" +from __future__ import annotations + +from . import session + +_MODULES = (session,) + + +def register_all(register_tool) -> None: + for module in _MODULES: + module.register(register_tool) diff --git a/steward/mcp/tools/session.py b/steward/mcp/tools/session.py new file mode 100644 index 0000000..ec48e69 --- /dev/null +++ b/steward/mcp/tools/session.py @@ -0,0 +1,33 @@ +"""Session tools: who the connected client is acting as.""" +from __future__ import annotations + +from steward.mcp._context import current_caller +from steward.models.users import UserRole + +_ROLE_MEANS = { + UserRole.viewer: "Read-only: every list_*/get_* tool. Nothing that changes Steward or your hosts.", + UserRole.operator: "Viewer tools, plus running playbooks and changing monitors, alerts and schedules.", + UserRole.admin: "Everything, including deleting monitors and alert rules.", +} + + +async def whoami() -> dict: + """Who this connection acts as in Steward, and what that role may do. + + Call this first if a tool refuses you for lack of role: `role` is the + LOWER of the token's ceiling and its user's current role, so it can be + less than the token was created with. A higher role needs a new token + from Steward's AI access page, and only up to the user's own role. + """ + caller = current_caller() + return { + "user": caller.username, + "user_id": caller.user_id, + "role": caller.role.value, + "role_allows": _ROLE_MEANS[caller.role], + "token_id": caller.token_id, + } + + +def register(register_tool) -> None: + register_tool(whoami, role=UserRole.viewer, read_only=True) diff --git a/tests/integration/test_mcp.py b/tests/integration/test_mcp.py new file mode 100644 index 0000000..8efd13a --- /dev/null +++ b/tests/integration/test_mcp.py @@ -0,0 +1,74 @@ +"""Integration: the /mcp endpoint end to end against real Postgres. + +A token is minted through the real service, then MCP requests go through the +app's ASGI callable (the same path a client's HTTP request takes): the +database gate, bearer resolution and role filtering all run for real. +""" +from __future__ import annotations + +import asyncio +import json +import os +import uuid + +import pytest + +from tests.mcp._http import HEADERS as _HEADERS, asgi_request, rpc, rpc_result + +pytestmark = pytest.mark.integration + +_NEEDS_DB = pytest.mark.skipif( + not os.environ.get("STEWARD_DATABASE_URL"), + reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)", +) + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +@_NEEDS_DB +def test_mcp_over_http_with_a_minted_token(app): + from steward.core import api_tokens + from steward.models.users import User, UserRole + + async def _go(): + uid = str(uuid.uuid4()) + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add(User(id=uid, username=f"mcp-{uid[:8]}", email=f"{uid}@test", + password_hash="x", role=UserRole.admin, is_active=True)) + async with app.db_sessionmaker() as s: + raw, token = await api_tokens.create_token( + s, user_id=uid, user_role=UserRole.admin, name="it", role=UserRole.viewer) + await s.commit() + + auth = {**_HEADERS, "Authorization": f"Bearer {raw}"} + async with app.mcp_server.session_manager.run(): + status, _, body = await asgi_request(app, "POST", "/mcp", auth, rpc("initialize", { + "protocolVersion": "2025-06-18", "capabilities": {}, + "clientInfo": {"name": "it", "version": "0"}})) + assert status == 200, body + + status, _, body = await asgi_request(app, "POST", "/mcp", auth, rpc("tools/list", id_=2)) + assert status == 200, body + assert "whoami" in {t["name"] for t in rpc_result(body)["result"]["tools"]} + + status, _, body = await asgi_request(app, "POST", "/mcp", auth, rpc( + "tools/call", {"name": "whoami", "arguments": {}}, id_=3)) + assert status == 200, body + me = json.loads(rpc_result(body)["result"]["content"][0]["text"]) + # The token's viewer ceiling wins over the user's admin role. + assert (me["user_id"], me["role"], me["token_id"]) == (uid, "viewer", token.id) + + async with app.db_sessionmaker() as s: + await api_tokens.revoke_token(s, user_id=uid, token_id=token.id) + await s.commit() + status, _, _ = await asgi_request(app, "POST", "/mcp", auth, rpc("tools/list", id_=4)) + assert status == 401 + + asyncio.run(_go()) diff --git a/tests/mcp/__init__.py b/tests/mcp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/mcp/_http.py b/tests/mcp/_http.py new file mode 100644 index 0000000..b3ea52e --- /dev/null +++ b/tests/mcp/_http.py @@ -0,0 +1,50 @@ +"""Drive /mcp through the app's ASGI callable, as a client's HTTP request would. + +Quart's test_client never reaches /mcp (it is dispatched before Quart's +pipeline), so MCP tests call app.asgi_app directly. +""" +from __future__ import annotations + +import json + + +async def asgi_request(app, method, path, headers=None, body=None): + raw = [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()] + if body is not None: + raw.append((b"content-length", str(len(body)).encode())) + scope = { + "type": "http", "asgi": {"version": "3.0", "spec_version": "2.3"}, + "http_version": "1.1", "method": method, "scheme": "http", + "path": path, "raw_path": path.encode(), "query_string": b"", + "root_path": "", "headers": raw, + "client": ("testclient", 50000), "server": ("testserver", 80), + } + pending = [{"type": "http.request", "body": body or b"", "more_body": False}] + sent = [] + + async def receive(): + return pending.pop(0) if pending else {"type": "http.disconnect"} + + async def send(message): + sent.append(message) + + await app.asgi_app(scope, receive, send) + start = next(m for m in sent if m["type"] == "http.response.start") + data = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body") + return start["status"], dict(start.get("headers", [])), data + + +def rpc_result(data: bytes) -> dict: + """The JSON-RPC message from a JSON or a single-event SSE response body.""" + text = data.decode() + if text.lstrip().startswith("{"): + return json.loads(text) + lines = [ln[5:].strip() for ln in text.splitlines() if ln.startswith("data:")] + return json.loads(lines[-1]) + + +def rpc(method, params=None, id_=1) -> bytes: + return json.dumps({"jsonrpc": "2.0", "id": id_, "method": method, "params": params or {}}).encode() + + +HEADERS = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"} diff --git a/tests/mcp/test_server.py b/tests/mcp/test_server.py new file mode 100644 index 0000000..5ff203d --- /dev/null +++ b/tests/mcp/test_server.py @@ -0,0 +1,284 @@ +"""The /mcp transport and the role-gated tool registry. + +HTTP tests drive the app's ASGI callable directly: /mcp is dispatched before +Quart's pipeline, so Quart's test_client never reaches it. The token store and +the database probe are stood in for; tests/integration/test_mcp.py runs the +same requests against real Postgres with a minted token. +""" +from __future__ import annotations + +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from mcp.server.mcpserver.exceptions import ToolError +from sqlalchemy.exc import OperationalError + +from steward.core.api_tokens import Caller +from steward.mcp import _context +from steward.mcp.server import INSTRUCTIONS, StewardMCPServer +from steward.models.users import UserRole +from tests.mcp._http import HEADERS as _HEADERS, asgi_request, rpc, rpc_result + +VIEWER = Caller(user_id="u1", username="ana", role=UserRole.viewer, token_id=1) +OPERATOR = Caller(user_id="u1", username="ana", role=UserRole.operator, token_id=2) + + +@pytest.fixture +def as_caller(): + tokens = [] + + def _set(caller): + tokens.append(_context._caller_ctx.set(caller)) + yield _set + for t in reversed(tokens): + _context._caller_ctx.reset(t) + + +def _server(plugins_cfg=None): + app = MagicMock() + app.config = {"PLUGINS": plugins_cfg or {}} + return StewardMCPServer(app) + + +# ── the registry ───────────────────────────────────────────────────────────── + +async def test_every_registered_tool_declares_a_role(app): + # Default-deny: the role map is the only door, so a tool the SDK knows about + # without a role would be unreachable — and a sign someone bypassed + # register_tool. Every tool in the built server must be in the map. + mcp = app.mcp_server + names = {t.name for t in mcp._tool_manager.list_tools()} + assert names, "no MCP tools registered" + assert names == set(mcp.tool_roles), f"tools without a declared role: {names - set(mcp.tool_roles)}" + + +def test_a_tool_added_around_register_tool_is_unreachable(): + mcp = _server() + + async def sneaky() -> dict: + return {} + mcp.add_tool(sneaky) + assert "no Steward tool" in mcp.refusal("sneaky", OPERATOR) + + +def test_register_tool_requires_a_role_enum_and_unique_names(): + mcp = _server() + + async def t1() -> dict: + return {} + with pytest.raises(TypeError): + mcp.register_tool(t1, role="viewer", read_only=True) + mcp.register_tool(t1, role=UserRole.viewer, read_only=True) + with pytest.raises(ValueError, match="already registered"): + mcp.register_tool(t1, role=UserRole.viewer, read_only=True) + + +def test_read_only_reaches_the_tool_annotations(): + mcp = _server() + + async def look() -> dict: + return {} + + async def change() -> dict: + return {} + mcp.register_tool(look, role=UserRole.viewer, read_only=True) + mcp.register_tool(change, role=UserRole.operator, read_only=False) + assert mcp._tool_manager.get_tool("look").annotations.read_only_hint is True + assert mcp._tool_manager.get_tool("change").annotations.read_only_hint is False + assert mcp._tool_manager.get_tool("change").annotations.destructive_hint is True + + +def test_instructions_fit_before_claude_codes_cut(): + # Claude Code keeps only the first ~2,048 characters of server instructions. + assert len(INSTRUCTIONS) <= 1600 + + +# ── role and plugin gating ─────────────────────────────────────────────────── + +def _gated_server(plugins_cfg=None): + mcp = _server(plugins_cfg) + + async def look() -> dict: + return {"ok": True} + + async def change() -> dict: + return {"changed": True} + + async def plug() -> dict: + return {} + mcp.register_tool(look, role=UserRole.viewer, read_only=True) + mcp.register_tool(change, role=UserRole.operator, read_only=False) + mcp.register_tool(plug, role=UserRole.viewer, read_only=True, plugin="docker") + return mcp + + +async def test_list_tools_hides_what_the_role_cannot_reach(as_caller): + mcp = _gated_server({"docker": {"enabled": True}}) + as_caller(VIEWER) + assert {t.name for t in await mcp.list_tools()} == {"look", "plug"} + as_caller(OPERATOR) + assert {t.name for t in await mcp.list_tools()} == {"look", "change", "plug"} + + +async def test_list_tools_is_empty_without_a_caller(): + assert await _gated_server().list_tools() == [] + + +async def test_call_above_role_is_refused_with_a_reason(as_caller): + mcp = _gated_server() + as_caller(VIEWER) + with pytest.raises(ToolError, match="needs the operator role.*acts as viewer"): + await mcp.call_tool("change", {}) + + +async def test_disabled_plugin_tools_are_hidden_and_refused(as_caller): + mcp = _gated_server({"docker": {"enabled": False}}) + as_caller(OPERATOR) + assert "plug" not in {t.name for t in await mcp.list_tools()} + with pytest.raises(ToolError, match="docker plugin, which is disabled"): + await mcp.call_tool("plug", {}) + + +async def test_undeclared_arguments_are_refused_not_dropped(as_caller): + mcp = _server() + + async def find(host: str = "") -> dict: + return {"host": host} + mcp.register_tool(find, role=UserRole.viewer, read_only=True) + as_caller(VIEWER) + with pytest.raises(ToolError, match=r"'hostname' \(did you mean 'host'\?\)"): + await mcp.call_tool("find", {"hostname": "nas"}) + + +async def test_value_error_reaches_the_agent_as_its_reason(as_caller): + mcp = _server() + + async def picky() -> dict: + raise ValueError("host 'nas' not found; list_hosts shows the names") + mcp.register_tool(picky, role=UserRole.viewer, read_only=True) + as_caller(VIEWER) + with pytest.raises(ToolError, match="host 'nas' not found"): + await mcp.call_tool("picky", {}) + + +async def test_database_failure_is_reported_as_an_outage(as_caller): + mcp = _server() + + async def reads() -> dict: + raise OperationalError("SELECT 1", {}, ConnectionRefusedError()) + mcp.register_tool(reads, role=UserRole.viewer, read_only=True) + as_caller(VIEWER) + with pytest.raises(ToolError, match="can't reach its database"): + await mcp.call_tool("reads", {}) + + +async def test_whoami_reports_the_effective_role(as_caller): + from steward.mcp.tools.session import whoami + as_caller(VIEWER) + out = await whoami() + assert out["user"] == "ana" and out["role"] == "viewer" and out["token_id"] == 1 + + +# ── plugin hook ────────────────────────────────────────────────────────────── + +def test_plugin_tools_register_with_their_plugin_and_reload_cleanly(): + mcp = _server({"docker": {"enabled": True}}) + + async def list_containers() -> dict: + return {} + module = MagicMock() + module.register_mcp_tools = lambda register: register( + list_containers, role=UserRole.viewer, read_only=True) + + mcp.register_plugin("docker", module) + assert mcp.tool_plugins == {"list_containers": "docker"} + mcp.register_plugin("docker", module) # hot-reload: no duplicate error + assert mcp.tool_plugins == {"list_containers": "docker"} + + +def test_a_raising_plugin_hook_leaves_the_server_intact(): + mcp = _server() + module = MagicMock() + module.register_mcp_tools = MagicMock(side_effect=RuntimeError("boom")) + mcp.register_plugin("broken", module) + assert mcp.tool_plugins == {} + + +def test_a_plugin_hook_failing_halfway_registers_none_of_its_tools(): + mcp = _server() + + async def first() -> dict: + return {} + + def hook(register): + register(first, role=UserRole.viewer, read_only=True) + register(first, role=UserRole.viewer, read_only=True) # duplicate → raises + module = MagicMock() + module.register_mcp_tools = hook + mcp.register_plugin("half", module) + assert mcp.tool_plugins == {} + assert "first" not in mcp.tool_roles + assert mcp._tool_manager.get_tool("first") is None + + +def test_a_plugin_without_the_hook_contributes_nothing(): + mcp = _server() + mcp.register_plugin("quiet", object()) + assert mcp.tool_plugins == {} + + +# ── HTTP ───────────────────────────────────────────────────────────────────── + +@pytest.fixture +def db_up(): + with patch("steward.database.ensure_database_reachable", new=AsyncMock()): + yield + + +async def test_http_without_a_token_is_401(app, db_up): + status, headers, body = await asgi_request(app, "POST", "/mcp", _HEADERS, rpc("tools/list")) + assert status == 401 + assert headers[b"www-authenticate"].startswith(b"Bearer") + assert b"AI access" in body + + +async def test_http_with_an_unknown_token_is_401(app, db_up): + with patch("steward.core.api_tokens.resolve_token", new=AsyncMock(return_value=None)): + status, _, _ = await asgi_request( + app, "POST", "/mcp", {**_HEADERS, "Authorization": "Bearer stw_nope"}, rpc("tools/list")) + assert status == 401 + + +async def test_http_when_the_database_is_down_is_503(app): + from steward.database import DatabaseUnavailable + app._db_engine = MagicMock() + with patch("steward.database.ensure_database_reachable", + new=AsyncMock(side_effect=DatabaseUnavailable("connection refused"))): + status, _, body = await asgi_request( + app, "POST", "/mcp", {**_HEADERS, "Authorization": "Bearer stw_x"}, rpc("tools/list")) + assert status == 503 + assert json.loads(body)["error"] == "database_unavailable" + + +async def test_http_initialize_list_and_call_with_a_valid_token(app, db_up): + auth = {**_HEADERS, "Authorization": "Bearer stw_valid"} + with patch("steward.core.api_tokens.resolve_token", new=AsyncMock(return_value=VIEWER)): + async with app.mcp_server.session_manager.run(): + status, _, body = await asgi_request(app, "POST", "/mcp", auth, rpc("initialize", { + "protocolVersion": "2025-06-18", "capabilities": {}, + "clientInfo": {"name": "test", "version": "0"}})) + assert status == 200, body + assert rpc_result(body)["result"]["serverInfo"]["name"] == "steward" + + status, _, body = await asgi_request(app, "POST", "/mcp", auth, rpc("tools/list", id_=2)) + assert status == 200, body + names = {t["name"] for t in rpc_result(body)["result"]["tools"]} + assert "whoami" in names + + status, _, body = await asgi_request(app, "POST", "/mcp", auth, rpc( + "tools/call", {"name": "whoami", "arguments": {}}, id_=3)) + assert status == 200, body + result = rpc_result(body)["result"] + assert not result.get("isError") + assert json.loads(result["content"][0]["text"])["role"] == "viewer"