Pridani novych skriptu, reportu a zpracovanych dat
- EmailsImport: jnj_mailbox_sync_v1.0 (sync JNJ schranky) - Covance: create_lab_results_report_v1.0 + zpracovane CSV (samples/kits/equeries/test-results), browser profily - Feasibility UCO2001: store_cda_*, store_sipiq_links, classify_krok, mark_sipiq_sent, report v1.1 (stary report do TRASH) - IWRS/Drugs: pregenerovane onsite inventory / shipment reporty - TrilliumMCP server + trilium upload/diacritics skripty - .mcp.json Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,463 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user