(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": "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"Justification : {justification or '-'}
")
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"Type : {type_historique}
Détail : {detail or '-'}
")
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"{full}
| Projet | {projet} |
| LLM | {llm} |
| Date | {date} |
| Synthèse de clôture | À remplir en fin de session |
{safe}
" 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 = "" + safe + "" 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_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"
{safe}")
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, "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 = """
Saisissez le mot de securite pour autoriser l acces.
{error} """ 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='Mot de securite incorrect.
', 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)