Master
7c56f125ad
- trilium_api.py : remove_relation_safe (retrait via attributeId, idempotent), move_relation_safe (reconnexion = add nouvelle puis remove ancienne, ordre sur) - mcp_server.py : 2 tools MCP exposes (22 tools au total) Comble le manque revele par la fusion render_engine.py : le systeme savait creer une relation mais pas la retirer ni la reconnecter. Rend triviales les corrections de mesusage et les fusions de hub. Teste de bout en bout.
908 lines
37 KiB
Python
908 lines
37 KiB
Python
"""
|
|
mcp_server.py — Serveur MCP pur Starlette (compatible Python 3.9)
|
|
Implémente le protocole MCP (JSON-RPC 2.0) sans FastMCP.
|
|
|
|
Transport : Streamable HTTP sur /mcp
|
|
Auth : Bearer token (clés API depuis .env)
|
|
|
|
Usage : python mcp_server.py → écoute sur 127.0.0.1:8766
|
|
"""
|
|
import os
|
|
import json
|
|
import secrets
|
|
import hashlib
|
|
import base64
|
|
import time
|
|
from datetime import datetime
|
|
from dotenv import load_dotenv
|
|
from starlette.applications import Starlette
|
|
from starlette.routing import Route
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse, Response, HTMLResponse, RedirectResponse
|
|
|
|
from trilium_api import (
|
|
create_note, get_note_id, get_note, get_note_content, get_children,
|
|
update_note_content, search_by_label, set_label, get_label_value,
|
|
find_note_by_title, delete_note_safe, move_note_safe, add_relation_safe,
|
|
remove_relation_safe, move_relation_safe,
|
|
)
|
|
|
|
load_dotenv()
|
|
|
|
API_KEY_CLAUDE = os.getenv("API_KEY_CLAUDE", "")
|
|
API_KEY_LECHAT = os.getenv("API_KEY_LECHAT", "")
|
|
VALID_TOKENS = {API_KEY_CLAUDE: "Claude", API_KEY_LECHAT: "LeChat"}
|
|
|
|
IDS_FILE = os.path.expanduser("~/App/Context_continuity/trilium_ids.json")
|
|
PROTOCOL_VERSION = "2025-06-18"
|
|
SERVER_INFO = {"name": "Context Continuity Trilium", "version": "1.0.0"}
|
|
|
|
# --- Config OAuth 2.1 (Couche 1) ---
|
|
OAUTH_BASE_URL = os.getenv("OAUTH_BASE_URL", "https://mcp-trilium.bertha-cloud.fr")
|
|
SECURITY_WORD = os.getenv("SECURITY_WORD", "")
|
|
OAUTH_STATE_FILE = os.path.expanduser("~/App/Context_continuity/oauth_state.json")
|
|
|
|
def load_oauth_state():
|
|
with open(OAUTH_STATE_FILE) as f:
|
|
return json.load(f)
|
|
|
|
def save_oauth_state(state):
|
|
with open(OAUTH_STATE_FILE, "w") as f:
|
|
json.dump(state, f, indent=2)
|
|
|
|
def load_ids():
|
|
with open(IDS_FILE) as f:
|
|
return json.load(f)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Définition des tools (schema JSON pour Le Chat)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
TOOLS = [
|
|
{
|
|
"name": "get_backlog",
|
|
"description": "Liste les items de backlog actifs d'un projet (hors fait/abandonne), tries par priorite.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"projet": {"type": "string", "description": "Nom du projet"}},
|
|
"required": ["projet"],
|
|
},
|
|
},
|
|
{
|
|
"name": "get_decisions",
|
|
"description": "Liste les decisions actives d'un projet.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"projet": {"type": "string"}},
|
|
"required": ["projet"],
|
|
},
|
|
},
|
|
{
|
|
"name": "get_historique",
|
|
"description": "Liste les entrees d'historique encore valides (faits, tests, contraintes).",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"projet": {"type": "string"}},
|
|
"required": ["projet"],
|
|
},
|
|
},
|
|
{
|
|
"name": "get_concepts",
|
|
"description": "Liste les concepts d'un projet avec leur definition (l'ancien glossaire, fusionne dans concept).",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"projet": {"type": "string"}},
|
|
"required": ["projet"],
|
|
},
|
|
},
|
|
{
|
|
"name": "get_contexte",
|
|
"description": "Genere le briefing de reprise complet d'un projet, pret a reprendre le travail.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"projet": {"type": "string"},
|
|
"llm_cible": {"type": "string", "description": "Nom du LLM cible"},
|
|
},
|
|
"required": ["projet"],
|
|
},
|
|
},
|
|
{
|
|
"name": "add_decision",
|
|
"description": "Enregistre une nouvelle decision active dans le projet.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"projet": {"type": "string"},
|
|
"enonce": {"type": "string"},
|
|
"justification": {"type": "string"},
|
|
},
|
|
"required": ["projet", "enonce"],
|
|
},
|
|
},
|
|
{
|
|
"name": "add_history",
|
|
"description": "Enregistre une entree d'historique. type_historique : Fait etabli, Test effectue, Hypothese invalidee, Contrainte decouverte.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"projet": {"type": "string"},
|
|
"enonce": {"type": "string"},
|
|
"type_historique": {"type": "string"},
|
|
"detail": {"type": "string"},
|
|
},
|
|
"required": ["projet", "enonce", "type_historique"],
|
|
},
|
|
},
|
|
{
|
|
"name": "delete_note",
|
|
"description": "Supprime une note par son id. Garde-fou : refuse les notes structurelles protegees (sans label projet ; ontologie, skills universels, dossiers).",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"note_id": {"type": "string"}},
|
|
"required": ["note_id"],
|
|
},
|
|
},
|
|
{
|
|
"name": "add_backlog",
|
|
"description": "Ajoute un item au backlog. priorite : haute, moyenne ou basse.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"projet": {"type": "string"},
|
|
"titre": {"type": "string"},
|
|
"priorite": {"type": "string"},
|
|
},
|
|
"required": ["projet", "titre"],
|
|
},
|
|
},
|
|
{
|
|
"name": "update_backlog",
|
|
"description": "Met a jour le statut d'un item backlog. statut : a faire, en cours, bloque, fait, abandonne.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"note_id": {"type": "string"},
|
|
"statut": {"type": "string"},
|
|
},
|
|
"required": ["note_id", "statut"],
|
|
},
|
|
},
|
|
{
|
|
"name": "new_conversation",
|
|
"description": "Cree une conversation. llm : Claude Sonnet, Claude Opus, Le Chat Large, Le Chat Medium.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"projet": {"type": "string"},
|
|
"titre": {"type": "string"},
|
|
"llm": {"type": "string"},
|
|
},
|
|
"required": ["projet", "titre", "llm"],
|
|
},
|
|
},
|
|
{
|
|
"name": "close_session",
|
|
"description": "Cloture une conversation en enregistrant sa synthese.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"note_id": {"type": "string"},
|
|
"synthese": {"type": "string"},
|
|
},
|
|
"required": ["note_id", "synthese"],
|
|
},
|
|
},
|
|
{
|
|
"name": "get_note",
|
|
"description": "Lit le contenu d une note par son id (titre, type, contenu HTML). Aucune restriction : toute note est lisible.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"note_id": {"type": "string"}},
|
|
"required": ["note_id"],
|
|
},
|
|
},
|
|
{
|
|
"name": "get_children",
|
|
"description": "Liste toutes les notes filles d un dossier (id, titre, type).",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"parent_id": {"type": "string"}},
|
|
"required": ["parent_id"],
|
|
},
|
|
},
|
|
{
|
|
"name": "get_skills",
|
|
"description": "Liste les skills disponibles. Sans projet, retourne aussi les skills universels.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"projet": {"type": "string", "description": "Optionnel — filtre par projet"},
|
|
"portee": {"type": "string", "description": "Optionnel — universel | reference-technique | projet | ..."},
|
|
},
|
|
"required": [],
|
|
},
|
|
},
|
|
{
|
|
"name": "update_note",
|
|
"description": "Met a jour le contenu d une note existante (skill, decision, historique, doc...). format=markdown enveloppe en <pre> (texte redige) ; format=html ecrit le HTML tel quel (notes structurees). Garde-fou : refuse les notes structurelles protegees (sans label projet).",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"note_id": {"type": "string"},
|
|
"contenu": {"type": "string"},
|
|
"format": {"type": "string", "description": "markdown | html (defaut: markdown)"},
|
|
},
|
|
"required": ["note_id", "contenu"],
|
|
},
|
|
},
|
|
{
|
|
"name": "add_relation",
|
|
"description": "Cree une relation typee (object property) entre deux notes : source ~nom cible. nom doit etre dans l ontologie (impacte, revise, documentePar, contraintPar, concerneProjet, provient, illustre, reference...). Affichee dans la Note Map.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"source_id": {"type": "string"},
|
|
"nom": {"type": "string"},
|
|
"cible_id": {"type": "string"},
|
|
},
|
|
"required": ["source_id", "nom", "cible_id"],
|
|
},
|
|
},
|
|
{
|
|
"name": "remove_relation",
|
|
"description": "Retire une relation existante : source ~nom cible. Idempotent (si absente, ne fait rien). Utile pour corriger un mesusage ou lors d une fusion de notes.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"source_id": {"type": "string"},
|
|
"nom": {"type": "string"},
|
|
"cible_id": {"type": "string"},
|
|
},
|
|
"required": ["source_id", "nom", "cible_id"],
|
|
},
|
|
},
|
|
{
|
|
"name": "move_relation",
|
|
"description": "Reconnecte une relation d une cible vers une autre en une operation : cree source ~nom nouvelle_cible puis retire source ~nom ancienne_cible. Geste de fusion de notes (reconnexion sure : cree avant de retirer).",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"source_id": {"type": "string"},
|
|
"nom": {"type": "string"},
|
|
"ancienne_cible": {"type": "string"},
|
|
"nouvelle_cible": {"type": "string"},
|
|
},
|
|
"required": ["source_id", "nom", "ancienne_cible", "nouvelle_cible"],
|
|
},
|
|
},
|
|
{
|
|
"name": "create_entite",
|
|
"description": "Cree une entite de la couche connaissance/technique. type : concept, composantLogiciel ou service. Range automatiquement dans le bon dossier. definition (optionnelle) n est posee que pour les concepts (les fait remonter dans le briefing). Garde-fou anti-doublon sur le titre.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"type": {"type": "string", "description": "concept | composantLogiciel | service"},
|
|
"projet": {"type": "string"},
|
|
"titre": {"type": "string"},
|
|
"contenu": {"type": "string"},
|
|
"definition": {"type": "string", "description": "concept uniquement"},
|
|
},
|
|
"required": ["type", "projet", "titre"],
|
|
},
|
|
},
|
|
{
|
|
"name": "move_note",
|
|
"description": "Deplace une note d un parent vers un autre (3 args : note_id, ancien_parent_id, nouveau_parent_id). Gere les clones.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"note_id": {"type": "string"},
|
|
"ancien_parent_id": {"type": "string"},
|
|
"nouveau_parent_id": {"type": "string"},
|
|
},
|
|
"required": ["note_id", "ancien_parent_id", "nouveau_parent_id"],
|
|
},
|
|
},
|
|
{
|
|
"name": "add_skill",
|
|
"description": "Enregistre un nouveau skill (savoir-faire reutilisable). portee='universel' si non lie a un projet precis.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"titre": {"type": "string"},
|
|
"contenu": {"type": "string"},
|
|
"portee": {"type": "string", "description": "universel | reference-technique | projet | ... (defaut: projet)"},
|
|
"projet": {"type": "string", "description": "Requis sauf si portee='universel'"},
|
|
},
|
|
"required": ["titre", "contenu"],
|
|
},
|
|
},
|
|
]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Implémentation des tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def tool_get_backlog(projet):
|
|
notes = search_by_label("type", "backlogItem")
|
|
actifs = [n for n in notes
|
|
if get_label_value(n["noteId"], "projet") == projet
|
|
and get_label_value(n["noteId"], "statut") not in ("fait", "abandonne")]
|
|
prio = {"haute": 0, "moyenne": 1, "basse": 2}
|
|
actifs.sort(key=lambda n: prio.get(get_label_value(n["noteId"], "priorite") or "basse", 99))
|
|
return [{"id": n["noteId"], "titre": n.get("title", ""),
|
|
"priorite": get_label_value(n["noteId"], "priorite"),
|
|
"statut": get_label_value(n["noteId"], "statut")} for n in actifs]
|
|
|
|
def tool_get_decisions(projet):
|
|
notes = search_by_label("type", "decision")
|
|
return [{"id": n["noteId"], "enonce": n.get("title", "")} for n in notes
|
|
if get_label_value(n["noteId"], "projet") == projet
|
|
and get_label_value(n["noteId"], "statut") == "active"]
|
|
|
|
def tool_get_historique(projet):
|
|
notes = search_by_label("type", "historiqueItem")
|
|
return [{"id": n["noteId"], "enonce": n.get("title", ""),
|
|
"type": get_label_value(n["noteId"], "typeHistorique")} for n in notes
|
|
if get_label_value(n["noteId"], "projet") == projet
|
|
and get_label_value(n["noteId"], "encoreValide") == "true"]
|
|
|
|
def tool_get_concepts(projet):
|
|
notes = search_by_label("type", "concept")
|
|
return [{"id": n["noteId"], "titre": n.get("title", ""),
|
|
"definition": get_label_value(n["noteId"], "definition") or ""} for n in notes
|
|
if get_label_value(n["noteId"], "projet") == projet]
|
|
|
|
def tool_get_contexte(projet, llm_cible="LLM"):
|
|
decisions = [n for n in search_by_label("type", "decision")
|
|
if get_label_value(n["noteId"], "projet") == projet
|
|
and get_label_value(n["noteId"], "statut") == "active"]
|
|
historique = [n for n in search_by_label("type", "historiqueItem")
|
|
if get_label_value(n["noteId"], "projet") == projet
|
|
and get_label_value(n["noteId"], "encoreValide") == "true"]
|
|
backlog = [n for n in search_by_label("type", "backlogItem")
|
|
if get_label_value(n["noteId"], "projet") == projet
|
|
and get_label_value(n["noteId"], "statut") not in ("fait", "abandonne")]
|
|
glossaire = [n for n in search_by_label("type", "concept")
|
|
if get_label_value(n["noteId"], "projet") == projet
|
|
and get_label_value(n["noteId"], "definition")]
|
|
date = datetime.now().strftime("%d/%m/%Y")
|
|
lines = [f"# REPRISE DE CONTEXTE — {projet} — {date}",
|
|
f"**LLM cible : {llm_cible}**", ""]
|
|
if decisions:
|
|
lines += ["## Décisions actives (ne pas remettre en question)"]
|
|
lines += [f"- {d.get('title','?')}" for d in decisions] + [""]
|
|
if historique:
|
|
lines += ["## Historique — déjà testé / établi"]
|
|
for h in historique:
|
|
t = get_label_value(h["noteId"], "typeHistorique") or ""
|
|
lines.append(f"- [{t}] {h.get('title','?')}")
|
|
lines.append("")
|
|
if glossaire:
|
|
lines += ["## Concepts"]
|
|
for g in glossaire:
|
|
d = get_label_value(g["noteId"], "definition") or "(voir note)"
|
|
lines.append(f"- **{g.get('title','?')}** : {d}")
|
|
lines.append("")
|
|
if backlog:
|
|
lines += ["## Backlog actif"]
|
|
prio = {"haute": 0, "moyenne": 1, "basse": 2}
|
|
backlog.sort(key=lambda b: prio.get(get_label_value(b["noteId"], "priorite") or "basse", 99))
|
|
for b in backlog:
|
|
p = get_label_value(b["noteId"], "priorite") or "?"
|
|
s = get_label_value(b["noteId"], "statut") or "?"
|
|
lines.append(f"- [{s}] [{p}] {b.get('title','?')}")
|
|
lines.append("")
|
|
lines += ["---", "Confirme en 3 lignes : (a) objectif (b) prochaine action (c) incertitude"]
|
|
return "\n".join(lines)
|
|
|
|
def tool_add_decision(projet, enonce, justification=""):
|
|
ids = load_ids()
|
|
res = create_note(ids["Decisions"], enonce,
|
|
content=f"<p><b>Justification</b> : {justification or '-'}</p>")
|
|
nid = get_note_id(res)
|
|
set_label(nid, "type", "decision"); set_label(nid, "projet", projet)
|
|
set_label(nid, "statut", "active")
|
|
return {"id": nid, "enonce": enonce, "status": "cree"}
|
|
|
|
def tool_add_history(projet, enonce, type_historique, detail=""):
|
|
valides = ["Fait etabli", "Test effectue", "Hypothese invalidee", "Contrainte decouverte"]
|
|
if type_historique not in valides:
|
|
return {"error": f"type_historique invalide. Valeurs : {valides}"}
|
|
ids = load_ids()
|
|
res = create_note(ids["Historique"], enonce,
|
|
content=f"<p><b>Type</b> : {type_historique}</p><p><b>Détail</b> : {detail or '-'}</p>")
|
|
nid = get_note_id(res)
|
|
set_label(nid, "type", "historiqueItem"); set_label(nid, "projet", projet)
|
|
set_label(nid, "typeHistorique", type_historique); set_label(nid, "encoreValide", "true")
|
|
return {"id": nid, "enonce": enonce, "status": "cree"}
|
|
|
|
def tool_add_backlog(projet, titre, priorite="moyenne"):
|
|
if priorite not in ("haute", "moyenne", "basse"):
|
|
return {"error": "priorite doit etre haute, moyenne ou basse"}
|
|
ids = load_ids()
|
|
res = create_note(ids["Backlog"], titre)
|
|
nid = get_note_id(res)
|
|
set_label(nid, "type", "backlogItem"); set_label(nid, "projet", projet)
|
|
set_label(nid, "priorite", priorite); set_label(nid, "statut", "a faire")
|
|
return {"id": nid, "titre": titre, "status": "cree"}
|
|
|
|
def tool_update_backlog(note_id, statut):
|
|
valides = ["a faire", "en cours", "bloque", "fait", "abandonne"]
|
|
if statut not in valides:
|
|
return {"error": f"statut invalide. Valeurs : {valides}"}
|
|
try:
|
|
get_note(note_id)
|
|
except Exception:
|
|
return {"error": "Backlog item introuvable"}
|
|
set_label(note_id, "statut", statut)
|
|
return {"id": note_id, "statut": statut, "status": "mis a jour"}
|
|
|
|
def tool_new_conversation(projet, titre, llm):
|
|
ids = load_ids()
|
|
date = datetime.now().strftime("%Y-%m-%d %H:%M")
|
|
full = f"[{llm}] {date} — {titre}"
|
|
content = (f"<h2>{full}</h2><table>"
|
|
f"<tr><td><b>Projet</b></td><td>{projet}</td></tr>"
|
|
f"<tr><td><b>LLM</b></td><td>{llm}</td></tr>"
|
|
f"<tr><td><b>Date</b></td><td>{date}</td></tr>"
|
|
f"<tr><td><b>Synthèse de clôture</b></td><td><i>À remplir en fin de session</i></td></tr></table>")
|
|
res = create_note(ids["Conversations"], full, content=content)
|
|
nid = get_note_id(res)
|
|
set_label(nid, "type", "conversation"); set_label(nid, "projet", projet)
|
|
set_label(nid, "llm", llm); set_label(nid, "statut", "en-cours"); set_label(nid, "date", date)
|
|
return {"id": nid, "titre": full, "status": "cree"}
|
|
|
|
def tool_close_session(note_id, synthese):
|
|
try:
|
|
content = get_note_content(note_id)
|
|
except Exception:
|
|
return {"error": "Conversation introuvable"}
|
|
safe = synthese.replace("<", "<").replace(">", ">")
|
|
if "À remplir en fin de session" in content:
|
|
content = content.replace("<i>À remplir en fin de session</i>", safe)
|
|
else:
|
|
content += f"<h3>Synthèse</h3><p>{safe}</p>"
|
|
update_note_content(note_id, content)
|
|
set_label(note_id, "syntheseCloture", synthese[:200]); set_label(note_id, "statut", "clos")
|
|
return {"id": note_id, "status": "cloture"}
|
|
|
|
def tool_get_skills(projet=None, portee=None):
|
|
notes = search_by_label("type", "skill")
|
|
result = []
|
|
for n in notes:
|
|
nid = n["noteId"]
|
|
note_projet = get_label_value(nid, "projet")
|
|
note_portee = get_label_value(nid, "portee")
|
|
if projet and note_projet != projet and note_portee != "universel":
|
|
continue
|
|
if portee and note_portee != portee:
|
|
continue
|
|
result.append({
|
|
"id": nid, "titre": n.get("title", ""),
|
|
"portee": note_portee or "", "projet": note_projet or "",
|
|
"contenu": get_note_content(nid),
|
|
})
|
|
return result
|
|
|
|
def _type_systeme(note):
|
|
for a in note.get("attributes", []):
|
|
if a.get("type") == "label" and a.get("name") == "type":
|
|
return a.get("value")
|
|
return None
|
|
|
|
def tool_get_note(note_id):
|
|
from trilium_api import get_note_content
|
|
note = get_note(note_id)
|
|
if not note:
|
|
return {"error": "Note introuvable"}
|
|
tv = _type_systeme(note)
|
|
return {"id": note_id, "titre": note.get("title", ""), "type": tv,
|
|
"contenu": get_note_content(note_id)}
|
|
|
|
def tool_get_children(parent_id):
|
|
out = []
|
|
for child in get_children(parent_id):
|
|
cid = child.get("noteId")
|
|
n = get_note(cid)
|
|
if not n:
|
|
continue
|
|
out.append({"id": cid, "titre": n.get("title", ""), "type": _type_systeme(n)})
|
|
return {"parent": parent_id, "count": len(out), "notes": out}
|
|
|
|
def tool_update_note(note_id, contenu, format="markdown"):
|
|
from trilium_api import _est_editable
|
|
note = get_note(note_id)
|
|
if not note:
|
|
return {"error": "Note introuvable"}
|
|
if not _est_editable(note):
|
|
return {"error": "Refus : note structurelle protegee (sans label projet)"}
|
|
if format == "html":
|
|
body = contenu
|
|
else:
|
|
safe = contenu.replace("<", "<").replace(">", ">")
|
|
body = "<pre>" + safe + "</pre>"
|
|
update_note_content(note_id, body)
|
|
return {"id": note_id, "titre": note.get("title", ""), "format": format, "status": "mis a jour"}
|
|
|
|
def tool_create_entite(type, projet, titre, contenu=None, definition=None):
|
|
"""Cree une entite de la couche connaissance/technique (concept, composantLogiciel,
|
|
service). Le type est valide contre une liste blanche ; le dossier cible est resolu
|
|
via trilium_ids.json. Garde-fou anti-doublon sur le titre dans le dossier cible.
|
|
La permission projet est comportementale (le tool accepte tout projet)."""
|
|
import json, os
|
|
from trilium_api import create_note, get_note_id, set_label, find_note_by_title, TYPES_CANONIQUES
|
|
# Routage dossier : uniquement la couche connaissance/technique.
|
|
# Les types de pilotage (backlogItem, decision...) ont leurs tools dedies.
|
|
MAPPING = {
|
|
"concept": "Concepts",
|
|
"composantLogiciel": "EnvironnementTechnique",
|
|
"service": "EnvironnementTechnique",
|
|
"infrastructure": "EnvironnementTechnique",
|
|
"outil": "EnvironnementTechnique",
|
|
"contrainte": "EnvironnementTechnique",
|
|
"principe": "EnvironnementTechnique",
|
|
"convention": "EnvironnementTechnique",
|
|
"methode": "Methodes",
|
|
"version": "EnvironnementTechnique",
|
|
}
|
|
# 1. Validite du type (source unique : ontologie via TYPES_CANONIQUES)
|
|
if type not in TYPES_CANONIQUES:
|
|
return {"error": "Type inconnu : %s. Types canoniques : %s" % (type, ", ".join(TYPES_CANONIQUES))}
|
|
# 2. Ce tool ne cree que la couche connaissance/technique
|
|
if type not in MAPPING:
|
|
return {"error": "Type '%s' valide mais non creable ici (type de pilotage : utiliser son tool dedie add_*)." % type}
|
|
with open(os.path.expanduser("~/App/Context_continuity/trilium_ids.json")) as f:
|
|
ids = json.load(f)
|
|
dossier_cle = MAPPING[type]
|
|
if dossier_cle not in ids:
|
|
return {"error": "Dossier cible absent de trilium_ids.json : %s" % dossier_cle}
|
|
parent_id = ids[dossier_cle]
|
|
# Anti-doublon : refuse si une note du meme titre existe deja dans le dossier cible
|
|
existant = find_note_by_title(titre, parent_id)
|
|
if existant:
|
|
return {"error": "Doublon : une note '%s' existe deja dans le dossier (%s)" % (titre, existant)}
|
|
body = contenu if contenu else " "
|
|
res = create_note(parent_id, titre, content=body)
|
|
nid = get_note_id(res)
|
|
set_label(nid, "type", type)
|
|
set_label(nid, "projet", projet)
|
|
if type == "concept" and definition:
|
|
set_label(nid, "definition", definition)
|
|
return {"id": nid, "titre": titre, "type": type, "projet": projet, "status": "cree"}
|
|
|
|
def tool_add_relation(source_id, nom, cible_id):
|
|
ok, message = add_relation_safe(source_id, nom, cible_id)
|
|
return {"success": ok, "message": message}
|
|
|
|
def tool_remove_relation(source_id, nom, cible_id):
|
|
ok, message = remove_relation_safe(source_id, nom, cible_id)
|
|
return {"success": ok, "message": message}
|
|
|
|
def tool_move_relation(source_id, nom, ancienne_cible, nouvelle_cible):
|
|
ok, message = move_relation_safe(source_id, nom, ancienne_cible, nouvelle_cible)
|
|
return {"success": ok, "message": message}
|
|
|
|
def tool_move_note(note_id, ancien_parent_id, nouveau_parent_id):
|
|
ok, message = move_note_safe(note_id, ancien_parent_id, nouveau_parent_id)
|
|
return {"success": ok, "message": message}
|
|
|
|
def tool_add_skill(titre, contenu, portee="projet", projet=None):
|
|
if portee != "universel" and not projet:
|
|
return {"error": "projet requis sauf si portee='universel'"}
|
|
ids = load_ids()
|
|
existant = find_note_by_title(titre, ids["Skills"])
|
|
if existant:
|
|
return {"error": f"Un skill '{titre}' existe déjà", "id": existant}
|
|
safe = contenu.replace("<", "<").replace(">", ">")
|
|
res = create_note(ids["Skills"], titre, content=f"<pre>{safe}</pre>")
|
|
nid = get_note_id(res)
|
|
set_label(nid, "type", "skill")
|
|
set_label(nid, "portee", portee)
|
|
if projet:
|
|
set_label(nid, "projet", projet)
|
|
return {"id": nid, "titre": titre, "status": "cree"}
|
|
|
|
def tool_delete_note(note_id):
|
|
ok, message = delete_note_safe(note_id)
|
|
return {"success": ok, "message": message}
|
|
|
|
DISPATCH = {
|
|
"get_backlog": tool_get_backlog, "get_decisions": tool_get_decisions,
|
|
"get_historique": tool_get_historique, "get_concepts": tool_get_concepts,
|
|
"get_contexte": tool_get_contexte, "add_decision": tool_add_decision,
|
|
"add_history": tool_add_history, "add_backlog": tool_add_backlog,
|
|
"delete_note": tool_delete_note,
|
|
"update_backlog": tool_update_backlog, "new_conversation": tool_new_conversation,
|
|
"close_session": tool_close_session, "get_skills": tool_get_skills,
|
|
"add_skill": tool_add_skill,
|
|
"move_note": tool_move_note,
|
|
"add_relation": tool_add_relation, "remove_relation": tool_remove_relation, "move_relation": tool_move_relation, "create_entite": tool_create_entite,
|
|
"update_note": tool_update_note,
|
|
"get_note": tool_get_note, "get_children": tool_get_children,
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Handler JSON-RPC MCP
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def check_auth(request: Request):
|
|
auth = request.headers.get("authorization", "")
|
|
token = auth.replace("Bearer ", "").strip()
|
|
# 1) Tokens statiques (Le Chat, cle Claude historique)
|
|
if token in VALID_TOKENS:
|
|
return VALID_TOKENS[token]
|
|
# 2) Tokens OAuth emis (Claude Desktop)
|
|
if token.startswith("at_"):
|
|
oauth = load_oauth_state()
|
|
meta = oauth["tokens"].get(token)
|
|
if meta:
|
|
return "Claude-OAuth"
|
|
return None
|
|
|
|
async def mcp_endpoint(request: Request):
|
|
# Auth
|
|
llm = check_auth(request)
|
|
if not llm:
|
|
return JSONResponse(
|
|
{"error": "invalid_token", "error_description": "Authentication required"},
|
|
status_code=401,
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
body = await request.json()
|
|
method = body.get("method")
|
|
req_id = body.get("id")
|
|
params = body.get("params", {})
|
|
|
|
# initialize
|
|
if method == "initialize":
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"result": {
|
|
"protocolVersion": PROTOCOL_VERSION,
|
|
"capabilities": {"tools": {}},
|
|
"serverInfo": SERVER_INFO,
|
|
},
|
|
})
|
|
|
|
# notifications/initialized (pas de réponse attendue)
|
|
if method == "notifications/initialized":
|
|
return Response(status_code=202)
|
|
|
|
# tools/list
|
|
if method == "tools/list":
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"result": {"tools": TOOLS},
|
|
})
|
|
|
|
# tools/call
|
|
if method == "tools/call":
|
|
tool_name = params.get("name")
|
|
args = params.get("arguments", {})
|
|
fn = DISPATCH.get(tool_name)
|
|
if not fn:
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"error": {"code": -32601, "message": f"Tool inconnu : {tool_name}"},
|
|
})
|
|
try:
|
|
result = fn(**args)
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"result": {
|
|
"content": [{"type": "text",
|
|
"text": json.dumps(result, ensure_ascii=False, indent=2)}],
|
|
},
|
|
})
|
|
except Exception as e:
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"error": {"code": -32603, "message": str(e)},
|
|
})
|
|
|
|
# méthode inconnue
|
|
return JSONResponse({
|
|
"jsonrpc": "2.0", "id": req_id,
|
|
"error": {"code": -32601, "message": f"Méthode inconnue : {method}"},
|
|
})
|
|
|
|
async def health(request: Request):
|
|
return JSONResponse({"status": "ok", "server": "mcp", "version": "1.0.0"})
|
|
|
|
async def well_known_protected_resource(request: Request):
|
|
return JSONResponse({
|
|
"resource": OAUTH_BASE_URL + "/mcp",
|
|
"authorization_servers": [OAUTH_BASE_URL],
|
|
})
|
|
|
|
async def well_known_authorization_server(request: Request):
|
|
return JSONResponse({
|
|
"issuer": OAUTH_BASE_URL,
|
|
"authorization_endpoint": OAUTH_BASE_URL + "/authorize",
|
|
"token_endpoint": OAUTH_BASE_URL + "/token",
|
|
"registration_endpoint": OAUTH_BASE_URL + "/register",
|
|
"response_types_supported": ["code"],
|
|
"grant_types_supported": ["authorization_code", "refresh_token"],
|
|
"code_challenge_methods_supported": ["S256"],
|
|
"token_endpoint_auth_methods_supported": ["none"],
|
|
})
|
|
|
|
async def register(request: Request):
|
|
"""Dynamic Client Registration (RFC 7591)."""
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
body = {}
|
|
client_id = "client_" + secrets.token_urlsafe(16)
|
|
state = load_oauth_state()
|
|
state["clients"][client_id] = {
|
|
"client_id": client_id,
|
|
"redirect_uris": body.get("redirect_uris", []),
|
|
"client_name": body.get("client_name", "unknown"),
|
|
"created_at": int(time.time()),
|
|
}
|
|
save_oauth_state(state)
|
|
# Reponse RFC 7591 : client public (pas de secret), auth method none
|
|
return JSONResponse({
|
|
"client_id": client_id,
|
|
"redirect_uris": body.get("redirect_uris", []),
|
|
"client_name": body.get("client_name", "unknown"),
|
|
"token_endpoint_auth_method": "none",
|
|
"grant_types": ["authorization_code", "refresh_token"],
|
|
"response_types": ["code"],
|
|
}, status_code=201)
|
|
|
|
AUTHORIZE_PAGE = """<!DOCTYPE html>
|
|
<html lang="fr"><head><meta charset="utf-8"><title>Autorisation Trilium MCP</title>
|
|
<style>body{{font-family:sans-serif;max-width:400px;margin:80px auto;padding:20px}}
|
|
input{{width:100%;padding:10px;margin:10px 0;box-sizing:border-box}}
|
|
button{{padding:10px 20px;background:#000;color:#fff;border:none;cursor:pointer}}
|
|
.err{{color:#c00}}</style></head>
|
|
<body><h2>Autorisation Trilium MCP</h2>
|
|
<p>Saisissez le mot de securite pour autoriser l acces.</p>
|
|
{error}
|
|
<form method="POST" action="/authorize">
|
|
<input type="hidden" name="client_id" value="{client_id}">
|
|
<input type="hidden" name="redirect_uri" value="{redirect_uri}">
|
|
<input type="hidden" name="state" value="{state}">
|
|
<input type="hidden" name="code_challenge" value="{code_challenge}">
|
|
<input type="hidden" name="code_challenge_method" value="{code_challenge_method}">
|
|
<input type="password" name="security_word" placeholder="Mot de securite" autofocus>
|
|
<button type="submit">Autoriser</button>
|
|
</form></body></html>"""
|
|
|
|
async def authorize_get(request: Request):
|
|
q = request.query_params
|
|
page = AUTHORIZE_PAGE.format(
|
|
error="",
|
|
client_id=q.get("client_id", ""),
|
|
redirect_uri=q.get("redirect_uri", ""),
|
|
state=q.get("state", ""),
|
|
code_challenge=q.get("code_challenge", ""),
|
|
code_challenge_method=q.get("code_challenge_method", "S256"),
|
|
)
|
|
return HTMLResponse(page)
|
|
|
|
async def authorize_post(request: Request):
|
|
form = await request.form()
|
|
word = form.get("security_word", "")
|
|
client_id = form.get("client_id", "")
|
|
redirect_uri = form.get("redirect_uri", "")
|
|
state_param = form.get("state", "")
|
|
code_challenge = form.get("code_challenge", "")
|
|
code_challenge_method = form.get("code_challenge_method", "S256")
|
|
|
|
if word != SECURITY_WORD:
|
|
page = AUTHORIZE_PAGE.format(
|
|
error='<p class="err">Mot de securite incorrect.</p>',
|
|
client_id=client_id, redirect_uri=redirect_uri,
|
|
state=state_param, code_challenge=code_challenge,
|
|
code_challenge_method=code_challenge_method,
|
|
)
|
|
return HTMLResponse(page, status_code=401)
|
|
|
|
oauth = load_oauth_state()
|
|
if client_id not in oauth["clients"]:
|
|
return JSONResponse({"error": "client inconnu"}, status_code=400)
|
|
|
|
code = "code_" + secrets.token_urlsafe(24)
|
|
oauth["auth_codes"][code] = {
|
|
"client_id": client_id,
|
|
"redirect_uri": redirect_uri,
|
|
"code_challenge": code_challenge,
|
|
"code_challenge_method": code_challenge_method,
|
|
"created_at": int(time.time()),
|
|
"used": False,
|
|
}
|
|
save_oauth_state(oauth)
|
|
|
|
sep = "&" if "?" in redirect_uri else "?"
|
|
location = redirect_uri + sep + "code=" + code
|
|
if state_param:
|
|
location += "&state=" + state_param
|
|
return RedirectResponse(location, status_code=302)
|
|
|
|
def verify_pkce(code_verifier, code_challenge):
|
|
"""Verifie PKCE S256 : base64url(sha256(verifier)) == challenge."""
|
|
digest = hashlib.sha256(code_verifier.encode("ascii")).digest()
|
|
computed = base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")
|
|
return computed == code_challenge
|
|
|
|
async def token(request: Request):
|
|
form = await request.form()
|
|
grant_type = form.get("grant_type", "")
|
|
oauth = load_oauth_state()
|
|
|
|
if grant_type == "authorization_code":
|
|
code = form.get("code", "")
|
|
code_verifier = form.get("code_verifier", "")
|
|
entry = oauth["auth_codes"].get(code)
|
|
if not entry:
|
|
return JSONResponse({"error": "invalid_grant"}, status_code=400)
|
|
if entry.get("used"):
|
|
return JSONResponse({"error": "invalid_grant", "error_description": "code deja utilise"}, status_code=400)
|
|
# Validation PKCE
|
|
challenge = entry.get("code_challenge", "")
|
|
if challenge:
|
|
if not code_verifier or not verify_pkce(code_verifier, challenge):
|
|
return JSONResponse({"error": "invalid_grant", "error_description": "PKCE echec"}, status_code=400)
|
|
# Emettre les tokens
|
|
access_token = "at_" + secrets.token_urlsafe(32)
|
|
refresh_token = "rt_" + secrets.token_urlsafe(32)
|
|
oauth["tokens"][access_token] = {
|
|
"client_id": entry["client_id"],
|
|
"refresh_token": refresh_token,
|
|
"created_at": int(time.time()),
|
|
"type": "access",
|
|
}
|
|
oauth["auth_codes"][code]["used"] = True
|
|
save_oauth_state(oauth)
|
|
return JSONResponse({
|
|
"access_token": access_token,
|
|
"token_type": "Bearer",
|
|
"expires_in": 31536000,
|
|
"refresh_token": refresh_token,
|
|
})
|
|
|
|
elif grant_type == "refresh_token":
|
|
rt = form.get("refresh_token", "")
|
|
# Retrouver le token associe a ce refresh
|
|
for at, meta in list(oauth["tokens"].items()):
|
|
if meta.get("refresh_token") == rt:
|
|
new_at = "at_" + secrets.token_urlsafe(32)
|
|
oauth["tokens"][new_at] = {
|
|
"client_id": meta["client_id"],
|
|
"refresh_token": rt,
|
|
"created_at": int(time.time()),
|
|
"type": "access",
|
|
}
|
|
save_oauth_state(oauth)
|
|
return JSONResponse({
|
|
"access_token": new_at,
|
|
"token_type": "Bearer",
|
|
"expires_in": 31536000,
|
|
"refresh_token": rt,
|
|
})
|
|
return JSONResponse({"error": "invalid_grant"}, status_code=400)
|
|
|
|
return JSONResponse({"error": "unsupported_grant_type"}, status_code=400)
|
|
|
|
app = Starlette(routes=[
|
|
Route("/mcp", mcp_endpoint, methods=["POST"]),
|
|
Route("/", mcp_endpoint, methods=["POST"]),
|
|
Route("/health", health, methods=["GET"]),
|
|
Route("/.well-known/oauth-protected-resource", well_known_protected_resource, methods=["GET"]),
|
|
Route("/.well-known/oauth-authorization-server", well_known_authorization_server, methods=["GET"]),
|
|
Route("/register", register, methods=["POST"]),
|
|
Route("/authorize", authorize_get, methods=["GET"]),
|
|
Route("/authorize", authorize_post, methods=["POST"]),
|
|
Route("/token", token, methods=["POST"]),
|
|
])
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="127.0.0.1", port=8766)
|