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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XhHNJKk3RsHgUsJ27WQSom
This commit is contained in:
Stefan Lohmaier
2026-07-21 09:28:48 +02:00
co-authored by Claude Opus 4.8
parent 330e0113f8
commit e34711d0ca
2 changed files with 116 additions and 3 deletions
+87 -3
View File
@@ -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()