diff --git a/docs/TOOL-JOPLIN-OPENWEBUI.md b/docs/TOOL-JOPLIN-OPENWEBUI.md new file mode 100644 index 0000000..e33f1f0 --- /dev/null +++ b/docs/TOOL-JOPLIN-OPENWEBUI.md @@ -0,0 +1,371 @@ +# Joplin Tool for Open WebUI + +## Description + +Permet a Effi de lire et ecrire des notes dans Joplin via l'API Clipper (locale). Gestion de notes, notebooks et TODOs directement depuis le chat. + +**Important** : L'API Clipper fonctionne uniquement quand l'app Joplin desktop est ouverte et le Web Clipper active (Options → Web Clipper). + +## Installation + +1. **Admin Panel** → **Tools** → **Create Tool** (ou +) +2. **Name** : `Joplin` +3. **Description** : `Notes Joplin. Commandes: list_notebooks(), list_notes(notebook_id), get_note(id), create_note(title, body, notebook_id), update_note(id, title, body), search_notes(query), create_notebook(name), add_todo(note_id, text), toggle_todo(note_id, completed)` +4. **Coller le code ci-dessous** +5. **Save and Activate** +6. **Configurer les Valves** via la roue crantee : + - `joplin_url` : `http://192.168.1.11:41184` (IP du PC ou Joplin est ouvert) + - `joplin_api_token` : votre jeton API (Options > Web Clipper > Jeton d'authentification) + +## Tool Code + +```python +""" +Joplin Tool for Open WebUI +Lire et ecrire des notes Joplin via l'API Clipper +API doc: https://joplinapp.org/help/api/references/rest_api/ +""" +import json +import requests +from pydantic import BaseModel, Field + + +class Tools: + class Valves(BaseModel): + joplin_url: str = Field( + default="http://192.168.1.11:41184", + description="URL de l'API Clipper Joplin (IP:port du PC ou Joplin est ouvert)", + ) + joplin_api_token: str = Field( + default="", + description="Jeton d'authentification (Options > Web Clipper dans Joplin)", + ) + + def __init__(self): + self.valves = self.Valves() + + def _url(self, path): + return f"{self.valves.joplin_url.rstrip('/')}/{path}" + + def _params(self, extra=None): + p = {"token": self.valves.joplin_api_token} + if extra: + p.update(extra) + return p + + def _check(self): + if not self.valves.joplin_api_token: + return "[ERROR] Jeton API Joplin non configure. Ajoutez-le dans les Valves du Tool (Options > Web Clipper dans Joplin)." + + def _get(self, path, params=None): + err = self._check() + if err: + return {"error": err} + try: + r = requests.get(self._url(path), params=self._params(params), timeout=15) + r.raise_for_status() + return r.json() + except requests.exceptions.ConnectionError: + return {"error": f"[ERROR] Impossible de joindre Joplin. Verifiez que l'app Joplin est ouverte et le Web Clipper active sur {self.valves.joplin_url}"} + except requests.exceptions.HTTPError as e: + if e.response.status_code == 403: + return {"error": "[ERROR] Acces refuse. Verifiez votre jeton API Joplin dans les Valves."} + return {"error": f"[ERROR] HTTP {e.response.status_code}: {e.response.text}"} + except Exception as e: + return {"error": f"[ERROR] {type(e).__name__}: {e}"} + + def _post(self, path, body, params=None): + err = self._check() + if err: + return {"error": err} + try: + r = requests.post(self._url(path), params=self._params(params), json=body, timeout=15) + r.raise_for_status() + return r.json() + except requests.exceptions.ConnectionError: + return {"error": f"[ERROR] Impossible de joindre Joplin. Verifiez que l'app est ouverte sur {self.valves.joplin_url}"} + except requests.exceptions.HTTPError as e: + if e.response.status_code == 403: + return {"error": "[ERROR] Acces refuse. Verifiez votre jeton API Joplin dans les Valves."} + return {"error": f"[ERROR] HTTP {e.response.status_code}: {e.response.text}"} + except Exception as e: + return {"error": f"[ERROR] {type(e).__name__}: {e}"} + + def _put(self, path, body, params=None): + err = self._check() + if err: + return {"error": err} + try: + r = requests.put(self._url(path), params=self._params(params), json=body, timeout=15) + r.raise_for_status() + return r.json() + except requests.exceptions.ConnectionError: + return {"error": f"[ERROR] Impossible de joindre Joplin. Verifiez que l'app est ouverte sur {self.valves.joplin_url}"} + except requests.exceptions.HTTPError as e: + if e.response.status_code == 403: + return {"error": "[ERROR] Acces refuse. Verifiez votre jeton API Joplin dans les Valves."} + return {"error": f"[ERROR] HTTP {e.response.status_code}: {e.response.text}"} + except Exception as e: + return {"error": f"[ERROR] {type(e).__name__}: {e}"} + + def _delete(self, path): + err = self._check() + if err: + return {"error": err} + try: + r = requests.delete(self._url(path), params=self._params(), timeout=15) + r.raise_for_status() + return {"ok": True} + except Exception as e: + return {"error": f"[ERROR] {type(e).__name__}: {e}"} + + def ping(self) -> str: + """Verifie que Joplin est accessible. + :return: Statut de la connexion""" + try: + r = requests.get(f"{self.valves.joplin_url.rstrip('/')}/ping", timeout=5) + r.raise_for_status() + return f"[OK] Joplin accessible: {r.text}" + except Exception as e: + return f"[ERROR] Joplin inaccessible sur {self.valves.joplin_url}: {e}" + + def list_notebooks(self) -> str: + """Liste tous les notebooks Joplin avec leur ID. + :return: Liste des notebooks""" + data = self._get("folders", params={"fields": "id,title,parent_id"}) + if isinstance(data, dict) and "error" in data: + return data["error"] + items = data if isinstance(data, list) else data.get("items", []) + if not items: + return "[INFO] Aucun notebook trouve." + # Build tree structure + by_id = {nb["id"]: nb for nb in items} + roots = [] + for nb in items: + parent = nb.get("parent_id", "") + if not parent: + roots.append(nb) + else: + parent_nb = by_id.get(parent) + if parent_nb: + if "children" not in parent_nb: + parent_nb["children"] = [] + parent_nb["children"].append(nb) + lines = [] + def _format(nb, depth=0): + prefix = " " * depth + nid = nb.get("id", "?") + title = nb.get("title", "Sans titre") + lines.append(f"{prefix}- {title} (ID: {nid})") + for child in nb.get("children", []): + _format(child, depth + 1) + for root in roots: + _format(root) + return "\n".join(lines) + + def create_notebook(self, name: str, parent_id: str = "") -> str: + """Cree un nouveau notebook Joplin. + :param name: Nom du notebook + :param parent_id: ID du notebook parent (optionnel, pour creer un sous-notebook) + :return: Confirmation avec ID""" + body = {"title": name} + if parent_id: + body["parent_id"] = parent_id + result = self._post("folders", body) + if isinstance(result, dict) and "error" in result: + return result["error"] + nid = result.get("id", "?") + return f"[OK] Notebook '{name}' cree (ID: {nid}). Il sera synchronise sur tous vos appareils via le serveur Joplin." + + def list_notes(self, notebook_id: str) -> str: + """Liste les notes dans un notebook Joplin. + :param notebook_id: ID du notebook + :return: Liste des notes avec TODOs identifies""" + data = self._get(f"folders/{notebook_id}/notes", params={"fields": "id,title,is_todo,todo_completed,updated_time"}) + if isinstance(data, dict) and "error" in data: + return data["error"] + items = data if isinstance(data, list) else data.get("items", []) + if not items: + return "[INFO] Aucune note dans ce notebook." + lines = [] + for note in items: + nid = note.get("id", "?") + title = note.get("title", "Sans titre") + is_todo = note.get("is_todo", 0) + todo_completed = note.get("todo_completed", 0) + if is_todo: + icon = "[x]" if todo_completed else "[ ]" + lines.append(f" {icon} {title} (ID: {nid})") + else: + lines.append(f" # {title} (ID: {nid})") + return "\n".join(lines) + + def get_note(self, note_id: str) -> str: + """Recupere le contenu complet d'une note Joplin. + :param note_id: ID de la note + :return: Contenu complet de la note en Markdown""" + data = self._get(f"notes/{note_id}", params={"fields": "id,title,body,is_todo,todo_completed,updated_time,parent_id"}) + if isinstance(data, dict) and "error" in data: + return data["error"] + if not data or not data.get("id"): + return f"[INFO] Note {note_id} non trouvee." + title = data.get("title", "Sans titre") + body = data.get("body", "") + is_todo = data.get("is_todo", 0) + updated = data.get("updated_time", "?") + header = f"== {title} == (ID: {data.get('id')})" + if is_todo: + status = "Terminee" if data.get("todo_completed") else "A faire" + header += f" | TODO: {status}" + header += f" | Modifie: {updated}" + return f"{header}\n\n{body}" + + def create_note(self, title: str, body: str = "", notebook_id: str = "", is_todo: bool = False) -> str: + """Cree une nouvelle note Joplin. + :param title: Titre de la note + :param body: Contenu Markdown de la note (optionnel) + :param notebook_id: ID du notebook ou placer la note (optionnel) + :param is_todo: Creer comme tache TODO (default: False) + :return: Confirmation avec ID""" + note_data = { + "title": title, + "is_todo": 1 if is_todo else 0, + } + if body: + note_data["body"] = body + if notebook_id: + note_data["parent_id"] = notebook_id + result = self._post("notes", note_data) + if isinstance(result, dict) and "error" in result: + return result["error"] + nid = result.get("id", "?") + kind = "TODO" if is_todo else "note" + return f"[OK] {kind} '{title}' creee (ID: {nid}). Visible dans Joplin apres synchronisation." + + def update_note(self, note_id: str, title: str = None, body: str = None) -> str: + """Met a jour une note Joplin existante. + :param note_id: ID de la note a modifier + :param title: Nouveau titre (optionnel) + :param body: Nouveau contenu Markdown (optionnel) + :return: Confirmation""" + note_data = {} + if title is not None: + note_data["title"] = title + if body is not None: + note_data["body"] = body + if not note_data: + return "[WARNING] Rien a mettre a jour. Specifiez title ou body." + result = self._put(f"notes/{note_id}", note_data) + if isinstance(result, dict) and "error" in result: + return result["error"] + return f"[OK] Note {note_id} mise a jour." + + def add_todo(self, note_id: str, todo_text: str) -> str: + """Ajoute une ligne TODO a la fin d'une note existante. + :param note_id: ID de la note + :param todo_text: Texte de la tache a ajouter + :return: Confirmation""" + current = self._get(f"notes/{note_id}", params={"fields": "body,title"}) + if isinstance(current, dict) and "error" in current: + return current["error"] + existing_body = current.get("body", "") + new_line = f"- [ ] {todo_text}" + updated_body = f"{existing_body.rstrip()}\n{new_line}" if existing_body.strip() else new_line + result = self._put(f"notes/{note_id}", {"body": updated_body}) + if isinstance(result, dict) and "error" in result: + return result["error"] + return f"[OK] TODO ajoute a '{current.get('title', '?')}' : {todo_text}" + + def search_notes(self, query: str) -> str: + """Recherche des notes dans Joplin par mot-cle. + :param query: Requete de recherche (supporte la syntaxe Joplin) + :return: Liste des notes correspondantes""" + data = self._get("search", params={"query": query, "fields": "id,title,is_todo,todo_completed", "limit": "20"}) + if isinstance(data, dict) and "error" in data: + return data["error"] + items = data if isinstance(data, list) else data.get("items", []) + if not items: + return f"[INFO] Aucun resultat pour '{query}'." + lines = [] + for note in items: + nid = note.get("id", "?") + title = note.get("title", "Sans titre") + is_todo = note.get("is_todo", 0) + status = "" + if is_todo: + status = " [TODO]" + (" OK" if note.get("todo_completed") else " a faire") + lines.append(f"- {title}{status} (ID: {nid})") + has_more = data.get("has_more", False) if isinstance(data, dict) else False + if has_more: + lines.append(f"... et d'autres resultats") + return "\n".join(lines) + + def toggle_todo(self, note_id: str, completed: bool = True) -> str: + """Marque un TODO comme termine ou non termine. + :param note_id: ID du TODO + :param completed: True pour termine, False pour a faire + :return: Confirmation""" + import time + note_data = { + "is_todo": 1, + "todo_completed": int(time.time() * 1000) if completed else 0, + } + result = self._put(f"notes/{note_id}", note_data) + if isinstance(result, dict) and "error" in result: + return result["error"] + status = "terminee" if completed else "a faire" + return f"[OK] TODO {note_id} marque comme {status}." + + def delete_note(self, note_id: str) -> str: + """Supprime une note Joplin (deplace dans la corbeille). + :param note_id: ID de la note a supprimer + :return: Confirmation""" + result = self._delete(f"notes/{note_id}") + if isinstance(result, dict) and "error" in result: + return result["error"] + return f"[OK] Note {note_id} deplacee dans la corbeille." + + +def ping() -> str: + t = Tools() + return t.ping() + +def list_notebooks() -> str: + t = Tools() + return t.list_notebooks() + +def create_notebook(name: str, parent_id: str = "") -> str: + t = Tools() + return t.create_notebook(name, parent_id) + +def list_notes(notebook_id: str) -> str: + t = Tools() + return t.list_notes(notebook_id) + +def get_note(note_id: str) -> str: + t = Tools() + return t.get_note(note_id) + +def create_note(title: str, body: str = "", notebook_id: str = "", is_todo: bool = False) -> str: + t = Tools() + return t.create_note(title, body, notebook_id, is_todo) + +def update_note(note_id: str, title: str = None, body: str = None) -> str: + t = Tools() + return t.update_note(note_id, title, body) + +def add_todo(note_id: str, todo_text: str) -> str: + t = Tools() + return t.add_todo(note_id, todo_text) + +def search_notes(query: str) -> str: + t = Tools() + return t.search_notes(query) + +def toggle_todo(note_id: str, completed: bool = True) -> str: + t = Tools() + return t.toggle_todo(note_id, completed) + +def delete_note(note_id: str) -> str: + t = Tools() + return t.delete_note(note_id) \ No newline at end of file