From e34711d0ca7eade341a9fe7bae7a5ba782cc58f4 Mon Sep 17 00:00:00 2001 From: Stefan Lohmaier Date: Tue, 21 Jul 2026 09:28:48 +0200 Subject: [PATCH] feat(notes): update_note + delete_note -- Notizen bearbeiten/loeschen Notes-MCP konnte nur create/read/search. Neu: - update_note(note_id, title?, body?, notebook?): partielles PUT, nur gesetzte Felder; notebook verschiebt (Name->parent_id-Lookup wie create). - delete_note(note_id, permanent=False): Default Papierkorb, permanent=True loescht endgueltig; Titel wird vorher gelesen fuer klare Rueckmeldung. - Helper _put/_delete analog _post/_get. - Test: voller CRUD-Zyklus in TestNotes (create->update->verify->delete, raeumt hinter sich auf). Alle Notes-Tests gruen (11 passed). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01XhHNJKk3RsHgUsJ27WQSom --- notes/server.py | 90 +++++++++++++++++++++++++++++++++++++++++++++-- tests/test_all.py | 29 +++++++++++++++ 2 files changed, 116 insertions(+), 3 deletions(-) diff --git a/notes/server.py b/notes/server.py index e3bf2d3..55c0c2b 100644 --- a/notes/server.py +++ b/notes/server.py @@ -12,6 +12,8 @@ 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}} @@ -54,6 +56,34 @@ def _post(user, path, json_data): return None, str(e) +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 = [] @@ -157,7 +187,7 @@ def list_note_resources( 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.""" + """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")] # Get metadata @@ -174,9 +204,13 @@ def read_resource( 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=f"joplin://resource/{resource_id}/{title}", - blob=base64.b64encode(content).decode(), mimeType=mime))] + uri=uri, blob=base64.b64encode(content).decode(), mimeType=mime))] @mcp.tool() @@ -201,6 +235,56 @@ def create_note( 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" + 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" + 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() diff --git a/tests/test_all.py b/tests/test_all.py index f6452dc..ee1df3b 100644 --- a/tests/test_all.py +++ b/tests/test_all.py @@ -468,6 +468,35 @@ class TestNotes: listing = tool_call(self.PORT, token, "list_notes", {"notebook": self.TEST_NOTEBOOK}) assert note_id in listing or f"Test Note {tag}" in listing + def test_update_delete_note(self): + """Voller CRUD-Zyklus: create -> update (title+body) -> read verify -> delete permanent.""" + token = get_token(self.PORT) + tag = f"mcptest-upd-{int(time.time())}" + result = tool_call(self.PORT, token, "create_note", { + "notebook": self.TEST_NOTEBOOK, + "title": f"Upd Note {tag}", + "body": "vorher", + }) + assert "erstellt" in result.lower(), f"Create failed: {result}" + note_id = result.split("id: ")[1].rstrip(")").strip() + + # Update title + body + result = tool_call(self.PORT, token, "update_note", { + "note_id": note_id, "title": f"Upd Note {tag} NEU", "body": f"NACHHER {tag}", + }) + assert "aktualisiert" in result.lower(), f"Update failed: {result}" + + body = tool_call(self.PORT, token, "read_note", {"note_id": note_id}) + assert f"NACHHER {tag}" in body and "NEU" in body, f"Update not applied: {body[:200]}" + + # Leerer Update-Aufruf gibt klare Meldung statt Fehler + result = tool_call(self.PORT, token, "update_note", {"note_id": note_id}) + assert "Nichts zu aendern" in result + + # Permanent loeschen (Testdaten nicht liegen lassen) + result = tool_call(self.PORT, token, "delete_note", {"note_id": note_id, "permanent": True}) + assert "ENDGUELTIG" in result, f"Delete failed: {result}" + # ============================================================ # File Type Tests (read_file auf vorbereiteten /testdata-Dateien)