"""Контекст MCP-запроса: какой пользователь за Bearer-ключом.
Личность резолвит гард require_mcp (app/security.py) и кладёт в
request.state.mcp_user; mcp_endpoint (app/mcp/server.py) переносит её в
ContextVar — тулы читают current_mcp_user(). Роль действует снейпшотом
выпуска ключа (канон mcp.md) ролью админа/юзера на момент выдачи.
MCP_USER — синтетический superadmin бэкдора статического TGCLIENT_ADMIN_TOKEN.
"""
from contextvars import ContextVar
from app.security import SUPERADMIN
_mcp_user: ContextVar[dict | None] = ContextVar("mcp_user", default=None)
def set_mcp_user(user: dict) -> object:
"""Гарду: установить пользователя на время запроса, вернуть token для reset()."""
return _mcp_user.set(user)
def reset_mcp_user(token: object) -> None:
_mcp_user.reset(token) # type: ignore[arg-type]
def current_mcp_user() -> dict:
"""{user_id, email, system_role} текущего MCP-вызова; бэкдор для прямых
вызовов вне запроса (тесты) — возвращаем супер-юзера."""
found = _mcp_user.get()
if found is not None:
return found
return dict(SUPERADMIN)
def current_user_id() -> str:
return current_mcp_user()["user_id"]
def user_role() -> str:
return (current_mcp_user().get("system_role") or "user").lower()
def is_admin() -> bool:
return user_role() in ("admin", "superadmin")