Folzo/Effi
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Effi/docs/TOOL-LEANTIME-OPENWEBUI.md
T
Folzo 873432362a 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
2026-05-10 22:28:28 +02:00

379 lines
16 KiB
Markdown
Raw Permalink Blame History

# 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 View Git Blame Copy Permalink