2026-06-29 11:02:22 +02:00
"""
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 ,
)
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 " ] ,
} ,
} ,
{
2026-07-01 21:10:19 +02:00
" name " : " get_concepts " ,
" description " : " Liste les concepts d ' un projet avec leur definition (l ' ancien glossaire, fusionne dans concept). " ,
2026-06-29 11:02:22 +02:00
" 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 du systeme (backlog, decision, etc.) par son id. Garde-fou : refuse les notes hors systeme. " ,
" 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 du systeme par son id (titre, type, contenu HTML). Garde-fou : refuse les notes hors types systeme. " ,
" inputSchema " : {
" type " : " object " ,
" properties " : { " note_id " : { " type " : " string " } } ,
" required " : [ " note_id " ] ,
} ,
} ,
{
" name " : " get_children " ,
" description " : " Liste les notes filles d un dossier (id, titre, type). Ne retourne que les notes de type systeme. " ,
" 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 hors types systeme. " ,
" 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 " : " 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 " ]
2026-07-01 21:10:19 +02:00
def tool_get_concepts ( projet ) :
notes = search_by_label ( " type " , " concept " )
return [ { " id " : n [ " noteId " ] , " titre " : n . get ( " title " , " " ) ,
2026-06-29 11:02:22 +02:00
" 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 " ) ]
2026-07-01 21:10:19 +02:00
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 " ) ]
2026-06-29 11:02:22 +02:00
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 :
2026-07-01 21:10:19 +02:00
lines + = [ " ## Concepts " ]
2026-06-29 11:02:22 +02:00
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 ) :
2026-07-01 21:10:19 +02:00
from trilium_api import get_note_content
2026-06-29 11:02:22 +02:00
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
2026-07-01 12:34:29 +02:00
out . append ( { " id " : cid , " titre " : n . get ( " title " , " " ) , " type " : _type_systeme ( n ) } )
2026-06-29 11:02:22 +02:00
return { " parent " : parent_id , " count " : len ( out ) , " notes " : out }
def tool_update_note ( note_id , contenu , format = " markdown " ) :
2026-07-01 12:34:29 +02:00
from trilium_api import _est_editable
2026-06-29 11:02:22 +02:00
note = get_note ( note_id )
if not note :
return { " error " : " Note introuvable " }
2026-07-01 12:34:29 +02:00
if not _est_editable ( note ) :
return { " error " : " Refus : note structurelle protegee (sans label projet) " }
2026-06-29 11:02:22 +02:00
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_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 " <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 ,
2026-07-01 21:10:19 +02:00
" get_historique " : tool_get_historique , " get_concepts " : tool_get_concepts ,
2026-06-29 11:02:22 +02:00
" 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 ,
" 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 )
