Plugins

Due to customer modifications, the UCS server allows Python scripts to run as part of the main one process.

Running (v2.4.0)

1. When UCS starts, the /opt/ucs/server/plugins directory is searched and processed all files with .py extension (except the __init__.py file).
2. The file is imported and the variable UCS_PLUGIN is searched for. If the variable is not in the file, then it is skipped.
3. According to the string content of the UCS_PLUGIN variable, UCS will try to create an instance given class. The ucs argument is passed to the constructor, which is the server's main class.
4. If the class contains a start method, then this method is called. This method must not be blocking because plugins are not automatically created as threads.
5. When reloading the plugin or stopping UCS, the stop method is called if the plugin eats contains.

#!/usr/bin/python3

"""Example plugin"""

import threading

from constants import AGENT_AUX
from events import EVENT_AGENT_STATUS

UCS_PLUGIN = 'Example'


class Example(threading.Thread):
   """Example plugin"""

   def __init__(self, ucs):
     super().__init__(self)
     self.name = 'ExamplePluginThread'
     self.ucs = ucs
     self.ucs.addCallback(EVENT_AGENT_STATUS, self.agent_status_changed)

     self.terminate = False
     self.work = threading.Event()

   def agent_status_changed(self, agent, args):
     """Called when agent changes status"""
     if agent['status'] == AGENT_AUX and agent['reason'] == 'Lunch':
       self.ucs.log.info(f'Agent {agent["displayname"]} left for lunch')

   def run(self):
     """Thread main loop"""
     while True:
       self.work.wait(60)
       self.work.clear()
       if self.terminate:
         return

       agents = self.ucs.agents.getAgents()
       for agent in agents:
         if agent['status'] == AGENT_AUX and agent['reason'] == 'Lunch':
           self.ucs.log.info(f'Agent {agent["displayname"]} is at lunch')

   def stop(self):
     """Called on plugin reload or UCS stop"""
     self.ucs.removeCallback(EVENT_AGENT_STATUS, self.agent_status_changed)
     self.terminate = True
     self.work.set()

API Methods (v2.4.0)

Plugins can be uploaded and removed while UCS is running (superuser permission is required). The following API methods are used for this:

• ucs.api.plugins.load(sid, plugin_name) loads the plugin or if the plugin already exists, then stops it and uploads it again.
• ucs.api.plugins.unload(sid, plugin_name) stops the plugin.

Attention, if the plugin registers callbacks, callbacks are required in the stop method unregister. Otherwise, they will remain referenced and will continue to be invoked.

Agent state change hook (v2.4.0)

The plugin can affect the agent state change. UCS provides a method for registration callback ucs.agents.addChangeStatusHook(callback) and to deregister it ucs.agents.removeChangeStatusHook() when stopping the plugin. He can use this method only one plugin, if another plugin tries to register then it is invoked ProcessingError exception.

Three arguments are passed to the callback:

• agent:agents/Agent is an instance of the agent class at the moment before the state change.
• status:int is the new status (constant AGENT_STATE_ENUM).
• reason:str is the reason for the not ready status.

The callback must return status:int, reason:str or throw a ProcessingError exception.

• If the state to which the agent is switched should not be affected, then callback returns the values of the status and reason arguments as passed to it.
• If there is to be a switch to another state and/or reason for not being ready, then returns the values to which the agent should be switched.
• If the switch is to be blocked by an error, then the plugin can either return the first argument with the value None and the second argument with the blocking reason. If there is no reason to block passed (bool(reason) is False) then Blocked by hook is set. This will cause throwing a ProcessingError(reason) exception. Alternatively, this exception can be raised directly from the callback.
• If an exception other than ProcessingError occurs during processing in the callback, then the agent is switched to the originally requested state and the backtrace is logged.

"""Agent status change hook example"""

import time

from constants import AGENT_READY, AGENT_AUX
from error import ProcessingError

UCS_PLUGIN = 'StatusHook'


class StatusHook:
   """Agent status change hook example"""

   def __init__(self, ucs):
     self.ucs = ucs
     self.ucs.agents.addChangeStatusHook(self.hook)

   def stop(self):
     """Called on plugin reload or UCS stop"""
     self.ucs.agents.removeChangeStatusHook()

   def hook(self, agent, new_status, new_reason):
     """Called when agent status is about to change"""

     hour = time.localtime().tm_hour

     if new_status == AGENT_AUX and new_reason == 'Lunch':
       if hour < 11:
         return None, 'It's too early for lunch'

       if hour > 17:
         raise ProcessingError('It's time for dinner')

     if agent.status == AGENT_OFFLINE and new_status == AGENT_READY:
       return AGENT_AUX, 'Getting to know the operating instructions'

     return new_status, new_reason

Shared data change hook (v3.1.0)

The plugin can modify the set data into shared data. UCS provides a method to register the callback ucs.shared.addChangeHook(identifier, callback) and for its deregistration ucs.agents.removeChangeHook(identifier) when stopping the plugin. Only one plugin can use this method for a given identifier, if o another plugin attempts to register, then a ProcessingError exception is thrown.

Three arguments are passed to the callback:

• identifier:str is the identifier of the changed shared data.
• old_data:any is the original data.
• new_data: any are newly set data.

The callback can return arbitrary data or throw a ProcessingError exception if the data change is to be blocked. If another exception occurs during processing in the callback than ProcessingError, then the data is updated as requested and is logged backtrace except in callback.

"""Shared data change hook example"""

from error import ProcessingError

UCS_PLUGIN = 'SharedDataHook'


class SharedDataHook:
   """Shared data change hook example"""

   def __init__(self, ucs):
     self.ucs = ucs
     self.ucs.shared.addChangeHook('agent_emails', self.hook)

   def stop(self):
     """Called on plugin reload or UCS stop"""
     self.ucs.shared.removeChangeHook('agent_emails')

   def hook(self, identifier, old_data, new_data):
     """Called when new shared data are about to change"""

     if not isinstance(new_data, dict):
       raise ProcessingError('Only associative array is accepted')

     data = {}
     username2id = {
       username: user_id
       for user_id, username
       in self.ucs.tree.getUsers('username').items()
     }
     for username, count in new_data.items():
       user_id = username2id.get(username)
       if user_id:
         data[user_id] = count
         old_count = old_data.get(user_id)
         if old_count is not None:
           self.ucs.log.dbug('User ID %d email count changed from %s to %s', user_id, old_count, count)
       otherwise:
         self.ucs.log.warn('Unknown username: %s', username)

     return data

API middleware (v3.8.0)

Individual API requests can be handled using middleware. UCS provides a method for middleware registration ucs.api.addMiddleware(method, middleware) and for its deregistration ucs.api.removeMiddleware(method, middleware) when stopping the plugin. Multiple middlewares can be registered for one API method. Middleware when calling gets a handler as its first argument (which is the API method itself or another middleware) which is followed by arguments passed as API call parameters. Middleware can modify the input arguments, then must call the passed handler or throw an exception. Returns the result of a built-in UCS API method, which it can modify.

"""Middleware plugin example"""


from error import InvalidDataInput, PermissionDenied

UCS_PLUGIN = "Middleware"


class Middleware:
     """Middleware plugin example"""

     def __init__(self, ucs):
         self.ucs = ucs
         self.ucs.api.addMiddleware("config.users.get", self.get_user)
         self.ucs.api.addMiddleware("config.users.edit", self.edit_user)
         self.ucs.api.addMiddleware("config.users.delete", self.delete_user)

     def stop(self):
         """Cleanup on UCS shutdown"""
         self.ucs.api.removeMiddleware("config.users.get", self.get_user)
         self.ucs.api.removeMiddleware("config.users.edit", self.edit_user)
         self.ucs.api.removeMiddleware("config.users.delete", self.delete_user)

     def get_user(self, handler, *args):
       """Just log middleware usage"""
       self.ucs.log.dev("Middleware get_user args: %s", args)
       result = handler(*args)
       self.ucs.log.dev("Middleware get_user result: %s", result)
       return result

     def edit_user(self, handler, user, data):
         """Keep username and password"""
         if data["id"] == 1:
             if "username" in data and data["username"] != "admin":
                 raise InvalidDataInput({"username": "Admin username can't be changed"})

             if "password" in data and data["password"] and len(data["password"]) < 10:
                 raise InvalidDataInput({"password": "Admin password is too short"})

         return handler(user, data)

     def delete_user(self, *args):
         """Disable user deletion"""
         raise PermissionDenied("This UCS forbids user removal")

Call metadata transformation (v4.38.0)

Metadata passed through the cdr.metadata API method can be modified using a callback. A metadata transformer is a function that accepts the uniqueid of the call for which the metadata is being set, and the metadata itself, which it modifies: Callable[[str, dict[str, Any]], None].

The transformer is registered using: ucs.api.cdr.set_metadata_transformer(callback) and removed using: ucs.api.cdr.unset_metadata_transformer()

Example of a plugin that transforms metadata {"call_code": ["Warranty claim", "Out-of-warranty"]} into {"accountcode": "Warranty claim|Out-of-warranty"}:

"""Transform metadata for CDR"""
from typing import Any

class MetadataTransformer:
    """Transform metadata for CDR"""

    def __init__(self, ucs):
        self.__ucs = ucs
        self.__ucs.api.cdr.set_metadata_transformer(self.transform)

    def stop(self):
        """Remove callback"""
        self.__ucs.api.cdr.unset_metadata_transformer()

    def transform(self, uniqueid: str, metadata: dict[str, Any]) -> None:
        """Transform call_code list to accountcode str"""
        call_code = metadata.pop("call_code", None)
        if call_code is not None:
            metadata["accountcode"] = "|".join(call_code)

Caller Lookup Transformation (v4.39.0)

UCS allows developers to modify the result of a caller-name lookup performed on an incoming phone number. A lookup transformer is a callback function that receives three arguments:

• lookup_id – identifier of the lookup provider that executed the search
• number – the caller’s phone number
• data – the lookup result as a dictionary

The callback must follow this signature:

Callable[[int, str, dict[str, Any]], dict[str, Any]]

Register a custom transformer:

ucs.lookups.set_lookup_transformer(callback)

Unregister it:

ucs.lookups.unset_lookup_transformer()

If the transformer raises an exception, UCS automatically falls back to the original (unmodified) lookup data.

Plugin Example

The following plugin prefixes the caller’s name with a question mark if the caller is associated with multiple companies. Given this lookup result:

{"name": "Our Customer", "companies": ["First", "Second"]}

it will be transformed into:

{"name": "?Our Customer", "companies": ["First", "Second"]}

---

from typing import Any

class LookupTransformer:
    def __init__(self, ucs):
        self.__ucs = ucs
        self.__ucs.lookups.set_lookup_transformer(self.transform)

    def stop(self):
        """Remove callback"""
        self.__ucs.lookups.unset_lookup_transformer()

    def transform(self, lookup_id: int, number: str, data: dict[str, Any]) -> dict[str, Any]:
        """Prefix caller name if they have multiple companies"""
        if not data:
            return {"name": "Dang!"}

        prefix = "?" if len(data.get("companies", [])) > 1 else ""
        data["name"] = f"{prefix}{data['name']}"
        return data