# 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)