Joplin via Data API + Mail attachments
Notes server rewritten to use Joplin CLI Data API (joplin-cli sync clients on host, ports 41184/41185). Clean fast search, proper resource handling. New tools: list_note_resources, read_resource (attachments as inline image/document/text). Mail server: read_mail now lists attachments; new read_attachment tool returns images inline, PDFs/docs as EmbeddedResource, text directly. Tests: 54 total. Notes now real (notebooks, list, create+read). Mail attachment listing + fetch tested. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Stefan Lohmaier
+147
-49
@@ -1,106 +1,204 @@
|
||||
"""MCP Notes Server — reads and writes notes via Joplin API."""
|
||||
"""MCP Notes Server — reads/writes Joplin notes via Joplin CLI Data API."""
|
||||
|
||||
import os, sys, contextlib
|
||||
import os, sys, contextlib, base64
|
||||
from typing import Annotated
|
||||
|
||||
import httpx
|
||||
from pydantic import Field
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
from mcp.types import TextContent, ImageContent, EmbeddedResource, BlobResourceContents
|
||||
from starlette.applications import Starlette
|
||||
from starlette.routing import Mount
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
||||
from common import get_current_user, OAUTH_ROUTES, BearerAuthMiddleware
|
||||
from common import get_current_user, OAUTH_ROUTES, BearerAuthMiddleware, load_config
|
||||
|
||||
from common import load_config as _lc
|
||||
_cfg = _lc()
|
||||
JOPLIN = _cfg['joplin_url']
|
||||
JOPLIN_TOKENS = _cfg['joplin_tokens']
|
||||
_cfg = load_config()
|
||||
DATA_API = _cfg["joplin_data_api"] # {user: {url, token}}
|
||||
|
||||
mcp = FastMCP("Notes", stateless_http=True,
|
||||
transport_security={"enable_dns_rebinding_protection": False})
|
||||
|
||||
def _api(user, endpoint, method="GET", json_data=None):
|
||||
token = JOPLIN_TOKENS.get(user)
|
||||
if not token: return None, "Kein Joplin API-Token konfiguriert. Bitte in Joplin-App generieren."
|
||||
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json", "Origin": "https://notes.home.slohmaier.de"}
|
||||
IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
|
||||
|
||||
|
||||
def _api(user):
|
||||
return DATA_API.get(user)
|
||||
|
||||
|
||||
def _get(user, path, params=None):
|
||||
api = _api(user)
|
||||
if not api:
|
||||
return None, "Kein Joplin-Zugang fuer diesen User"
|
||||
p = dict(params or {})
|
||||
p["token"] = api["token"]
|
||||
try:
|
||||
r = getattr(httpx, method.lower())(f"{JOPLIN}{endpoint}", headers=headers, json=json_data, timeout=15)
|
||||
if r.status_code >= 400: return None, f"Joplin-Fehler: HTTP {r.status_code}"
|
||||
return (r.json() if r.text else {}), None
|
||||
except Exception as e: return None, str(e)
|
||||
r = httpx.get(f"{api['url']}{path}", params=p, timeout=30)
|
||||
if r.status_code >= 400:
|
||||
return None, f"Joplin HTTP {r.status_code}"
|
||||
return r, None
|
||||
except Exception as e:
|
||||
return None, str(e)
|
||||
|
||||
|
||||
def _post(user, path, json_data):
|
||||
api = _api(user)
|
||||
if not api:
|
||||
return None, "Kein Joplin-Zugang"
|
||||
try:
|
||||
r = httpx.post(f"{api['url']}{path}?token={api['token']}", json=json_data, timeout=30)
|
||||
if r.status_code >= 400:
|
||||
return None, f"Joplin HTTP {r.status_code}"
|
||||
return r.json(), None
|
||||
except Exception as e:
|
||||
return None, str(e)
|
||||
|
||||
|
||||
def _all_items(user, path, params=None):
|
||||
"""Fetch all pages from a Data API list endpoint."""
|
||||
items = []
|
||||
page = 1
|
||||
while True:
|
||||
p = dict(params or {})
|
||||
p["page"] = page
|
||||
r, err = _get(user, path, p)
|
||||
if err:
|
||||
return [], err
|
||||
d = r.json()
|
||||
items.extend(d.get("items", []))
|
||||
if not d.get("has_more"):
|
||||
break
|
||||
page += 1
|
||||
if page > 50:
|
||||
break
|
||||
return items, None
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_notebooks() -> str:
|
||||
"""List all Joplin notebooks (folders). Call this first to see available notebooks."""
|
||||
"""List all Joplin notebooks (folders). Call this first to see notebook names and IDs."""
|
||||
user = get_current_user()
|
||||
if not user: return "Error: not authenticated"
|
||||
data, err = _api(user, "/api/folders")
|
||||
items, err = _all_items(user, "/folders")
|
||||
if err: return f"Fehler: {err}"
|
||||
return "\n".join(f"{nb['title']} (ID: {nb['id']})" for nb in data.get("items", [])) or "Keine Notizbuecher"
|
||||
return "\n".join(f"{i['title']} (id: {i['id']})" for i in items) if items else "Keine Notizbuecher"
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_notes(
|
||||
notebook: Annotated[str, Field(description="Notebook name or ID from list_notebooks. Leave empty for all notes.")] = "",
|
||||
limit: Annotated[int, Field(description="Maximum number of notes to return")] = 50,
|
||||
limit: Annotated[int, Field(description="Maximum number of notes")] = 50,
|
||||
) -> str:
|
||||
"""List notes, optionally filtered by notebook. Shows title and ID. Use read_note with the ID to get content."""
|
||||
"""List notes, optionally filtered by notebook. Shows title and ID."""
|
||||
user = get_current_user()
|
||||
if not user: return "Error: not authenticated"
|
||||
nb_id = ""
|
||||
if notebook:
|
||||
data, err = _api(user, "/api/folders")
|
||||
folders, err = _all_items(user, "/folders")
|
||||
if err: return f"Fehler: {err}"
|
||||
nb_id = next((nb["id"] for nb in data.get("items", []) if notebook.lower() in nb.get("title","").lower() or notebook == nb.get("id")), None)
|
||||
if not nb_id: return f"Notizbuch '{notebook}' nicht gefunden"
|
||||
data, err = _api(user, f"/api/folders/{nb_id}/notes?limit={limit}")
|
||||
for f in folders:
|
||||
if notebook.lower() in f["title"].lower() or notebook == f["id"]:
|
||||
nb_id = f["id"]
|
||||
break
|
||||
if not nb_id:
|
||||
return f"Notizbuch nicht gefunden: {notebook}"
|
||||
path = f"/folders/{nb_id}/notes"
|
||||
else:
|
||||
data, err = _api(user, f"/api/notes?limit={limit}&order_by=updated_time&order_dir=DESC")
|
||||
path = "/notes"
|
||||
items, err = _all_items(user, path, {"fields": "id,title", "order_by": "updated_time", "order_dir": "DESC"})
|
||||
if err: return f"Fehler: {err}"
|
||||
return "\n".join(f"{n['title']} (ID: {n['id']})" for n in data.get("items", [])) or "Keine Notizen"
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def read_note(
|
||||
note_id: Annotated[str, Field(description="Note ID from list_notes or search_notes results")],
|
||||
) -> str:
|
||||
"""Read the full content of a note (Markdown format)."""
|
||||
user = get_current_user()
|
||||
if not user: return "Error: not authenticated"
|
||||
data, err = _api(user, f"/api/notes/{note_id}?fields=id,title,body")
|
||||
if err: return f"Fehler: {err}"
|
||||
return f"# {data.get('title','')}\n\n{data.get('body','')}"
|
||||
return "\n".join(f"{i['title']} (id: {i['id']})" for i in items[:limit]) if items else "Keine Notizen"
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def search_notes(
|
||||
query: Annotated[str, Field(description="Search term — matches note titles and content. Example: 'Einkaufsliste', 'Passwort', 'Rezept'")],
|
||||
query: Annotated[str, Field(description="Search term — matches note titles and content. Example: 'Einkaufsliste', 'Rezept', 'Passwort'")],
|
||||
) -> str:
|
||||
"""Search notes by keyword. Returns matching note titles and IDs."""
|
||||
"""Search notes by keyword (full text). Returns matching note titles and IDs."""
|
||||
user = get_current_user()
|
||||
if not user: return "Error: not authenticated"
|
||||
data, err = _api(user, f"/api/search?query={query}&limit=20")
|
||||
items, err = _all_items(user, "/search", {"query": query, "type": "note", "fields": "id,title"})
|
||||
if err: return f"Fehler: {err}"
|
||||
return "\n".join(f"{n['title']} (ID: {n['id']})" for n in data.get("items", [])) or "Keine Treffer"
|
||||
return "\n".join(f"{i['title']} (id: {i['id']})" for i in items[:30]) if items else "Keine Treffer"
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def read_note(
|
||||
note_id: Annotated[str, Field(description="Note ID from list_notes or search_notes")],
|
||||
) -> str:
|
||||
"""Read a note's full content (Markdown). Resource links appear as :/resourceId — use list_note_resources + read_resource for attachments."""
|
||||
user = get_current_user()
|
||||
if not user: return "Error: not authenticated"
|
||||
r, err = _get(user, f"/notes/{note_id}", {"fields": "id,title,body"})
|
||||
if err: return f"Fehler: {err}"
|
||||
d = r.json()
|
||||
return f"# {d.get('title','?')}\n\n{d.get('body','')}"
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_note_resources(
|
||||
note_id: Annotated[str, Field(description="Note ID from list_notes or search_notes")],
|
||||
) -> str:
|
||||
"""List attachments (images, PDFs, files) embedded in a note. Use read_resource with the resource ID to fetch one."""
|
||||
user = get_current_user()
|
||||
if not user: return "Error: not authenticated"
|
||||
items, err = _all_items(user, f"/notes/{note_id}/resources", {"fields": "id,title,mime,size"})
|
||||
if err: return f"Fehler: {err}"
|
||||
if not items:
|
||||
return "Keine Anhaenge"
|
||||
lines = []
|
||||
for i in items:
|
||||
size = i.get("size", 0)
|
||||
lines.append(f"{i.get('title','?')} ({i.get('mime','?')}, {size:,} bytes) id: {i['id']}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def read_resource(
|
||||
resource_id: Annotated[str, Field(description="Resource/attachment ID from list_note_resources")],
|
||||
) -> list[TextContent | ImageContent | EmbeddedResource]:
|
||||
"""Read an attachment. Images shown inline, documents (PDF/docx) as binary, text directly."""
|
||||
user = get_current_user()
|
||||
if not user: return [TextContent(type="text", text="Error: not authenticated")]
|
||||
# Get metadata
|
||||
rmeta, err = _get(user, f"/resources/{resource_id}", {"fields": "id,title,mime,size"})
|
||||
if err: return [TextContent(type="text", text=f"Fehler: {err}")]
|
||||
meta = rmeta.json()
|
||||
mime = meta.get("mime", "application/octet-stream")
|
||||
title = meta.get("title", "attachment")
|
||||
# Get blob
|
||||
rfile, err = _get(user, f"/resources/{resource_id}/file")
|
||||
if err: return [TextContent(type="text", text=f"Fehler: {err}")]
|
||||
content = rfile.content
|
||||
if mime in IMAGE_TYPES and len(content) < 10_000_000:
|
||||
return [ImageContent(type="image", data=base64.b64encode(content).decode(), mimeType=mime)]
|
||||
if mime.startswith("text/"):
|
||||
return [TextContent(type="text", text=content.decode("utf-8", errors="replace")[:100000])]
|
||||
return [EmbeddedResource(type="resource", resource=BlobResourceContents(
|
||||
uri=f"joplin://resource/{resource_id}/{title}",
|
||||
blob=base64.b64encode(content).decode(), mimeType=mime))]
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def create_note(
|
||||
notebook: Annotated[str, Field(description="Notebook name or ID where the note should be created")],
|
||||
notebook: Annotated[str, Field(description="Notebook name or ID where the note is created")],
|
||||
title: Annotated[str, Field(description="Note title")],
|
||||
body: Annotated[str, Field(description="Note content in Markdown format")],
|
||||
body: Annotated[str, Field(description="Note content in Markdown")],
|
||||
) -> str:
|
||||
"""Create a new note in the specified notebook."""
|
||||
user = get_current_user()
|
||||
if not user: return "Error: not authenticated"
|
||||
folders, err = _api(user, "/api/folders")
|
||||
folders, err = _all_items(user, "/folders")
|
||||
if err: return f"Fehler: {err}"
|
||||
nb_id = next((nb["id"] for nb in folders.get("items", []) if notebook.lower() in nb.get("title","").lower() or notebook == nb.get("id")), None)
|
||||
if not nb_id: return f"Notizbuch '{notebook}' nicht gefunden"
|
||||
data, err = _api(user, "/api/notes", "POST", {"title": title, "body": body, "parent_id": nb_id})
|
||||
nb_id = None
|
||||
for f in folders:
|
||||
if notebook.lower() in f["title"].lower() or notebook == f["id"]:
|
||||
nb_id = f["id"]
|
||||
break
|
||||
if not nb_id: return f"Notizbuch nicht gefunden: {notebook}"
|
||||
d, err = _post(user, "/notes", {"title": title, "body": body, "parent_id": nb_id})
|
||||
if err: return f"Fehler: {err}"
|
||||
return f"Notiz erstellt: {title}"
|
||||
return f"Notiz erstellt: {title} (id: {d.get('id','?')})"
|
||||
|
||||
|
||||
def create_app():
|
||||
|
||||
Reference in New Issue
Block a user