#!/usr/bin/env python3
import argparse, json, requests, sys
# --- INFORMATION ---
# This will replace ALL queue agents in the target queue
# --- CONFIGURATION ---
fqdn = None
QUEUE_NUMBER = 800
QUEUE_AGENTS = [103, 104, 105, 106]
def get_token(client_id, client_secret):
url = f"https://{fqdn}/connect/token"
data = {
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret
}
response = requests.post(url, data=data)
response.raise_for_status()
return response.json()["access_token"]
def request_headers():
hh = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
return hh
def get_Queues(token):
url = f"https://{fqdn}/xapi/v1/Queues"
params = {
'$select': 'Id,Number'
}
response = requests.get(url, params=params, headers=request_headers())
if response.ok:
response_data = response.json()
else:
print(f"Response Code: {response.status_code}\nResponse Text: {response.text}")
error_report = {
"Error": [
{
"Response Code": response.status_code,
"Response Body": response.text
}
]
}
return error_report, False
return response_data, True
def patch_Queues(queue_id, payload, token):
url = f"https://{fqdn}/xapi/v1/Queues({queue_id})"
try:
response = requests.patch(url, data=payload, headers=request_headers())
response.raise_for_status()
if not response.text.strip():
print("Success! (Status {response.status_code}, No body returned)")
else:
result=response.json()
print("Success! Response body:", result)
except requests.exceptions.HTTPError as err:
print(f"HTTP error occurred: {err}")
print(f"Response Body: {response.text}")
except requests.exceptions.JSONDecodeError:
print("Success, but the response was not valid JSON.")
def main():
global fqdn, token
parser = argparse.ArgumentParser()
# Define arguments
parser.add_argument("--fqdn", required=True)
parser.add_argument("--client_id", required=True)
parser.add_argument("--client_secret", required=True)
args = parser.parse_args()
fqdn = args.fqdn
token = get_token(args.client_id, args.client_secret)
data, success = get_Queues(token)
if not success:
print("Stopping process due to error:")
print(json.dumps(data, indent=2))
sys.exit()
queues_list = data["value"]
target_id = next((item['Id'] for item in queues_list if item['Number'] == str(QUEUE_NUMBER)), None)
if target_id is None:
print(f"Cannot find Queue with number {QUEUE_NUMBER}")
sys.exit()
print(f"Queue with Number {str(QUEUE_NUMBER)} has ID {target_id}")
payload_string = '{ "Number": "800", "Agents": [ '
for agent in QUEUE_AGENTS:
if payload_string.endswith("}"):
payload_string += ", "
payload_string += '{ "Number": "' + str(agent) + '" }'
payload_string += ' ] }'
print(f"Sending Payload:\n{payload_string}")
patch_Queues(target_id, payload_string, token)
# --- EXECUTION ---
if __name__ == "__main__":
main()