Weather MCP

from typing import Any import httpx from app.core.auth.exceptions import InvalidTokenError from app.core.auth.models import AuthContext class AdminIntrospectAuthVerifier: def __init__(self, base_url: str, me_path: str, timeout_seconds: float) -> None: self._base_url = base_url.rstrip("/") self._me_path = me_path self._timeout = timeout_seconds async def verify(self, token: str) -> AuthContext: url = f"{self._base_url}{self._me_path}" headers = {"Authorization": f"Bearer {token}"} async with httpx.AsyncClient(timeout=self._timeout) as client: resp = await client.get(url, headers=headers) if resp.status_code != 200: raise InvalidTokenError("token introspection failed") data: dict[str, Any] = resp.json() if resp.content else {} user_id = data.get("id") or data.get("user_id") or data.get("sub") if not user_id: raise InvalidTokenError("admin /me did not return user id") return AuthContext(user_id=str(user_id), token=token, claims=data)