File size: 1,000 Bytes
92d7038
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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