mcp-home/notes/server.py

"""MCP Notes Server — reads and writes notes via Joplin API."""

import os, sys, contextlib
from typing import Annotated

import httpx
from pydantic import Field
from mcp.server.fastmcp import FastMCP
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 load_config as _lc
_cfg = _lc()
JOPLIN = _cfg['joplin_url']
JOPLIN_TOKENS = _cfg['joplin_tokens']

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


@mcp.tool()
def list_notebooks() -> str:
    """List all Joplin notebooks (folders). Call this first to see available notebooks."""
    user = get_current_user()
    if not user: return "Error: not authenticated"
    data, err = _api(user, "/api/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"


@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,
) -> str:
    """List notes, optionally filtered by notebook. Shows title and ID. Use read_note with the ID to get content."""
    user = get_current_user()
    if not user: return "Error: not authenticated"
    if notebook:
        data, err = _api(user, "/api/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}")
    else:
        data, err = _api(user, f"/api/notes?limit={limit}&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','')}"


@mcp.tool()
def search_notes(
    query: Annotated[str, Field(description="Search term — matches note titles and content. Example: 'Einkaufsliste', 'Passwort', 'Rezept'")],
) -> str:
    """Search notes by keyword. 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")
    if err: return f"Fehler: {err}"
    return "\n".join(f"{n['title']} (ID: {n['id']})" for n in data.get("items", [])) or "Keine Treffer"


@mcp.tool()
def create_note(
    notebook: Annotated[str, Field(description="Notebook name or ID where the note should be created")],
    title: Annotated[str, Field(description="Note title")],
    body: Annotated[str, Field(description="Note content in Markdown format")],
) -> 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")
    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})
    if err: return f"Fehler: {err}"
    return f"Notiz erstellt: {title}"


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)