mcp-home/notes/server.py

"""MCP Notes Server — reads/writes Joplin notes via Joplin CLI Data API."""

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, load_config
from pdfutil import pdf_to_content
from officeutil import office_to_content, is_office

_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})

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 = 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)


import re
_JOPLIN_ID = re.compile(r"^[0-9a-f]{32}$")

def _valid_id(x):
    """Joplin item-IDs sind 32 Hex-Zeichen. Verhindert URL/Query-Injection ueber note_id."""
    return isinstance(x, str) and bool(_JOPLIN_ID.match(x))


def _put(user, path, json_data):
    api = _api(user)
    if not api:
        return None, "Kein Joplin-Zugang"
    try:
        r = httpx.put(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 _delete(user, path, params=None):
    api = _api(user)
    if not api:
        return None, "Kein Joplin-Zugang"
    p = dict(params or {})
    p["token"] = api["token"]
    try:
        r = httpx.delete(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 _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 notebook names and IDs."""
    user = get_current_user()
    if not user: return "Error: not authenticated"
    items, err = _all_items(user, "/folders")
    if err: return f"Fehler: {err}"
    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")] = 50,
) -> str:
    """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:
        folders, err = _all_items(user, "/folders")
        if err: return f"Fehler: {err}"
        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:
        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"{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', 'Rezept', 'Passwort'")],
) -> str:
    """Search notes by keyword (full text). Returns matching note titles and IDs."""
    user = get_current_user()
    if not user: return "Error: not authenticated"
    items, err = _all_items(user, "/search", {"query": query, "type": "note", "fields": "id,title"})
    if err: return f"Fehler: {err}"
    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"
    if not _valid_id(note_id): return "Fehler: ungueltige note_id (erwartet 32 Hex-Zeichen)"
    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"
    if not _valid_id(note_id): return "Fehler: ungueltige note_id (erwartet 32 Hex-Zeichen)"
    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, PDFs and Office documents (docx, xlsx, pptx) as extracted text, text directly, other formats as binary."""
    user = get_current_user()
    if not user: return [TextContent(type="text", text="Error: not authenticated")]
    if not _valid_id(resource_id): return [TextContent(type="text", text="Fehler: ungueltige resource_id (erwartet 32 Hex-Zeichen)")]
    # 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])]
    uri = f"joplin://resource/{resource_id}/{title}"
    if mime == "application/pdf" or title.lower().endswith(".pdf"):
        return pdf_to_content(content, title, mime, uri=uri)
    if is_office(mime, title):
        return office_to_content(content, title, mime, uri=uri)
    return [EmbeddedResource(type="resource", resource=BlobResourceContents(
        uri=uri, blob=base64.b64encode(content).decode(), mimeType=mime))]


@mcp.tool()
def create_note(
    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")],
) -> str:
    """Create a new note in the specified notebook."""
    user = get_current_user()
    if not user: return "Error: not authenticated"
    folders, err = _all_items(user, "/folders")
    if err: return f"Fehler: {err}"
    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} (id: {d.get('id','?')})"


@mcp.tool()
def update_note(
    note_id: Annotated[str, Field(description="Note ID from list_notes or search_notes")],
    title: Annotated[str, Field(description="New title. Leave empty to keep the current title.")] = "",
    body: Annotated[str, Field(description="New content in Markdown, replaces the whole body. Leave empty to keep the current body. Tip: read_note first, modify, then pass the full new text.")] = "",
    notebook: Annotated[str, Field(description="Notebook name or ID to move the note to. Leave empty to keep it where it is.")] = "",
) -> str:
    """Update an existing note: change title and/or body, or move it to another notebook. Only the given fields are changed."""
    user = get_current_user()
    if not user: return "Error: not authenticated"
    if not _valid_id(note_id): return "Fehler: ungueltige note_id (erwartet 32 Hex-Zeichen)"
    payload = {}
    if title: payload["title"] = title
    if body: payload["body"] = body
    if notebook:
        folders, err = _all_items(user, "/folders")
        if err: return f"Fehler: {err}"
        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}"
        payload["parent_id"] = nb_id
    if not payload:
        return "Nichts zu aendern: title, body oder notebook angeben"
    d, err = _put(user, f"/notes/{note_id}", payload)
    if err: return f"Fehler: {err}"
    changed = ", ".join(sorted(payload))
    return f"Notiz aktualisiert ({changed}): {d.get('title','?')} (id: {d.get('id', note_id)})"


@mcp.tool()
def delete_note(
    note_id: Annotated[str, Field(description="Note ID from list_notes or search_notes")],
    permanent: Annotated[bool, Field(description="False (default): move to trash, recoverable in Joplin. True: delete permanently, NOT recoverable.")] = False,
) -> str:
    """Delete a note. By default it goes to Joplin's trash; permanent=True removes it for good."""
    user = get_current_user()
    if not user: return "Error: not authenticated"
    if not _valid_id(note_id): return "Fehler: ungueltige note_id (erwartet 32 Hex-Zeichen)"
    r, err = _get(user, f"/notes/{note_id}", {"fields": "id,title"})
    if err: return f"Fehler: {err}"
    note_title = r.json().get("title", "?")
    params = {"permanent": "1"} if permanent else None
    _, err = _delete(user, f"/notes/{note_id}", params)
    if err: return f"Fehler: {err}"
    if permanent:
        return f"Notiz ENDGUELTIG geloescht: {note_title}"
    return f"Notiz in den Papierkorb verschoben: {note_title} (in Joplin wiederherstellbar)"


def create_app():
    from contextlib import asynccontextmanager
    mcp_app = mcp.streamable_http_app()
    @asynccontextmanager
    async def lifespan(app):
        async with contextlib.AsyncExitStack() as stack:
            await stack.enter_async_context(mcp_app.router.lifespan_context(mcp_app))
            yield
    routes = list(OAUTH_ROUTES) + [Mount("/", app=mcp_app)]
    app = Starlette(routes=routes, lifespan=lifespan)
    app.add_middleware(BearerAuthMiddleware)
    return app

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(create_app(), host="127.0.0.1", port=5104)