feat(mcp): /mcp endpoint with bearer auth and role-gated tools

Second step of milestone 461. An MCP client (Claude Code, Claude Desktop
via mcp-remote, Open WebUI) connects to /mcp with a personal API token.

- mcp>=2.2 (SDK v2 MCPServer). Mounted by wrapping app.asgi_app, so /mcp
  never enters Quart's pipeline; stateless streamable HTTP so a redeploy
  never strands a client on a session id the new process didn't issue.
- Auth: Authorization: Bearer stw_… resolves to a Caller (user +
  effective role) in a ContextVar; 401 with WWW-Authenticate otherwise.
  The web pipeline's database gate runs first, so a DB restart answers
  503 instead of failing inside a tool.
- Registry, default-deny: register_tool(fn, role=, read_only=) is the only
  door; role and read_only are required. Tools above the caller's role,
  or from a disabled plugin, are neither listed nor callable. Undeclared
  arguments are refused (not silently dropped); a tool's ValueError
  reaches the agent as its reason; a DB failure mid-tool says so.
- Plugins contribute tools via an optional register_mcp_tools(register)
  export, all-or-nothing, replaced on hot-reload. Documented in
  docs/plugins/writing-a-plugin.md.
- First tool: whoami.
- Tests drive app.asgi_app directly (Quart's test client never reaches
  /mcp); the integration test mints a real token against Postgres and
  runs initialize → tools/list → whoami → revoke → 401.

Task #4966 "MCP-2 — /mcp transport".

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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())

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"}

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"