Automate Risk Management with ServiceNow's API

1. 1
   Configure API Credentials

   To interact with the ServiceNow API, you need your instance URL, a username, and a password. The user account requires roles like `sn_risk.risk_manager` and `rest_api_explorer`. Store these securely as environment variables.
2. 2
   Create a New Risk

   Programmatically add a new risk to the risk register by sending a POST request to the Risk table API (`/api/now/table/sn_risk_risk`). This allows you to integrate external event sources or bulk-import risks. The response will contain the `sys_id` of the new risk.
3. 3
   Move Risk to Assess State

   To trigger the Risk Assessment Agent, you must change the risk's state to 'Assess'. Send a PATCH request to the specific risk record using its `sys_id`. This action initiates the automated analysis and scoring workflows.
4. 4
   Retrieve Assessment Results

   After the Risk Assessment Agent has run, retrieve the results by sending a GET request. The response will contain the calculated scores (`inherent_score`, `residual_score`), the updated state, and other assessment details.

import os
import requests
import json

# Load credentials from environment variables
SN_INSTANCE = os.getenv("SN_INSTANCE")
SN_USER = os.getenv("SN_USER")
SN_PASSWORD = os.getenv("SN_PASSWORD")

if not all([SN_INSTANCE, SN_USER, SN_PASSWORD]):
    print("Error: Set SN_INSTANCE, SN_USER, and SN_PASSWORD environment variables.")
    exit(1)

BASE_URL = f"https://{SN_INSTANCE}/api/now/table/sn_risk_risk"
AUTH = (SN_USER, SN_PASSWORD)
HEADERS = {
    "Accept": "application/json",
    "Content-Type": "application/json"
}

def create_risk(description):
    """Creates a new risk in ServiceNow and returns its sys_id."""
    print(f"Creating risk: '{description}'...")
    payload = {
        "short_description": description,
        "category": "operational",
        "state": "1" # Draft
    }
    response = requests.post(BASE_URL, auth=AUTH, headers=HEADERS, data=json.dumps(payload))
    response.raise_for_status()
    result = response.json().get('result', {})
    risk_sys_id = result.get('sys_id')
    print(f"Successfully created risk with sys_id: {risk_sys_id}")
    return risk_sys_id

def get_risk_details(sys_id):
    """Retrieves details for a specific risk."""
    print(f"\nFetching details for risk {sys_id}...")
    url = f"{BASE_URL}/{sys_id}"
    params = {'sysparm_fields': 'number,state,short_description,inherent_score,residual_score'}
    response = requests.get(url, auth=AUTH, headers=HEADERS, params=params)
    response.raise_for_status()
    result = response.json().get('result', {})
    print("Risk Details:")
    print(json.dumps(result, indent=2))

if __name__ == "__main__":
    try:
        # Note: In a real scenario, you'd find an entity's sys_id to link it.
        risk_id = create_risk("Automated Test: Unauthorized access to financial reporting system")
        if risk_id:
            get_risk_details(risk_id)
            # To trigger assessment, you would PATCH the state to -2 ('Assess')
            # print("\nNext step: Update the risk state to 'Assess' to trigger the agent.")
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e.response.status_code} {e.response.text}")
    except Exception as e:
        print(f"An error occurred: {e}")