Newer
Older
tgclient-mcp / backend / app / mcp / context.py
"""Контекст 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")