Effi/docs/TOOL-LEANTIME-OPENWEBUI.md

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

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