diff --git a/docs/TOOL-LEANTIME-OPENWEBUI.md b/docs/TOOL-LEANTIME-OPENWEBUI.md new file mode 100644 index 0000000..8f66e51 --- /dev/null +++ b/docs/TOOL-LEANTIME-OPENWEBUI.md @@ -0,0 +1,379 @@ +# Leantime Tool for Open WebUI + +## Description + +Permet a Effi d'interagir avec Leantime (gestion de projet) directement depuis Open WebUI, sans passer par MCPO. + +## Installation + +1. **Admin Panel** → **Tools** → **Create Tool** (ou +) +2. **Name** : `Leantime` +3. **Description** : `Gestion de projet Leantime. Commandes: list_projects(), get_project(id), create_project(name), list_tickets(project_id), create_ticket(headline, project_id), update_ticket(ticket_id, project_id), get_ticket(id), add_comment(module, module_id, comment), list_users(), get_user(id), get_status_labels()` +4. **Coller le code ci-dessous** +5. **Save and Activate** +6. **Configurer les Valves** via la roue crantee : + - `leantime_url` : `http://192.168.1.25:8380` + - `leantime_api_key` : votre cle API (Company > API Keys dans Leantime) + +## Tool Code + +```python +""" +Leantime Tool for Open WebUI +Gestion de projet Leantime via API JSON-RPC 2.0 +""" +import json +import requests +from pydantic import BaseModel, Field + + +class Tools: + class Valves(BaseModel): + leantime_url: str = Field( + default="http://192.168.1.25:8380", + description="URL de Leantime", + ) + leantime_api_key: str = Field( + default="", + description="Cle API Leantime (Company > API Keys)", + ) + + def __init__(self): + self.valves = self.Valves() + self._request_id = 0 + + def _call(self, method: str, params: dict = None) -> dict: + if not self.valves.leantime_api_key: + return {"error": "[ERROR] Cle API Leantime non configuree. Ajoutez-la dans les Valves du Tool."} + url = f"{self.valves.leantime_url.rstrip('/')}/api/jsonrpc" + self._request_id += 1 + payload = { + "jsonrpc": "2.0", + "method": method, + "params": params or {}, + "id": self._request_id, + } + headers = { + "Content-Type": "application/json", + "X-API-KEY": self.valves.leantime_api_key, + } + try: + response = requests.post(url, json=payload, headers=headers, timeout=15) + response.raise_for_status() + data = response.json() + if "error" in data: + return {"error": f"[ERROR] API Leantime: {data['error'].get('message', data['error'])}"} + return data.get("result") + except requests.exceptions.ConnectionError: + return {"error": f"[ERROR] Impossible de joindre Leantime a {self.valves.leantime_url}. Verifiez l'URL et que Leantime est en marche."} + except requests.exceptions.Timeout: + return {"error": "[ERROR] Timeout: Leantime met trop longtemps a repondre."} + except Exception as e: + return {"error": f"[ERROR] {type(e).__name__}: {e}"} + + def list_projects(self) -> str: + """Liste tous les projets Leantime accessibles. + Retourne la liste des projets avec leur ID, nom, statut et dates.""" + result = self._call("leantime.rpc.Projects.getAll") + if isinstance(result, dict) and "error" in result: + return result["error"] + if not result: + return "[INFO] Aucun projet trouve." + if isinstance(result, list): + projects = [] + for p in result: + projects.append(f"- ID {p.get('id')}: {p.get('name', 'Sans nom')} | Statut: {p.get('state', 'N/A')} | Du {p.get('start', '?')} au {p.get('end', '?')}") + return "\n".join(projects) + return json.dumps(result, indent=2, ensure_ascii=False) + + def get_project(self, project_id: int) -> str: + """Recupere les details d'un projet Leantime par son ID. + :param project_id: ID du projet + :return: Details du projet""" + result = self._call("leantime.rpc.Projects.getProject", {"id": project_id}) + if isinstance(result, dict) and "error" in result: + return result["error"] + if not result: + return f"[INFO] Projet {project_id} non trouve." + name = result.get("name", "Sans nom") + details = result.get("details", "Aucun detail") + state = result.get("state", "N/A") + start = result.get("start", "?") + end = result.get("end", "?") + return f"Projet #{project_id}: {name}\nStatut: {state}\nPeriode: {start} → {end}\nDetails: {details}" + + def create_project(self, name: str, details: str = "") -> str: + """Cree un nouveau projet Leantime. + :param name: Nom du projet + :param details: Description du projet (optionnel) + :return: Confirmation de creation""" + params = {"name": name} + if details: + params["details"] = details + result = self._call("leantime.rpc.Projects.addProject", params) + if isinstance(result, dict) and "error" in result: + return result["error"] + return f"[OK] Projet cree: {name} (ID: {result.get('id', '?') if isinstance(result, dict) else result})" + + def list_tickets(self, project_id: int = None) -> str: + """Liste les tickets/taches Leantime, optionnellement filtres par projet. + :param project_id: ID du projet pour filtrer (optionnel) + :return: Liste des tickets""" + params = {} + if project_id: + params = {"searchCriteria": {"currentProject": project_id}} + result = self._call("leantime.rpc.Tickets.Tickets.getAll", params) + if isinstance(result, dict) and "error" in result: + return result["error"] + if not result: + return "[INFO] Aucun ticket trouve." + if isinstance(result, list): + tickets = [] + for t in result: + status = t.get("status", "?") + priority = t.get("priority", "?") + headline = t.get("headline", "Sans titre") + tid = t.get("id", "?") + tickets.append(f"- #{tid}: {headline} | Statut: {status} | Priorite: {priority}") + return "\n".join(tickets) + return json.dumps(result, indent=2, ensure_ascii=False) + + def get_ticket(self, ticket_id: int) -> str: + """Recupere les details d'un ticket Leantime par son ID. + :param ticket_id: ID du ticket + :return: Details du ticket""" + result = self._call("leantime.rpc.Tickets.Tickets.getTicket", {"id": ticket_id}) + if isinstance(result, dict) and "error" in result: + return result["error"] + if not result: + return f"[INFO] Ticket {ticket_id} non trouve." + headline = result.get("headline", "Sans titre") + description = result.get("description", "Aucune description") + status = result.get("status", "?") + priority = result.get("priority", "?") + project_id = result.get("projectId", "?") + return f"Ticket #{ticket_id}: {headline}\nProjet: {project_id}\nStatut: {status} | Priorite: {priority}\nDescription: {description}" + + def create_ticket(self, headline: str, project_id: int, description: str = "", priority: str = "4", tags: str = "") -> str: + """Cree un nouveau ticket/tache dans Leantime. + :param headline: Titre du ticket + :param project_id: ID du projet dans lequel creer le ticket + :param description: Description du ticket (optionnel) + :param priority: Priorite (3=haute, 4=moyenne, 5=basse, defaut: 4) + :param tags: Tags separes par des virgules (optionnel) + :return: Confirmation de creation""" + from datetime import datetime + values = { + "headline": headline, + "projectId": project_id, + "userId": 1, + "date": datetime.now().strftime("%Y-%m-%d"), + "priority": priority, + } + if description: + values["description"] = description + if tags: + values["tags"] = tags + result = self._call("leantime.rpc.Tickets.Tickets.addTicket", {"values": values}) + if isinstance(result, dict) and "error" in result: + return result["error"] + return f"[OK] Ticket cree: {headline} dans le projet {project_id}" + + def update_ticket(self, ticket_id: int, project_id: int, headline: str = None, description: str = None, status: int = None, priority: str = None) -> str: + """Met a jour un ticket existant dans Leantime. + :param ticket_id: ID du ticket a modifier + :param project_id: ID du projet du ticket + :param headline: Nouveau titre (optionnel) + :param description: Nouvelle description (optionnel) + :param status: Nouveau statut (optionnel, ex: 3=done, 4=waiting, etc.) + :param priority: Nouvelle priorite (optionnel) + :return: Confirmation de mise a jour""" + values = {"id": ticket_id, "projectId": project_id} + if headline is not None: + values["headline"] = headline + if description is not None: + values["description"] = description + if status is not None: + values["status"] = status + if priority is not None: + values["priority"] = priority + result = self._call("leantime.rpc.Tickets.Tickets.updateTicket", {"values": values}) + if isinstance(result, dict) and "error" in result: + return result["error"] + return f"[OK] Ticket #{ticket_id} mis a jour." + + def add_comment(self, module: str, module_id: int, comment: str) -> str: + """Ajoute un commentaire a un module Leantime (ticket, projet, etc.). + :param module: Type de module ('ticket' ou 'project') + :param module_id: ID du module + :param comment: Texte du commentaire + :return: Confirmation d'ajout""" + result = self._call("leantime.rpc.Comments.addComment", { + "module": module, + "moduleId": module_id, + "comment": comment, + }) + if isinstance(result, dict) and "error" in result: + return result["error"] + return f"[OK] Commentaire ajoute au {module} #{module_id}." + + def get_comments(self, module: str, module_id: int) -> str: + """Recupere les commentaires d'un module Leantime. + :param module: Type de module ('ticket' ou 'project') + :param module_id: ID du module + :return: Liste des commentaires""" + result = self._call("leantime.rpc.Comments.getComments", { + "module": module, + "moduleId": module_id, + }) + if isinstance(result, dict) and "error" in result: + return result["error"] + if not result: + return f"[INFO] Aucun commentaire sur {module} #{module_id}." + if isinstance(result, list): + comments = [] + for c in result: + author = c.get("author", "Inconnu") + text = c.get("comment", "") + date = c.get("date", "") + comments.append(f"- [{author}] ({date}): {text}") + return "\n".join(comments) + return json.dumps(result, indent=2, ensure_ascii=False) + + def list_users(self) -> str: + """Liste tous les utilisateurs Leantime. + :return: Liste des utilisateurs""" + result = self._call("leantime.rpc.Users.getAll") + if isinstance(result, dict) and "error" in result: + return result["error"] + if not result: + return "[INFO] Aucun utilisateur trouve." + if isinstance(result, list): + users = [] + for u in result: + uid = u.get("id", "?") + name = f"{u.get('firstname', '')} {u.get('lastname', '')}".strip() + email = u.get("username", u.get("email", "?")) + users.append(f"- ID {uid}: {name} ({email})") + return "\n".join(users) + return json.dumps(result, indent=2, ensure_ascii=False) + + def get_user(self, user_id: int) -> str: + """Recupere les details d'un utilisateur Leantime par son ID. + :param user_id: ID de l'utilisateur + :return: Details de l'utilisateur""" + result = self._call("leantime.rpc.Users.getUser", {"id": user_id}) + if isinstance(result, dict) and "error" in result: + return result["error"] + if not result: + return f"[INFO] Utilisateur {user_id} non trouve." + name = f"{result.get('firstname', '')} {result.get('lastname', '')}".strip() + email = result.get("username", result.get("email", "?")) + role = result.get("role", "?") + return f"Utilisateur #{user_id}: {name} ({email})\nRole: {role}" + + def get_status_labels(self) -> str: + """Recupere les labels de statut disponibles pour les tickets. + :return: Dictionnaire des statuts ID → label""" + result = self._call("leantime.rpc.Tickets.Tickets.getStatusLabels") + if isinstance(result, dict) and "error" in result: + return result["error"] + if isinstance(result, dict): + lines = [f"- {k}: {v}" for k, v in result.items()] + return "Statuts disponibles:\n" + "\n".join(lines) + return json.dumps(result, indent=2, ensure_ascii=False) + + def add_timesheet(self, user_id: int, ticket_id: int, hours: float, date: str, description: str = "") -> str: + """Ajoute une entree de timesheet. + :param user_id: ID de l'utilisateur + :param ticket_id: ID du ticket + :param hours: Nombre d'heures (ex: 2.5) + :param date: Date au format YYYY-MM-DD + :param description: Description (optionnel) + :return: Confirmation d'ajout""" + params = { + "userId": user_id, + "ticketId": ticket_id, + "hours": hours, + "date": date, + } + if description: + params["description"] = description + result = self._call("leantime.rpc.Timesheets.addTime", params) + if isinstance(result, dict) and "error" in result: + return result["error"] + return f"[OK] Timesheet ajoute: {hours}h sur ticket #{ticket_id} le {date}." + + def get_timesheets(self, project_id: int = None, user_id: int = None) -> str: + """Recupere les entrees de timesheet, filtrees par projet et/ou utilisateur. + :param project_id: ID du projet (optionnel) + :param user_id: ID de l'utilisateur (optionnel) + :return: Liste des timesheets""" + params = {} + if project_id: + params["projectId"] = project_id + if user_id: + params["userId"] = user_id + result = self._call("leantime.rpc.Timesheets.getTimesheets", params) + if isinstance(result, dict) and "error" in result: + return result["error"] + if not result: + return "[INFO] Aucun timesheet trouve." + return json.dumps(result, indent=2, ensure_ascii=False) + + +def list_projects() -> str: + t = Tools() + return t.list_projects() + +def get_project(project_id: int) -> str: + t = Tools() + return t.get_project(project_id) + +def create_project(name: str, details: str = "") -> str: + t = Tools() + return t.create_project(name, details) + +def list_tickets(project_id: int = None) -> str: + t = Tools() + return t.list_tickets(project_id) + +def get_ticket(ticket_id: int) -> str: + t = Tools() + return t.get_ticket(ticket_id) + +def create_ticket(headline: str, project_id: int, description: str = "", priority: str = "4", tags: str = "") -> str: + t = Tools() + return t.create_ticket(headline, project_id, description, priority, tags) + +def update_ticket(ticket_id: int, project_id: int, headline: str = None, description: str = None, status: int = None, priority: str = None) -> str: + t = Tools() + return t.update_ticket(ticket_id, project_id, headline, description, status, priority) + +def add_comment(module: str, module_id: int, comment: str) -> str: + t = Tools() + return t.add_comment(module, module_id, comment) + +def get_comments(module: str, module_id: int) -> str: + t = Tools() + return t.get_comments(module, module_id) + +def list_users() -> str: + t = Tools() + return t.list_users() + +def get_user(user_id: int) -> str: + t = Tools() + return t.get_user(user_id) + +def get_status_labels() -> str: + t = Tools() + return t.get_status_labels() + +def add_timesheet(user_id: int, ticket_id: int, hours: float, date: str, description: str = "") -> str: + t = Tools() + return t.add_timesheet(user_id, ticket_id, hours, date, description) + +def get_timesheets(project_id: int = None, user_id: int = None) -> str: + t = Tools() + return t.get_timesheets(project_id, user_id) \ No newline at end of file