Python script to retrieve a token from Keycloak using OAuth2's Client Credential flow

keycloak-auth-curl.py

import sys
import os
import subprocess
import json

# Get input parameters.
KEYCLOAK_URL = os.environ["KEYCLOAK_URL"]
REALM = os.environ["KEYCLOAK_REALM"]
CLIENT_ID = os.environ["KEYCLOAK_CLIENT_ID"]
CLIENT_SECRET = os.environ["KEYCLOAK_CLIENT_SECRET"]

# Construct the token retrieval URL.
TOKEN_URL = f"https://{KEYCLOAK_URL}/realms/{REALM}/protocol/openid-connect/token"

# Construct the curl command.
curl_command = [
    "curl",
    "-X", "POST",
    "-d", f"grant_type=client_credentials",
    "-d", f"client_id={CLIENT_ID}",
    "-d", f"client_secret={CLIENT_SECRET}",
    TOKEN_URL
]

# Execute the curl command and capture the output.
try:
    result = subprocess.run(
        curl_command,
        check=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )

    # Parse the JSON response.
    response_data = json.loads(result.stdout)

    # Print the access token to standard output, without a trailing newline.
    sys.stdout.write(response_data["access_token"])

except subprocess.CalledProcessError as e:
    sys.stderr.write(f"Error occurred during curl command execution: {e.stderr}\n")
    sys.exit(1)
except json.JSONDecodeError:
    sys.stderr.write("Failed to parse JSON response from the server.\n")
    sys.exit(1)

keycloak-auth.py

import sys
import os
import requests
import pip_system_certs.wrapt_requests


# Get input parameters.
KEYCLOAK_URL = os.environ["KEYCLOAK_URL"]
REALM = os.environ["KEYCLOAK_REALM"]
CLIENT_ID = os.environ["KEYCLOAK_CLIENT_ID"]
CLIENT_SECRET = os.environ["KEYCLOAK_CLIENT_SECRET"]

# Construct the token retrieval URL.
TOKEN_URL = f"{KEYCLOAK_URL}/realms/{REALM}/protocol/openid-connect/token"

# Make the request
response = requests.post(
    url=TOKEN_URL,
    data={
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    }
)

# Raise an error if the request failed.
response.raise_for_status()

# Decode the response body.
response_data = response.json()

# Print the access token to standard output, without a trailing newline.
sys.stdout.write(response_data["access_token"])