Start task sample

python automate-mft-api-calls.py <environment> <tenant_id> <path_to_private_key_file> <kid> [task_id]

To call the licenses endpoint in the US environment, run:

python automate-mft-api-calls.py us <tenant_id> <path_to_private_key_file> <kid>

To start a task and then check its status, include the optional task_id:

python automate-mft-api-calls.py us <tenant_id> <path_to_private_key_file> <kid> <task_id>

import jwt
import time
import uuid
import sys
import requests
import json

# Mandatory information required:
# - environment - "us" or "eu" (provided as first argument)
# - tenant_id - The Tenant ID in the UI
# - private key file path
# - kid - The Public Key ID in the UI
#
# Order of steps:
# 1. Create JWT token
# 2. Get OAuth access token using JWT client assertion (signed JWT token with the private key)
# 3. Call APIs with the access token

def get_base_url(environment):
    """
    Get the base URL for the API based on the environment
    Args:
        environment (str): The environment - "us" or "eu"
    Returns:
        str: The base URL
    """
    env_lower = environment.lower()
    if env_lower in ["us", "eu"]:
        return f"https://api.{env_lower}.mft.progress.com"
    elif env_lower == "staging":
        return "https://api.staging.mft.progress.com"
    else:
        print(f"Error: Invalid environment '{environment}'. Must be 'us', 'eu', or 'staging'", file=sys.stderr)
        sys.exit(1)

def retry_on_429(func, *args, **kwargs):
    """
    Retry a function if it returns 429 (Too Many Requests)
    Args:
        func: The function to call
        *args: Arguments to pass to the function
        **kwargs: Keyword arguments to pass to the function
    Returns:
        The result of the function call
    """
    max_retries = 3
    retry_count = 0
    while retry_count < max_retries:
        try:
            return func(*args, **kwargs)
        except requests.exceptions.HTTPError as error:
            if error.response.status_code == 429:
                retry_count += 1
                if retry_count < max_retries:
                    print(f"Received 429, retrying in 5 seconds... (attempt {retry_count}/{max_retries})")
                    time.sleep(5)
                else:
                    print(f"Max retries reached for 429", file=sys.stderr)
                    raise
            else:
                raise

def create_jwt_token(private_key, kid, tenant_id, base_url):
    """
    Create a JWT token for authentication
    Args:
        private_key (str): The private key content
        kid (str): The key ID
        tenant_id (str): The tenant ID
        base_url (str): The base URL for the API
    Returns:
        str: The JWT token
    """
    now = int(time.time())
    token = jwt.encode(
        {
            "aud": f"{base_url}/v1/oauth/token",  # Audience - the token endpoint - must be included
            "jti": str(uuid.uuid4()),            # Unique identifier for the token NOT MANDATORY but recommended
            "iat": now,                          # Issued at time, NOT MANDATORY but recommended
            "nbf": now,                          # Not before time, NOT MANDATORY but recommended
            "exp": now + 240,                    # Expiration time, mandatory, ensure it is short-lived
            "iss": tenant_id,                    # Issuer - The Tenant ID, mandatory
            "sub": tenant_id                     # Subject - The Tenant ID, mandatory
        },
        private_key,
        algorithm="ES256",
        headers={"kid": kid}
    )
    return token

def get_access_token(token, base_url):
    """
    Get OAuth access token using JWT client assertion
    Args:
        token (str): The JWT token
        base_url (str): The base URL for the API
    Returns:
        str: The access token
    """
    response = requests.post(
        f"{base_url}/v1/oauth/token",
        headers={
            "Content-Type": "application/json"
        },
        json={
            "grant_type": "client_credentials",
            "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
            "client_assertion": token
        },
        timeout=10
    )
    response.raise_for_status()
    return response.json()["access_token"]

def get_current_licenses(access_token, base_url):
    """
    Get current licenses
    Args:
        access_token (str): The OAuth access token
        base_url (str): The base URL for the API
    Returns:
        dict: The licenses data
    """
    def _get_licenses():
        response = requests.get(
            f"{base_url}/v1/licenses/current",
            headers={
                "Authorization": f"bearer {access_token}"
            }
        )
        response.raise_for_status()
        return response.json()
    return retry_on_429(_get_licenses)

def start_task(access_token, task_id, base_url):
    """
    Start a task
    Args:
        access_token (str): The OAuth access token
        task_id (str): The task ID
        base_url (str): The base URL for the API
    Returns:
        dict: The task start response
    """
    def _start_task():
        response = requests.post(
            f"{base_url}/v1/tasks/{task_id}/start",
            headers={
                "Authorization": f"bearer {access_token}"
            },
            json={}
        )
        response.raise_for_status()
        return response.json()
    return retry_on_429(_start_task)

def get_task_run(access_token, task_run_id, base_url):
    """
    Get task run status
    Args:
        access_token (str): The OAuth access token
        task_run_id (str): The task run ID
        base_url (str): The base URL for the API
    Returns:
        dict: The task run data
    """
    def _get_task_run():
        response = requests.get(
            f"{base_url}/v1/task-runs/{task_run_id}",
            headers={
                "Authorization": f"bearer {access_token}"
            }
        )
        response.raise_for_status()
        return response.json()
    return retry_on_429(_get_task_run)

def call_apis(environment, tenant_id, private_key_file, kid, task_id=None):
    """
    Main function to call APIs
    Args:
        environment (str): The environment - "us" or "eu"
        tenant_id (str): The Tenant ID
        private_key_file (str): Path to the private key file
        kid (str): The Public Key ID in the UI
        task_id (str, optional): The task ID - currently can be taken from the URL when a task is selected in the tasks page
    """
    try:
        # Get base URL from environment
        base_url = get_base_url(environment)

        # Read the private key
        with open(private_key_file, "r", encoding="utf-8") as f:
            private_key = f.read()

        # Create JWT token
        token = create_jwt_token(private_key, kid, tenant_id, base_url)

        # Get access token
        access_token = get_access_token(token, base_url)

        # Call the licenses API
        licenses_data = get_current_licenses(access_token, base_url)
        print("Licenses data:", json.dumps(licenses_data, indent=2))

        if not task_id:
            return

        # Start a task (example task_id: '02115928-16fe-4493-949b-d77d302a88ee')
        start_task_id = start_task(access_token, task_id, base_url)
        print("Started Task:", json.dumps(start_task_id, indent=2))

        task_run_status = ""
        task_run_data = None
        while task_run_status not in ["success", "failed"]:
            try:
                task_run_data = get_task_run(access_token, start_task_id["id"], base_url)
                task_run_status = task_run_data.get("status", {}).get("taskRunStatus", "")
                print(f"Task Run Status: {task_run_status}")
                # Sleep for 5 seconds before checking again
                time.sleep(5)
            except requests.exceptions.HTTPError as error:
                if error.response.status_code == 403:
                    print("Received 403, refreshing access token...")
                    # Refresh the access token
                    token = create_jwt_token(private_key, kid, tenant_id, base_url)
                    access_token = get_access_token(token, base_url)
                    print("Access token refreshed, retrying...")
                else:
                    raise

        print("Task Completed:", json.dumps(task_run_data, indent=2))

    except requests.exceptions.RequestException as error:
        if hasattr(error, 'response') and error.response is not None:
            print(f"API Error: {error.response.status_code}", file=sys.stderr)
            try:
                print(error.response.json(), file=sys.stderr)
            except:
                print(error.response.text, file=sys.stderr)
        else:
            print(f"Error: {str(error)}", file=sys.stderr)
        sys.exit(1)
    except Exception as error:
        print(f"Error: {str(error)}", file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    args = sys.argv[1:]
    if len(args) < 4:
        print("Usage: python automate-mft-api-calls.py <environment> <tenant_id> <private_key_file> <kid> [task_id]", file=sys.stderr)
        print("  environment: 'us', 'eu', or 'staging'", file=sys.stderr)
        sys.exit(1)

    environment = args[0]
    tenant_id = args[1]
    private_key_file = args[2]
    kid = args[3]
    task_id = args[4] if len(args) > 4 else None

    call_apis(environment, tenant_id, private_key_file, kid, task_id)