464 lines
18 KiB
Python
464 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
==============================================================================
|
||
MCP server: TRILIUM (zápis poznámek do Trilium Notes přes ETAPI)
|
||
Verze: 1.0
|
||
Datum: 2026-06-09
|
||
Autor: vladimir.buzalka
|
||
|
||
Účel:
|
||
Umožnit libovolnému MCP klientovi (Claude Code v jiné session, Claude
|
||
Chat, Cowork, …) ČÍST a hlavně PSÁT poznámky do Trilia běžícího na
|
||
https://trilium.buzalka.cz. Slouží k rychlému předávání poznámek/souborů
|
||
do prostředí, kam se jinak nedá kopírovat (např. JNJ remote PC).
|
||
|
||
Komunikace:
|
||
Trilium ETAPI (REST), autentizace hlavičkou Authorization: <token>.
|
||
Používá jen Python stdlib (urllib) — žádné externí HTTP knihovny.
|
||
|
||
Konvence řazení:
|
||
Výchozí rodič je složka "Claude" (root > Claude), která má nastaveno
|
||
#sorted=dateCreated + #sortDirection=desc => nové poznámky jdou NAHORU
|
||
automaticky, není třeba řešit pozice.
|
||
|
||
Bezpečnost zápisu (dle zvyklostí ostatních zdejších MCP serverů):
|
||
- create_note / append_note ... bez tření (hlavní účel serveru)
|
||
- set_note_content (přepis) ... vyžaduje confirmed=True
|
||
- delete_note ... vyžaduje confirmed=True
|
||
|
||
Spuštění:
|
||
python mcp_trilium_v1.0.py (stdio MCP server)
|
||
python mcp_trilium_v1.0.py --selftest (rychlý test create/read/append/delete)
|
||
|
||
Registrace v .mcp.json jako "trilium" (viz doprovodné .md).
|
||
==============================================================================
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import html
|
||
import hashlib
|
||
import mimetypes
|
||
import urllib.parse
|
||
import urllib.request
|
||
import urllib.error
|
||
from typing import Optional
|
||
|
||
from mcp.server.fastmcp import FastMCP
|
||
|
||
# --- konfigurace ------------------------------------------------------------
|
||
BASE_URL = os.environ.get("TRILIUM_URL", "https://trilium.buzalka.cz").rstrip("/")
|
||
ETAPI = BASE_URL + "/etapi"
|
||
# Token lze přepsat proměnnou prostředí TRILIUM_ETAPI_TOKEN; jinak default níže.
|
||
TOKEN = os.environ.get(
|
||
"TRILIUM_ETAPI_TOKEN",
|
||
"WoPH9O8hn2y6_r6pQSjpOVSmuL0os2hIQsLBHDOawebOx8l+MUc8v+GE=",
|
||
)
|
||
# Výchozí složka, kam se zapisuje (root > Claude), newest-first.
|
||
CLAUDE_NOTE_ID = os.environ.get("TRILIUM_DEFAULT_PARENT", "NeoXOIw0uBK2")
|
||
|
||
HTTP_TIMEOUT = 20
|
||
|
||
|
||
def log(msg: str) -> None:
|
||
print(msg, file=sys.stderr, flush=True)
|
||
|
||
|
||
# --- nízkoúrovňové ETAPI volání ---------------------------------------------
|
||
def _req(method: str, path: str, *, json_body=None, raw_body: Optional[bytes] = None,
|
||
raw_ctype: Optional[str] = None, expect_bytes: bool = False):
|
||
"""Jedno ETAPI volání. Vrací dict/list/str (JSON nebo text), nebo bytes
|
||
(expect_bytes=True). Vyhazuje RuntimeError s tělem chyby při HTTP >= 400."""
|
||
body = None
|
||
ctype = None
|
||
if json_body is not None:
|
||
body = json.dumps(json_body, ensure_ascii=False).encode("utf-8")
|
||
ctype = "application/json; charset=utf-8"
|
||
elif raw_body is not None:
|
||
body = raw_body
|
||
ctype = raw_ctype or "application/octet-stream"
|
||
|
||
req = urllib.request.Request(ETAPI + path, data=body, method=method)
|
||
req.add_header("Authorization", TOKEN)
|
||
if ctype:
|
||
req.add_header("Content-Type", ctype)
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
|
||
data = resp.read()
|
||
if expect_bytes:
|
||
return data
|
||
text = data.decode("utf-8", errors="replace")
|
||
if text.strip().startswith(("{", "[")):
|
||
return json.loads(text)
|
||
return text
|
||
except urllib.error.HTTPError as e:
|
||
detail = e.read().decode("utf-8", errors="replace")
|
||
raise RuntimeError(f"ETAPI {method} {path} -> HTTP {e.code}: {detail}") from None
|
||
except urllib.error.URLError as e:
|
||
raise RuntimeError(f"ETAPI {method} {path} -> spojení selhalo: {e.reason}") from None
|
||
|
||
|
||
def _text_to_html(text: str, is_html: bool) -> str:
|
||
"""Plain text -> bezpečné HTML (prázdné řádky = odstavce, \\n = <br>).
|
||
Pokud is_html=True, obsah se bere tak, jak je."""
|
||
if is_html:
|
||
return text
|
||
text = text.replace("\r\n", "\n").replace("\r", "\n")
|
||
parts = []
|
||
for para in text.split("\n\n"):
|
||
if para.strip() == "":
|
||
continue
|
||
parts.append("<p>" + html.escape(para).replace("\n", "<br>") + "</p>")
|
||
return "".join(parts) or "<p></p>"
|
||
|
||
|
||
def _put_text_content(note_id: str, html_content: str) -> None:
|
||
"""Uloží HTML obsah textové poznámky (UTF-8, ověřený způsob: text/plain;charset)."""
|
||
_req("PUT", f"/notes/{note_id}/content",
|
||
raw_body=html_content.encode("utf-8"),
|
||
raw_ctype="text/plain; charset=utf-8")
|
||
|
||
|
||
def _note_summary(note: dict) -> dict:
|
||
"""Zhuštěné info o poznámce do tool response."""
|
||
return {
|
||
"noteId": note.get("noteId"),
|
||
"title": note.get("title"),
|
||
"type": note.get("type"),
|
||
"mime": note.get("mime"),
|
||
"parentNoteIds": note.get("parentNoteIds"),
|
||
"childNoteIds": note.get("childNoteIds"),
|
||
"dateCreated": note.get("dateCreated"),
|
||
"dateModified": note.get("dateModified"),
|
||
"url": f"{BASE_URL}/#root/{note.get('noteId')}",
|
||
}
|
||
|
||
|
||
# --- startup check ----------------------------------------------------------
|
||
try:
|
||
_info = _req("GET", "/app-info")
|
||
log(f"Trilium OK ({BASE_URL}) — appVersion {_info.get('appVersion')}")
|
||
except Exception as e: # noqa: BLE001
|
||
log(f"Trilium ETAPI nedostupné: {e}")
|
||
sys.exit(1)
|
||
|
||
|
||
# --- MCP --------------------------------------------------------------------
|
||
mcp = FastMCP("trilium")
|
||
|
||
|
||
@mcp.tool()
|
||
def ping() -> dict:
|
||
"""Health check Trilium ETAPI. Vrátí verzi aplikace, DB a default složku.
|
||
Zavolej jako první pro ověření, že je server dostupný a token platí."""
|
||
try:
|
||
info = _req("GET", "/app-info")
|
||
return {
|
||
"status": "ok",
|
||
"base_url": BASE_URL,
|
||
"appVersion": info.get("appVersion"),
|
||
"dbVersion": info.get("dbVersion"),
|
||
"default_parent": CLAUDE_NOTE_ID,
|
||
"default_parent_url": f"{BASE_URL}/#root/{CLAUDE_NOTE_ID}",
|
||
}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
@mcp.tool()
|
||
def create_note(
|
||
title: str,
|
||
content: str = "",
|
||
parent_note_id: Optional[str] = None,
|
||
is_html: bool = False,
|
||
) -> dict:
|
||
"""Vytvoří NOVOU textovou poznámku. Hlavní nástroj pro předávání poznámek.
|
||
|
||
title: název poznámky (zobrazí se ve stromu)
|
||
content: text poznámky; prostý text se převede na HTML (prázdný
|
||
řádek = nový odstavec). Pokud už posíláš HTML, nastav is_html=True.
|
||
parent_note_id: kam ji zařadit. Default = složka "Claude" (root > Claude),
|
||
která řadí potomky nejnovější-nahoře automaticky.
|
||
is_html: True pokud content už JE HTML.
|
||
|
||
Vrací noteId a URL nové poznámky.
|
||
"""
|
||
try:
|
||
parent = parent_note_id or CLAUDE_NOTE_ID
|
||
html_content = _text_to_html(content, is_html)
|
||
res = _req("POST", "/create-note", json_body={
|
||
"parentNoteId": parent,
|
||
"title": title,
|
||
"type": "text",
|
||
"content": html_content,
|
||
})
|
||
note = res["note"]
|
||
log(f"create_note: {note['noteId']} '{title}' pod {parent}")
|
||
return {"status": "ok", **_note_summary(note)}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
@mcp.tool()
|
||
def append_note(note_id: str, text: str, is_html: bool = False) -> dict:
|
||
"""Připíše text na KONEC existující textové poznámky (nedestruktivní).
|
||
|
||
note_id: ID poznámky, ke které připisujeme
|
||
text: co připsat; prostý text -> HTML (is_html=True pokud posíláš HTML)
|
||
"""
|
||
try:
|
||
meta = _req("GET", f"/notes/{note_id}")
|
||
if meta.get("type") != "text":
|
||
return {"status": "error",
|
||
"error": f"Poznámka {note_id} není typu 'text' (je '{meta.get('type')}')."}
|
||
current = _req("GET", f"/notes/{note_id}/content")
|
||
if not isinstance(current, str):
|
||
current = json.dumps(current, ensure_ascii=False)
|
||
addition = _text_to_html(text, is_html)
|
||
_put_text_content(note_id, (current or "") + addition)
|
||
log(f"append_note: {note_id} (+{len(text)} znaků)")
|
||
return {"status": "ok", "noteId": note_id,
|
||
"url": f"{BASE_URL}/#root/{note_id}"}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
@mcp.tool()
|
||
def set_note_content(note_id: str, content: str, is_html: bool = False,
|
||
confirmed: bool = False) -> dict:
|
||
"""PŘEPÍŠE celý obsah textové poznámky (DESTRUKTIVNÍ — smaže stávající text).
|
||
|
||
Vyžaduje confirmed=True. Při confirmed=False jen vrátí náhled: délku
|
||
stávajícího obsahu a co by se zapsalo — to ukaž uživateli a teprve po
|
||
schválení zavolej znovu s confirmed=True.
|
||
"""
|
||
try:
|
||
meta = _req("GET", f"/notes/{note_id}")
|
||
if meta.get("type") != "text":
|
||
return {"status": "error",
|
||
"error": f"Poznámka {note_id} není typu 'text' (je '{meta.get('type')}')."}
|
||
new_html = _text_to_html(content, is_html)
|
||
if not confirmed:
|
||
current = _req("GET", f"/notes/{note_id}/content")
|
||
cur_len = len(current) if isinstance(current, str) else 0
|
||
return {
|
||
"status": "preview",
|
||
"noteId": note_id,
|
||
"title": meta.get("title"),
|
||
"current_content_length": cur_len,
|
||
"new_content_preview": new_html[:500],
|
||
"note": "Přepis smaže stávající obsah. Ukaž uživateli a zavolej znovu s confirmed=True.",
|
||
}
|
||
_put_text_content(note_id, new_html)
|
||
log(f"set_note_content: {note_id} přepsáno")
|
||
return {"status": "ok", "noteId": note_id,
|
||
"url": f"{BASE_URL}/#root/{note_id}"}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
@mcp.tool()
|
||
def create_folder(title: str, parent_note_id: str = "root",
|
||
newest_first: bool = True, place_on_top: bool = False) -> dict:
|
||
"""Vytvoří podsložku (poznámku, pod kterou se vnořují další).
|
||
|
||
title: název složky
|
||
parent_note_id: rodič (default "root")
|
||
newest_first: nastaví #sorted=dateCreated + #sortDirection=desc, takže
|
||
potomci se řadí nejnovější-nahoře (jako složka Claude)
|
||
place_on_top: posune složku na úplný začátek u svého rodiče (pozice 0)
|
||
"""
|
||
try:
|
||
res = _req("POST", "/create-note", json_body={
|
||
"parentNoteId": parent_note_id,
|
||
"title": title,
|
||
"type": "text",
|
||
"content": "",
|
||
})
|
||
note = res["note"]
|
||
nid = note["noteId"]
|
||
if newest_first:
|
||
_req("POST", "/attributes", json_body={
|
||
"noteId": nid, "type": "label", "name": "sorted",
|
||
"value": "dateCreated", "isInheritable": False})
|
||
_req("POST", "/attributes", json_body={
|
||
"noteId": nid, "type": "label", "name": "sortDirection",
|
||
"value": "desc", "isInheritable": False})
|
||
if place_on_top:
|
||
branch_id = res["branch"]["branchId"]
|
||
_req("PATCH", f"/branches/{branch_id}", json_body={"notePosition": 0})
|
||
log(f"create_folder: {nid} '{title}' pod {parent_note_id}")
|
||
return {"status": "ok", "newest_first": newest_first,
|
||
"place_on_top": place_on_top, **_note_summary(note)}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
@mcp.tool()
|
||
def upload_file(file_path: str, parent_note_id: Optional[str] = None,
|
||
title: Optional[str] = None) -> dict:
|
||
"""Nahraje LOKÁLNÍ soubor (z disku stroje, kde běží tento MCP) jako
|
||
samostatnou poznámku typu 'file'. Po nahrání ověří integritu přes SHA-256.
|
||
|
||
file_path: absolutní cesta k souboru na tomto stroji
|
||
parent_note_id: kam (default složka Claude)
|
||
title: název poznámky (default = jméno souboru)
|
||
"""
|
||
try:
|
||
if not os.path.isfile(file_path):
|
||
return {"status": "error", "error": f"Soubor neexistuje: {file_path}"}
|
||
parent = parent_note_id or CLAUDE_NOTE_ID
|
||
fname = os.path.basename(file_path)
|
||
with open(file_path, "rb") as fh:
|
||
blob = fh.read()
|
||
mime = mimetypes.guess_type(fname)[0] or "application/octet-stream"
|
||
sha_local = hashlib.sha256(blob).hexdigest()
|
||
|
||
res = _req("POST", "/create-note", json_body={
|
||
"parentNoteId": parent, "title": title or fname,
|
||
"type": "file", "mime": mime, "content": ""})
|
||
nid = res["note"]["noteId"]
|
||
_req("PUT", f"/notes/{nid}/content", raw_body=blob,
|
||
raw_ctype="application/octet-stream")
|
||
_req("POST", "/attributes", json_body={
|
||
"noteId": nid, "type": "label", "name": "originalFileName",
|
||
"value": fname, "isInheritable": False})
|
||
|
||
back = _req("GET", f"/notes/{nid}/content", expect_bytes=True)
|
||
sha_remote = hashlib.sha256(back).hexdigest()
|
||
log(f"upload_file: {nid} '{fname}' {len(blob)}B match={sha_local == sha_remote}")
|
||
return {
|
||
"status": "ok",
|
||
"noteId": nid,
|
||
"title": title or fname,
|
||
"mime": mime,
|
||
"size_bytes": len(blob),
|
||
"sha256": sha_local,
|
||
"integrity_ok": sha_local == sha_remote,
|
||
"url": f"{BASE_URL}/#root/{nid}",
|
||
}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
@mcp.tool()
|
||
def read_note(note_id: str) -> dict:
|
||
"""Přečte obsah poznámky (text/HTML) + základní metadata."""
|
||
try:
|
||
meta = _req("GET", f"/notes/{note_id}")
|
||
result = {"status": "ok", **_note_summary(meta)}
|
||
if meta.get("type") in ("text", "code", "mermaid"):
|
||
content = _req("GET", f"/notes/{note_id}/content")
|
||
result["content"] = content if isinstance(content, str) else json.dumps(content, ensure_ascii=False)
|
||
else:
|
||
result["content"] = f"<binární obsah typu {meta.get('type')}/{meta.get('mime')}; použij stažení v UI>"
|
||
return result
|
||
except Exception as e: # noqa: BLE001
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
@mcp.tool()
|
||
def list_children(note_id: Optional[str] = None) -> dict:
|
||
"""Vypíše přímé potomky poznámky (default = složka Claude) v pořadí stromu."""
|
||
try:
|
||
nid = note_id or CLAUDE_NOTE_ID
|
||
meta = _req("GET", f"/notes/{nid}")
|
||
children = []
|
||
for child_branch in meta.get("childBranchIds", []):
|
||
br = _req("GET", f"/branches/{child_branch}")
|
||
ch = _req("GET", f"/notes/{br['noteId']}")
|
||
children.append({
|
||
"noteId": ch.get("noteId"),
|
||
"title": ch.get("title"),
|
||
"type": ch.get("type"),
|
||
"notePosition": br.get("notePosition"),
|
||
"dateCreated": ch.get("dateCreated"),
|
||
})
|
||
return {"status": "ok", "parent": nid, "parent_title": meta.get("title"),
|
||
"count": len(children), "children": children}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
@mcp.tool()
|
||
def search_notes(query: str, limit: int = 20, ancestor_note_id: Optional[str] = None) -> dict:
|
||
"""Fulltextové / atributové vyhledávání poznámek (Trilium search syntax).
|
||
|
||
query: např. 'protokol', '#book', 'note.title *=* report'
|
||
limit: max počet výsledků (default 20)
|
||
ancestor_note_id: omez hledání na podstrom (např. složka Claude)
|
||
"""
|
||
try:
|
||
params = {"search": query, "limit": str(max(1, min(limit, 200)))}
|
||
if ancestor_note_id:
|
||
params["ancestorNoteId"] = ancestor_note_id
|
||
qs = urllib.parse.urlencode(params)
|
||
res = _req("GET", f"/notes?{qs}")
|
||
results = res.get("results", []) if isinstance(res, dict) else []
|
||
out = [{
|
||
"noteId": n.get("noteId"),
|
||
"title": n.get("title"),
|
||
"type": n.get("type"),
|
||
"dateModified": n.get("dateModified"),
|
||
"url": f"{BASE_URL}/#root/{n.get('noteId')}",
|
||
} for n in results]
|
||
return {"status": "ok", "query": query, "count": len(out), "results": out}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
@mcp.tool()
|
||
def delete_note(note_id: str, confirmed: bool = False) -> dict:
|
||
"""SMAŽE poznámku (a její potomky) — DESTRUKTIVNÍ.
|
||
|
||
Vyžaduje confirmed=True. Při confirmed=False vrátí náhled (název + počet
|
||
potomků), který ukaž uživateli; teprve po schválení zavolej s confirmed=True.
|
||
"""
|
||
try:
|
||
meta = _req("GET", f"/notes/{note_id}")
|
||
if not confirmed:
|
||
return {
|
||
"status": "preview",
|
||
"noteId": note_id,
|
||
"title": meta.get("title"),
|
||
"type": meta.get("type"),
|
||
"child_count": len(meta.get("childNoteIds", [])),
|
||
"note": "Smazání je nevratné a zahrne i potomky. Ukaž uživateli a zavolej znovu s confirmed=True.",
|
||
}
|
||
_req("DELETE", f"/notes/{note_id}")
|
||
log(f"delete_note: {note_id} '{meta.get('title')}' smazáno")
|
||
return {"status": "ok", "deleted": note_id, "title": meta.get("title")}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"status": "error", "error": str(e)}
|
||
|
||
|
||
# --- selftest (mimo MCP) ----------------------------------------------------
|
||
def selftest() -> None:
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
print("== Trilium MCP selftest ==")
|
||
print("ping:", json.dumps(ping(), ensure_ascii=False))
|
||
c = create_note("MCP selftest – ěščřž", "První řádek.\n\nDruhý odstavec — diakritika.")
|
||
print("create:", json.dumps(c, ensure_ascii=False))
|
||
nid = c.get("noteId")
|
||
if nid:
|
||
print("append:", json.dumps(append_note(nid, "Připsaný řádek."), ensure_ascii=False))
|
||
r = read_note(nid)
|
||
print("read.content:", r.get("content"))
|
||
print("delete:", json.dumps(delete_note(nid, confirmed=True), ensure_ascii=False))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if "--selftest" in sys.argv:
|
||
selftest()
|
||
else:
|
||
log("MCP trilium server started (FastMCP)")
|
||
mcp.run()
|