Files
FabledSteward/steward/mcp/tools/session.py
T
bvandeusenandClaude Opus 5.5 3984f639e8
CI / lint (push) Successful in 2s
CI / integration (push) Failing after 35s
CI / unit (push) Failing after 1m0s
CI / publish (push) Skipped
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>
2026-10-06 07:24:00 -04:00

34 lines
1.2 KiB
Python

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