InfluxDB MCP Server

import jwt from datetime import datetime, timedelta from .config import JWT_SECRET, JWT_ALGORITHM class AuthError(Exception): """Exception raised for authentication errors.""" pass def validate_token(token): """ Validate a JWT token. Args: token (str): The JWT token to validate Returns: dict: The decoded payload if valid Raises: AuthError: If the token is invalid or expired """ if not JWT_SECRET: raise AuthError("JWT_SECRET not configured") try: payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) return payload except jwt.ExpiredSignatureError: raise AuthError("Token expired") except jwt.InvalidTokenError: raise AuthError("Invalid token") def create_token(username, expires_delta=timedelta(hours=1)): """ Create a JWT token. Args: username (str): The username to include in the token expires_delta (timedelta): How long the token should be valid Returns: str: The JWT token """ if not JWT_SECRET: raise AuthError("JWT_SECRET not configured") payload = { "sub": username, "iat": datetime.utcnow(), "exp": datetime.utcnow() + expires_delta } return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)