utils/auth.py

from typing import Optional
import httpx
from dataclasses import dataclass

from utils.config import KEYCLOAK_CONFIG, TOKEN_ENDPOINT


@dataclass
class DSContext:
    """Typed context for DS MCP server"""

    client: httpx.AsyncClient



async def authenticate() -> Optional[str]:
    """Authenticate with Keycloak and return access token"""
    data = {
        "grant_type": KEYCLOAK_CONFIG["grant_type"],
        "client_id": KEYCLOAK_CONFIG["client_id"],
        "username": KEYCLOAK_CONFIG["username"],
        "password": KEYCLOAK_CONFIG["password"],
    }

    if KEYCLOAK_CONFIG["client_secret"]:
        data["client_secret"] = KEYCLOAK_CONFIG["client_secret"]

    async with httpx.AsyncClient() as client:
        response = await client.post(TOKEN_ENDPOINT, data=data)
        if response.status_code == 200:
            return response.json().get("access_token")
        else:
            print(f"Authentication failed: {response.status_code} - {response.text}")
            return None