# ---------------------- # ----- Enneo SDK ------ # ---------------------- import os import json import requests from requests import Response import sys import warnings from datetime import datetime from pydantic import BaseModel, Field from typing import Union, Optional, List, Dict from enum import Enum ENNEO_SDK_VERSION = "0.2.4" # Conversation context flags # This mirrors the PHP SDK's isBotConversation behavior IS_BOT_CONVERSATION: bool = False def get_sdk_version() -> str: """ Returns the version of the Enneo SDK. Returns: str: The version of the Enneo SDK. """ return str(ENNEO_SDK_VERSION) def load_input_data(include_metadata: bool = True) -> Dict: """ Loads the input parameters for the AI function that enneo was able to extract from the ticket/contract This function first checks if the input parameters are passed through STDIN. If not, it checks if the input parameters are passed through a 'input.json' file in the working directory. If the input parameters are passed through STDIN or the 'input.json' file, it returns the loaded input data as a dictionary. Returns: Dict: The loaded input data as a dictionary. Raises: FileNotFoundError: If the 'input.json' file is not found in the working directory. json.JSONDecodeError: If the input parameters are not valid JSON. Exception: If any other unexpected error occurs. """ input_data = None try: # Option 1: Parameters are passed through STDIN if not sys.stdin.isatty(): input_data = sys.stdin.read() # Option 2: Parameters are passed through a file if not input_data: input_json_path = os.getenv('INPUT_JSON', 'input.json') with open(input_json_path, 'r') as file: input_data = json.load(file) else: input_data = json.loads(input_data) # Some input values might be strings that are actually JSON objects # We try to decode them here for key, value in input_data.items(): try: # Try converting the string value to a dict decoded_value = json.loads(value) input_data[key] = decoded_value except (TypeError, json.JSONDecodeError): pass # Determine conversation context (bot vs. human-assisted) global IS_BOT_CONVERSATION metadata = input_data.get("_metadata", {}) IS_BOT_CONVERSATION = (metadata.get("aiSupportLevel","") == "bot") or (metadata.get("channel","") == "phone") # Strip any metadata unless requested if include_metadata == False: if "_metadata" in input_data: del input_data["_metadata"] return input_data except FileNotFoundError: print("Error: Input parameters must either be passed through STDIN or through a 'input.json' file in the working directory.") except json.JSONDecodeError: print("Error: Input parameters must be valid JSON.") print("Your input was: ", input_data) except Exception as e: print(f"An unexpected error occurred: {e}") exit(1) def is_bot_conversation() -> bool: """ Check if we are in a bot conversation. Bot conversation = Chatbot or Voicebot Human conversation = Email, or Chat/Voice when assisting the customer service representative Returns: bool: True if conversation is bot-driven, otherwise False. """ return bool(IS_BOT_CONVERSATION) class IntentType(str, Enum): success = 'success' neutral = 'neutral' warning = 'warning' danger = 'danger' class IntentInfo(BaseModel): """ Creates a new intent info (=message box visible to the agent in the Enneo UI). Attributes: type (IntentType): The type of the intent. Must be one of 'success', 'neutral', 'warning', 'danger'. message (str): The message to be shown to the agent, e.g. "Reading is plausible". extra_info (Optional[str], optional): Optional supplemental information about the intent, e.g. "Expected reading was 421 kWh. Plausbible because difference to 317 kWh is below threshold of 200 kWh" code (Optional[str], optional): Optional internal code associated with the info. Not used by enneo. Defaults to None. """ type: IntentType message: str extra_info: Optional[str] = None code: Optional[str] = None class IntentOption(BaseModel): """ Creates a new intent option (=button visible to the agent in the Enneo UI). Attributes: type (str): The type of the intent. Must be unique for each option. Usually accompanied by a matching output case name (str): The name of the option, e.g. "Enter into system". icon (Optional[str], optional): The icon to be shown next to the option. Defaults to 'check'. Options: check, cancel recommended (bool, optional): True for default action. An Intent can only have one default action. Defaults to False. order (int, optional): Sorting index order. Frontend sorts in ascending order handler (Optional[str], optional): Which microservice should handle this action, Options: cortex, mind or fe. ONLY USE IF YOU KNOW WHAT YOU ARE DOING. Defaults to None. """ type: str name: str icon: str = 'check' recommended: bool = False order: int = 0 handler: Optional[str] = None class Interaction(BaseModel): """ Creates a new interaction object. This is an enneo-specific object that defines the form shown to the agent in the Enneo UI. The standard AI function returns an json-encoded Interaction object to STDOUT Attributes: data (dict): The input data for the AI function, i.e. form values for the input variables. It is recommended to pre-fill this with the input data coming from load_input_data(). options (List[IntentOption], optional): A list of intent options (=buttons visible to the agent in the Enneo UI). Defaults to []. infos (List[IntentInfo], optional): A list of intent infos (=message boxes visible to the agent in the Enneo UI). Defaults to []. form (Optional[dict], optional): Optional dictionary to override the forms to be shown to the agent. Usually not needed. Defaults to None. """ data: dict options: List[IntentOption] = Field(default_factory=list) infos: List[IntentInfo] = Field(default_factory=list) form: Optional[dict] = None class ApiEnneo: """ Helper class for calling the Enneo API. """ @staticmethod def get_contract(contract_id: Union[int, str]) -> dict: """ Get a contract by its ID. See https://docs.enneo.ai/de/api-reference for documentation. """ return ApiEnneo.get(f'/api/mind/contract/{contract_id}?includeRawData=true') @staticmethod def get_ticket(ticket_id: int) -> dict: """ Get a ticket by its ID. See https://docs.enneo.ai/de/api-reference for documentation. """ return ApiEnneo.get(f'/api/mind/ticket/{ticket_id}') @staticmethod def get(endpoint: str, params: dict = {}, authorizeAs: str = 'user') -> dict: """ Do an API GET call to an Enneo API endpoint See https://docs.enneo.ai/de/api-reference for documentation. """ url = ApiEnneo.__get_enneo_url(endpoint) headers = ApiEnneo.__build_authorization_header(authorizeAs) return Api.call("GET", url, headers, params) @staticmethod def post(endpoint: str, body: dict, authorizeAs: str = 'user') -> dict: """ Do an API POST call to an Enneo API endpoint See https://docs.enneo.ai/de/api-reference for documentation. """ url = ApiEnneo.__get_enneo_url(endpoint) headers = ApiEnneo.__build_authorization_header(authorizeAs) return Api.call("POST", url, headers, body) @staticmethod def patch(endpoint: str, body: dict, authorizeAs: str = 'user') -> dict: """ Do an API PATCH call to an Enneo API endpoint See https://docs.enneo.ai/de/api-reference for documentation. """ url = ApiEnneo.__get_enneo_url(endpoint) headers = ApiEnneo.__build_authorization_header(authorizeAs) return Api.call("PATCH", url, headers, body) @staticmethod def put(endpoint: str, body: dict, authorizeAs: str = 'user') -> dict: """ Do an API PUT call to an Enneo API endpoint See https://docs.enneo.ai/de/api-reference for documentation. """ url = ApiEnneo.__get_enneo_url(endpoint) headers = ApiEnneo.__build_authorization_header(authorizeAs) return Api.call("PUT", url, headers, body) @staticmethod def delete(endpoint: str, authorizeAs: str = 'user') -> dict: """ Do an API DELETE call to an Enneo API endpoint See https://docs.enneo.ai/de/api-reference for documentation. """ url = ApiEnneo.__get_enneo_url(endpoint) headers = ApiEnneo.__build_authorization_header(authorizeAs) return Api.call("DELETE", url, headers) @staticmethod def executeUdf(name: str, params: dict, authorizeAs: str = 'user') -> dict: """ Execute a UDF (User Defined Function) in Enneo. See https://docs.enneo.ai/de/api-reference for documentation. """ if not authorizeAs: authorizeAs = 'serviceWorker' return ApiEnneo.post('/api/mind/executor/execute/' + name, {'parameters': params}, authorizeAs) @staticmethod def get_file_from_storage(path: str) -> Optional[bytes]: """ Extract a file from an Enneo storage endpoint. """ endpoint = f"/api/mind/storage/{path}" url = ApiEnneo.__get_enneo_url(endpoint) headers = ApiEnneo.__build_authorization_header('user') response = Api.call_raw("GET", url, headers) return response.content @staticmethod def __build_authorization_header(authorize_as: str) -> dict: if authorize_as == 'serviceWorker' or not os.getenv('ENNEO_USER_AUTH_HEADER'): return {'Authorization': f'Bearer {os.environ["ENNEO_SESSION_TOKEN"]}'} elif authorize_as == 'user': header_split = os.environ["ENNEO_USER_AUTH_HEADER"].split(': ') return {header_split[0]: header_split[1]} else: raise Exception(f"Unknown authorization type {authorize_as}. Use 'user' or 'serviceWorker'.", 400) @staticmethod def __get_enneo_url(endpoint: str) -> str: enneo_url = os.getenv('ENNEO_API_URL','') if endpoint.startswith('/api/mind'): pass elif endpoint.startswith('/api/cortex'): enneo_url = enneo_url.replace(':8005', ':8006') elif endpoint.startswith('/api/auth'): enneo_url = enneo_url.replace(':8005', ':8002') else: raise Exception(f"Unknown Enneo API endpoint {endpoint}. They should start with /api/mind or /api/cortex or /api/auth") return enneo_url + endpoint class Api: """ Helper class for calling APIs. """ @staticmethod def call_raw(method: str, url: str, headers: dict = {}, params: Union[dict, object, str, bool] = False, verify: bool = True) -> Response: """ Do an API call to an API endpoint and return the Response object. Args: method (str): HTTP method: GET, POST, PUT, DELETE or PATCH url (str): URL to call, e.g. https://my-api.com/api/v1/endpoint headers (dict, optional): The headers to send with the request, e.g. {'Authorization': 'Bearer xxx'}. Defaults to {}. params (Union[dict, object, str, bool], optional): The body to send with the request if request is POST/PUT/DELETE. Defaults to False. """ if method == "GET": response = requests.get(url, headers=headers, verify=verify) elif method == "POST": response = requests.post(url, headers=headers, json=params, verify=verify) elif method == "PUT": response = requests.put(url, headers=headers, json=params, verify=verify) elif method == "DELETE": response = requests.delete(url, headers=headers, json=params, verify=verify) elif method == "PATCH": response = requests.patch(url, headers=headers, json=params, verify=verify) else: raise Exception(f"Unknown method {method}") if response.status_code == 200: return response else: if url.startswith(os.environ['ENNEO_API_URL']): # For internal Enneo API calls, we show the original error message directly raise Exception(response.content) else: raise Exception(f"Api call to {url} failed with code {response.status_code} and response: {response.content}") @staticmethod def call(method: str, url: str, headers: dict = {}, params: Union[dict, object, str, bool] = False, verify: bool = True) -> dict: """ Do an API call to an API endpoint Args: method (str): HTTP method: GET, POST, PUT, DELETE or PATCH url (str): URL to call, e.g. https://my-api.com/api/v1/endpoint headers (dict, optional): The headers to send with the request, e.g. {'Authorization': 'Bearer xxx'}. Defaults to {}. params (Union[dict, object, str, bool], optional): The body to send with the request if request is POST/PUT/DELETE. Defaults to False. """ response = Api.call_raw(method, url, headers, params, verify) return response.json() class Setting: """ Helper class for getting settings from Enneo through the API. """ @staticmethod def get(setting_name: str) -> Union[object, None, str]: return ApiEnneo.get(f'/api/mind/settings/compact?showSecrets=true&filterByName={setting_name}', authorizeAs='serviceWorker').get(setting_name, None) @staticmethod def set(setting_name: str, setting_value: Union[dict, str, int, float, bool, List, None]) -> dict: """ Set a setting by its name. This will update the setting in Enneo. """ settings_data = {setting_name: setting_value} return ApiEnneo.post('/api/mind/settings', settings_data, authorizeAs='serviceWorker') class ApiPowercloud: """ Helper class for calling the Powercloud API. Authorization is handled by this SDK. Note: As Powercloud unfortunately has frequent issues with invalid SSL certificates, we disable SSL verification for all calls to the Powercloud API. """ @staticmethod def get_call(endpoint: str) -> dict: warnings.filterwarnings("ignore", message="Unverified HTTPS request") url = str(Setting.get('powercloudApiUrl')) + endpoint headers = {'Authorization': f'Basic {Setting.get("powercloudApiAuth")}'} return Api.call("GET", url, headers, verify=False) @staticmethod def post_call(endpoint: str, params: Union[dict, object]) -> dict: warnings.filterwarnings("ignore", message="Unverified HTTPS request") url = str(Setting.get('powercloudApiUrl')) + endpoint headers = {'Authorization': f'Basic {Setting.get("powercloudApiAuth")}'} return Api.call("POST", url, headers, params, verify=False) @staticmethod def extract_error(result: Union[object, str]) -> str: message = 'Powercloud-Fehler: ' if not isinstance(result, dict): message += str(result) elif len(result.get('errors', [])) > 0: for error in result['errors']: message += f"{error['messageLocalized']}; " elif 'response' in result and isinstance(result['response'], str): message += result['response'] elif 'messageLocalized' in result: message += result['messageLocalized'] else: message += str(result) return message class AppInfo: """ Helper class for accessing app metadata from environment variables. These are set automatically by the Enneo platform when running an app. """ @staticmethod def get_id() -> Optional[str]: """Get the app UUID.""" return os.getenv('ENNEO_APP_ID') @staticmethod def get_slug() -> Optional[str]: """Get the app slug.""" return os.getenv('ENNEO_APP_SLUG') @staticmethod def get_version() -> Optional[str]: """Get the app version.""" return os.getenv('ENNEO_APP_VERSION') class AppStorage: """ Persistent key-value storage for apps. Values must be JSON-compliant (null, string, array, object, float, int). """ @staticmethod def _get_app_id() -> str: app_id = AppInfo.get_id() if not app_id: raise Exception("ENNEO_APP_ID not set. AppStorage is only available when running as an app.") return app_id @staticmethod def get(key: str) -> Union[dict, list, str, int, float, bool, None]: """ Get a value by key. Args: key: The storage key Returns: The stored value, or raises Exception if not found """ app_id = AppStorage._get_app_id() result = ApiEnneo.get(f'/api/mind/app/{app_id}/data/{key}', authorizeAs='serviceWorker') return result.get('value') @staticmethod def get_meta(key: str) -> dict: """ Get a value with metadata (createdAt, modifiedAt, lastAccessAt). Args: key: The storage key Returns: Dict with keys: key, value, createdAt, modifiedAt, lastAccessAt """ app_id = AppStorage._get_app_id() return ApiEnneo.get(f'/api/mind/app/{app_id}/data/{key}/meta', authorizeAs='serviceWorker') @staticmethod def save(key: str, value: Union[dict, list, str, int, float, bool, None]) -> dict: """ Save a value. Creates a new entry if the key doesn't exist, updates if it does. Args: key: The storage key value: The value to store (must be JSON-compliant) Returns: Dict with success status """ app_id = AppStorage._get_app_id() return ApiEnneo.put(f'/api/mind/app/{app_id}/data/{key}', value, authorizeAs='serviceWorker') @staticmethod def delete(key: str) -> dict: """ Delete a key from storage. Args: key: The storage key to delete """ app_id = AppStorage._get_app_id() return ApiEnneo.delete(f'/api/mind/app/{app_id}/data/{key}', authorizeAs='serviceWorker') @staticmethod def list_keys() -> list: """ List all storage keys for this app. Returns: List of dicts with key, createdAt, modifiedAt, lastAccessAt """ app_id = AppStorage._get_app_id() result = ApiEnneo.get(f'/api/mind/app/{app_id}/data', authorizeAs='serviceWorker') return result.get('keys', []) class Helpers: @staticmethod def format_date(date: Union[str, datetime], format: str = 'de') -> str: """ Format a date string or datetime object to a specific format. Needed to convert date strings coming from the input data (yyyy-mm-dd) to a human-readable format (dd.mm.yyyy). Args: date (Union[str, datetime]): The date to format. format (str, optional): The format to use. Must be one of 'en' or 'de'. Defaults to 'de'. """ if isinstance(date, str): try: d = datetime.strptime(date, '%Y-%m-%d') except ValueError: d = datetime.strptime(date, '%Y-%m-%d %H:%M:%S') if not d: if date == '': raise Exception('No date to format') else: raise Exception(f"Date '{date}' invalid") else: d = datetime.strptime(date, '%Y-%m-%d') if format == 'en': return d.strftime('%Y-%m-%d') elif format == 'de': return d.strftime('%d.%m.%Y') else: raise Exception(f"Unknown language {format}") else: raise Exception("Invalid date format")