feat: add Leantime Open WebUI Tool (direct API, no MCPO needed)
- Tool with Valves (leantime_url, leantime_api_key) - 14 functions: projects, tickets, comments, users, timesheets, status labels - JSON-RPC 2.0 client matching Leantime API - Activate/deactivate per conversation in Open WebUI
This commit is contained in:
Folzo
@@ -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)
|
||||
Reference in New Issue
Block a user