feat(mcp): connect card, MCP docs, and an end-to-end roles/audit test over HTTP

MCP-8 #4972 "Connect card, docs, and end-to-end integration test".

- AI access page: a "Connect a client" card with the /mcp URL (from
  Settings → General → Public base URL, else the request host) and recipes
  for Claude Code, Claude Desktop (mcp-remote) and other Streamable HTTP
  clients, plus the reverse-proxy note. The token-created page shows the
  same recipes with the new token filled in, ready to paste.
- docs/core/mcp.md: connecting, the reverse-proxy requirements, the
  security model (token ceiling, default deny, check-mode default, secrets,
  audit), tools by role, response conventions, adding tools.
  docs/architecture.md describes the /mcp dispatch outside the blueprint
  pipeline; docs/reference/code-map.md gains an MCP section.
- tests/integration/test_mcp.py: over HTTP, a viewer token lists no write
  tools and is refused one by name; an operator token reads get_overview
  and its maintenance-window create/end land in the audit log as
  "<user> via MCP" with the token id. (Revoked → 401 was already covered.)

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

tests/integration/test_api_tokens.py

@@ -139,6 +139,8 @@ def test_routes_create_list_and_revoke(app):
         assert "You have no tokens yet" in page
         # An operator can't offer admin.
         assert 'value="admin"' not in page
+        # The connect card names the endpoint, with a placeholder for the token.
+        assert 'value="http://localhost/mcp"' in page and "claude mcp add" in page
 
         resp = await client.post("/ai-access/tokens", form={
             "name": "laptop", "role": "admin", "expires_in_days": ""})
@@ -148,7 +150,10 @@ def test_routes_create_list_and_revoke(app):
         resp = await client.post("/ai-access/tokens", form={
             "name": "laptop", "role": "operator", "expires_in_days": "30"})
         assert resp.status_code == 200
-        raw = re.search(r'value="(stw_[^"]+)"', await resp.get_data(as_text=True)).group(1)
+        created = await resp.get_data(as_text=True)
+        raw = re.search(r'value="(stw_[^"]+)"', created).group(1)
+        # The recipes on the created page carry the new token, ready to paste.
+        assert f"Bearer {raw}" in created and "mcp-remote" in created
         caller = await api_tokens.resolve_token(app, raw)
         assert caller is not None and caller.role == UserRole.operator
 

tests/integration/test_mcp.py

@@ -72,3 +72,74 @@ def test_mcp_over_http_with_a_minted_token(app):
             assert status == 401
 
     asyncio.run(_go())
+
+
+def _tool_result(body) -> dict:
+    return rpc_result(body)["result"]
+
+
+@_NEEDS_DB
+def test_roles_reads_and_audited_writes_over_http(app):
+    """Over HTTP: a viewer sees no write tools and is refused one by name; an
+    operator reads get_overview and makes an audited change."""
+    from sqlalchemy import select
+    from steward.core import api_tokens
+    from steward.models.audit import AuditEvent
+    from steward.models.users import User, UserRole
+
+    async def _go():
+        uid = str(uuid.uuid4())
+        username = f"mcp-op-{uid[:8]}"
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                s.add(User(id=uid, username=username, email=f"{uid}@test",
+                           password_hash="x", role=UserRole.operator, is_active=True))
+        async with app.db_sessionmaker() as s:
+            viewer_raw, _ = await api_tokens.create_token(
+                s, user_id=uid, user_role=UserRole.operator, name="ro", role=UserRole.viewer)
+            op_raw, op_token = await api_tokens.create_token(
+                s, user_id=uid, user_role=UserRole.operator, name="rw", role=UserRole.operator)
+            await s.commit()
+        viewer = {**_HEADERS, "Authorization": f"Bearer {viewer_raw}"}
+        operator = {**_HEADERS, "Authorization": f"Bearer {op_raw}"}
+
+        async with app.mcp_server.session_manager.run():
+            _, _, body = await asgi_request(app, "POST", "/mcp", viewer, rpc("tools/list", id_=1))
+            names = {t["name"] for t in _tool_result(body)["tools"]}
+            assert "get_overview" in names and "list_playbooks" in names
+            assert not names & {"run_playbook", "create_monitor", "delete_alert_rule"}
+
+            _, _, body = await asgi_request(app, "POST", "/mcp", viewer, rpc(
+                "tools/call", {"name": "create_maintenance_window",
+                               "arguments": {"name": "x", "duration_minutes": 5}}, id_=2))
+            refused = _tool_result(body) if "result" in rpc_result(body) else rpc_result(body)["error"]
+            assert "needs the operator role" in json.dumps(refused)
+
+            _, _, body = await asgi_request(app, "POST", "/mcp", operator, rpc(
+                "tools/call", {"name": "get_overview", "arguments": {}}, id_=3))
+            result = _tool_result(body)
+            assert not result.get("isError"), result
+            overview = json.loads(result["content"][0]["text"])
+            assert {"generated_at", "all_clear", "counts", "findings"} <= set(overview)
+
+            name = f"it-window-{uid[:8]}"
+            _, _, body = await asgi_request(app, "POST", "/mcp", operator, rpc(
+                "tools/call", {"name": "create_maintenance_window",
+                               "arguments": {"name": name, "duration_minutes": 5}}, id_=4))
+            result = _tool_result(body)
+            assert not result.get("isError"), result
+            window_id = json.loads(result["content"][0]["text"])["window_id"]
+            _, _, body = await asgi_request(app, "POST", "/mcp", operator, rpc(
+                "tools/call", {"name": "end_maintenance_window",
+                               "arguments": {"window": window_id}}, id_=5))
+            assert not _tool_result(body).get("isError")
+
+        async with app.db_sessionmaker() as s:
+            rows = list((await s.execute(select(AuditEvent).where(
+                AuditEvent.user_id == uid, AuditEvent.entity_id == name))).scalars())
+        assert sorted(e.action for e in rows) == ["maintenance_window.created",
+                                                  "maintenance_window.ended"]
+        assert {e.username for e in rows} == {f"{username} via MCP"}
+        assert all(json.loads(e.detail_json)["token_id"] == op_token.id for e in rows)
+
+    asyncio.run(_go())