Effi/docs/TOOL-JOPLIN-OPENWEBUI.md

Joplin Tool for Open WebUI (Clipper API)

Description

Permet a Effi de lire et ecrire des notes dans Joplin via l'API Clipper locale. Necessite Joplin desktop ouvert avec le Web Clipper active (Options → Web Clipper → Enable). Les notes sont synchronisees automatiquement via le serveur Joplin (joplin.folzocorp.fr).

Installation

1. Admin Panel → Tools → Create Tool (ou +)
2. Name : Joplin
3. Description : Notes Joplin. Commandes: ping(), 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), delete_note(id)
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 d'authentification (Options → Web Clipper dans Joplin)

Note : Quand le PC gamer est eteint, le Tool affichera une erreur. C'est normal — il faut Joplin ouvert.

Tool Code

"""
Joplin Tool for Open WebUI - Clipper API Version
Necessite Joplin desktop ouvert + Web Clipper active.
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 (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 dans les Valves."}
            return {"error": f"[ERROR] HTTP {e.response.status_code}: {e.response.text[:200]}"}
        except Exception as e:
            return {"error": f"[ERROR] {type(e).__name__}: {e}"}

    def _post(self, path, body):
        err = self._check()
        if err:
            return {"error": err}
        try:
            r = requests.post(self._url(path), params=self._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 dans les Valves."}
            return {"error": f"[ERROR] HTTP {e.response.status_code}: {e.response.text[:200]}"}
        except Exception as e:
            return {"error": f"[ERROR] {type(e).__name__}: {e}"}

    def _put(self, path, body):
        err = self._check()
        if err:
            return {"error": err}
        try:
            r = requests.put(self._url(path), params=self._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 dans les Valves."}
            return {"error": f"[ERROR] HTTP {e.response.status_code}: {e.response.text[:200]}"}
        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 r.json() if r.text else {"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"""
        if not self.valves.joplin_api_token:
            return "[ERROR] Jeton API Joplin non configure. Ajoutez-le dans les Valves."
        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 requests.exceptions.ConnectionError:
            return f"[ERROR] Joplin inaccessible. Verifiez que l'app est ouverte et le Web Clipper active sur {self.valves.joplin_url}"
        except Exception as e:
            return f"[ERROR] {type(e).__name__}: {e}"

    def list_notebooks(self) -> str:
        """Liste tous les notebooks Joplin avec leur ID.
        :return: Liste des notebooks en arborescence"""
        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."
        by_id = {nb["id"]: nb for nb in items}
        for nb in items:
            nb.setdefault("children", [])
        roots = []
        for nb in items:
            parent = nb.get("parent_id", "")
            if not parent:
                roots.append(nb)
            elif parent in by_id:
                by_id[parent]["children"].append(nb)
        lines = []
        def _fmt(nb, d=0):
            lines.append(f"{'  '*d}- {nb['title']} (ID: {nb['id']})")
            for c in nb.get("children", []):
                _fmt(c, d + 1)
        for r in roots:
            _fmt(r)
        return "\n".join(lines)

    def create_notebook(self, name: str, parent_id: str = "") -> str:
        """Cree un nouveau notebook.
        :param name: Nom du notebook
        :param parent_id: ID du notebook parent (optionnel)
        :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})"

    def list_notes(self, notebook_id: str) -> str:
        """Liste les notes dans un notebook.
        :param notebook_id: ID du notebook
        :return: Liste des notes"""
        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.
        :param note_id: ID de la note
        :return: Contenu Markdown de la note"""
        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.
        :param title: Titre de la note
        :param body: Contenu Markdown (optionnel)
        :param notebook_id: ID du notebook (optionnel)
        :param is_todo: Creer comme 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})"

    def update_note(self, note_id: str, title: str = None, body: str = None) -> str:
        """Met a jour une note existante.
        :param note_id: ID de la note
        :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.
        :param note_id: ID de la note
        :param todo_text: Texte de la tache
        :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 par mot-cle.
        :param query: Requete de recherche
        :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 "")
            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("... 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.
        :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 (corbeille).
        :param note_id: ID de la note
        :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)