Hallo Markus,

es gibt ja in der Paketverwaltung „Third Party Addons“ das Addon „Ayte-TicketLoadBalancer“. Eventuell kannst du damit deine Anforderung umsetzen.

Alternativ per Rest Api (Webservice) ein eigenes Tool schreiben.

Den Workflow / Verteilung so nachbauen, wie man es auch händisch machen würde, um dei Anforderung zu erfüllen.

Gruß Marcel

Ich würde es über Webservice machen.

Kannst dir ja mal das Tool „Postman“ installieren. Darin kannst du dann den Webservice testen.

Unter Admin-Agent, dann am besten noch einen User „Webservice“ anlegen, der dann auch die entsprechenden Queues RW Rechte hat.  Dann im Postman Tool entsprechend den Webservice einrichten.
http://serverURL/otobo/nph-genericinterface.pl/Webservice/TicketSearch/TicketSearch

{
  "UserLogin": "dein_user",
  "Password": "dein_password",
  "Queues": ["Eingang"],
  "States": ["new"]
}

Ich würde es vorher in einem Otobo Testsystem einmal testen.

Oben der erste Teil zum auslesen neuer Tickets.

Nun braucht es noch den Teil zum verschieben der Tickets.

Um ein Ticket über den Webservice in eine andere Queue zu verschieben, nutzt du die Operation Ticket::TicketUpdate.
Hier ist das passende JSON-Format für deinen REST-Request:

Der JSON-Body zum Verschieben

json
{
„UserLogin“: „dein_api_user“,
„Password“: „dein_password“,
„TicketID“: 123,
„Ticket“: {
„Queue“: „ZIEL_QUEUE_NAME“
}
}

Diese Codesnipsel dann einfach in einer Anwendung zusammen bringen, um die einzelnen Queues zu bedienen.

Bei der Verteilungsregel wäre dann die Frage, wie man das Rechnerisch umsetzt, um die Zuteilung fair zu gestalten.

Eventuell hat aber noch jemand eine andere Idee.

Gruß Marcel

folgend mal ein Python script (ungetstet) welches die Aufgabe übernimmt

##################################

import requests
import time

# OTOBO REST-Konfiguration
BASE_URL = „https://dein-otobo.de“
USER_AUTH = {
„UserLogin“: „api_user“,
„Password“: „api_password“
}

SOURCE_QUEUE = „Eingang“
TARGET_QUEUES = [„Queue_A“, „Queue_B“, „Queue_C“]

def get_new_tickets():
„““Sucht nach neuen Ticket-IDs in der Quell-Queue.“““
url = f“{BASE_URL}/TicketSearch“
payload = {**USER_AUTH, „Queues“: [SOURCE_QUEUE], „States“: [„new“]}

try:
response = requests.post(url, json=payload)
response.raise_for_status()
return response.json().get(„TicketID“, [])
except Exception as e:
print(f“Fehler bei Suche: {e}“)
return []

def process_and_move(ticket_id, target_queue):
„““Verarbeitet das Ticket (Platzhalter) und verschiebt es in OTOBO.“““
# 1. Ticket-Details abrufen (falls Daten für die Verarbeitung gebraucht werden)
# details_url = f“{BASE_URL}/TicketGet/{ticket_id}“

# 2. HIER: Deine Logik (z.B. Daten in DB schreiben, E-Mail senden, etc.)
print(f“Verarbeite Ticket {ticket_id}…“)

# 3. Ticket in die Ziel-Queue verschieben
update_url = f“{BASE_URL}/TicketUpdate/{ticket_id}“
update_payload = {
**USER_AUTH,
„Ticket“: {
„Queue“: target_queue,
„State“: „open“ # Optional: Status direkt auf ‚offen‘ setzen
}
}

try:
res = requests.patch(update_url, json=update_payload)
if res.status_code == 200:
print(f“Ticket {ticket_id} erfolgreich nach {target_queue} verschoben.“)
except Exception as e:
print(f“Fehler beim Verschieben von {ticket_id}: {e}“)

# Hauptschleife
current_index = 0
while True:
print(f“\n[{time.strftime(‚%H:%M:%S‘)}] Prüfe auf neue Tickets…“)
tickets = get_new_tickets()

if tickets:
print(f“{len(tickets)} Tickets gefunden. Starte Verteilung…“)
for t_id in tickets:
target = TARGET_QUEUES[current_index]
process_and_move(t_id, target)

# Round-Robin Index weiterschalten
current_index = (current_index + 1) % len(TARGET_QUEUES)
else:
print(„Keine neuen Tickets.“)

time.sleep(120) # 2 Minuten Pause

#####################################

Prima, wenn es erstmal klappt. Wobei bei 10 Generic Tasks die Aufteilung auf 3 Queues da auch nicht 100% stimmt :)

Wir nutzen zum Beispiel ein selbst entwickletes IT-Helpdesktool, hier werden die Tickets per Webservice erstellt.

Früher hatten wir darüber die Tickets per SMTP versendet und Otobo hatte die wieder abgeholt. Per Webservice geht das nun alles ohne den Umweg.

Hallo Markus,

das folgenden Python Script „otobo_distributor.py“ schaut in der Source Queue nach neuen Tickets und schiebt diese dann automatisch in die 3 Ziel Queues.
Das Script einfach auf den Otobo Server kopieren (mit Winscp zum Beispiel),  R/W/X Rechte vergeben, in den Pfad wechseln und das Script dann mit dem Befehl „python3 otobo_distributor.py“ per per SSH Konsole starten.

STRG+C zum beenden oder pkill -f otobo_distributor.py (in einem anderen Terminal Fenster)

#############################

import requests
import time
import sys

# — Konfiguration —
BASE_URL = „http://serverURL/otobo/nph-genericinterface.pl/Webservice/“
USER_LOGIN = „webservice_user“
PASSWORD = „webservice_pass“

SOURCE_QUEUE = „it-support“
TARGET_QUEUE_IDS = [11, 12, 13]
CURRENT_INDEX = 0

def get_new_tickets():
„““Sucht nach neuen Tickets in der Quell-Queue.“““
url = f“{BASE_URL}TicketSearch/TicketSearch“
payload = {
„UserLogin“: USER_LOGIN,
„Password“: PASSWORD,
„Queues“: [SOURCE_QUEUE],
„States“: [„new“]
}

try:
response = requests.post(url, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
# OTOBO gibt die IDs meist als Liste im Feld „TicketID“ zurück
return data.get(„TicketID“, [])
except Exception as e:
print(f“Fehler bei der Ticket-Suche: {e}“)
return []

def process_and_move(ticket_id, queue_id):
„““Verschiebt ein Ticket in eine neue QueueID via PATCH.“““
url = f“{BASE_URL}TicketUpdate/TicketUpdate“
payload = {
„UserLogin“: USER_LOGIN,
„Password“: PASSWORD,
„TicketID“: ticket_id,
„Ticket“: {
„QueueID“: queue_id
}
}

try:
response = requests.patch(url, json=payload, timeout=30)
return response.status_code == 200
except Exception as e:
print(f“Fehler beim Verschieben von Ticket {ticket_id}: {e}“)
return False

# — Hauptschleife —
if __name__ == „__main__“:
print(f“[{time.strftime(‚%H:%M:%S‘)}] Skript gestartet. Überwachung von: {SOURCE_QUEUE}“)

try:
while True:
tickets = get_new_tickets()

if tickets and isinstance(tickets, list):
print(f“[{time.strftime(‚%H:%M:%S‘)}] >> {len(tickets)} Tickets gefunden.“)

for t_id in tickets:
target_id = TARGET_QUEUE_IDS[CURRENT_INDEX]

if process_and_move(t_id, target_id):
print(f“OK: Ticket {t_id} wurde in QueueID {target_id} verschoben.“)
else:
print(f“FEHLER: Ticket {t_id} konnte nicht verschoben werden.“)

# Round-Robin Index berechnen
CURRENT_INDEX = (CURRENT_INDEX + 1) % len(TARGET_QUEUE_IDS)
else:
# Optional: Konsolenausgabe für „Leerlauf“
sys.stdout.write(„.“)
sys.stdout.flush()

time.sleep(120) # 2 Minuten warten

except KeyboardInterrupt:
print(„\nSkript durch Benutzer beendet.“)
<h6>#######################</h6>
Gruß Marcel

Damit das Script läuft, müssen im Otobo die Webservices (TicketSearch und TicketUpdate) angelegt und konfiguriert werden. Und Im Skript müssen die Paramter dann auch entsprechend hinterlegt werden.

Gruß Marcel

Hallo Markus,

wir nutzen das Otobo auch über Docker.

Schau dir doch auch mal den folgenden Post an, da kannst du dir über einen VM Clone ein Testsystem aufsetzen.

https://otobo.io/forums/topic/otobo-testsystem-from-vm-clone/

Gruß Marcel