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:
co-authored by
Claude Opus 4.8
parent
330e0113f8
commit
e34711d0ca
+87
-3
@@ -12,6 +12,8 @@ from starlette.routing import Mount
|
|||||||
|
|
||||||
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
||||||
from common import get_current_user, OAUTH_ROUTES, BearerAuthMiddleware, load_config
|
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()
|
_cfg = load_config()
|
||||||
DATA_API = _cfg["joplin_data_api"] # {user: {url, token}}
|
DATA_API = _cfg["joplin_data_api"] # {user: {url, token}}
|
||||||
@@ -54,6 +56,34 @@ def _post(user, path, json_data):
|
|||||||
return None, str(e)
|
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):
|
def _all_items(user, path, params=None):
|
||||||
"""Fetch all pages from a Data API list endpoint."""
|
"""Fetch all pages from a Data API list endpoint."""
|
||||||
items = []
|
items = []
|
||||||
@@ -157,7 +187,7 @@ def list_note_resources(
|
|||||||
def read_resource(
|
def read_resource(
|
||||||
resource_id: Annotated[str, Field(description="Resource/attachment ID from list_note_resources")],
|
resource_id: Annotated[str, Field(description="Resource/attachment ID from list_note_resources")],
|
||||||
) -> list[TextContent | ImageContent | EmbeddedResource]:
|
) -> 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()
|
user = get_current_user()
|
||||||
if not user: return [TextContent(type="text", text="Error: not authenticated")]
|
if not user: return [TextContent(type="text", text="Error: not authenticated")]
|
||||||
# Get metadata
|
# Get metadata
|
||||||
@@ -174,9 +204,13 @@ def read_resource(
|
|||||||
return [ImageContent(type="image", data=base64.b64encode(content).decode(), mimeType=mime)]
|
return [ImageContent(type="image", data=base64.b64encode(content).decode(), mimeType=mime)]
|
||||||
if mime.startswith("text/"):
|
if mime.startswith("text/"):
|
||||||
return [TextContent(type="text", text=content.decode("utf-8", errors="replace")[:100000])]
|
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(
|
return [EmbeddedResource(type="resource", resource=BlobResourceContents(
|
||||||
uri=f"joplin://resource/{resource_id}/{title}",
|
uri=uri, blob=base64.b64encode(content).decode(), mimeType=mime))]
|
||||||
blob=base64.b64encode(content).decode(), mimeType=mime))]
|
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
@@ -201,6 +235,56 @@ def create_note(
|
|||||||
return f"Notiz erstellt: {title} (id: {d.get('id','?')})"
|
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():
|
def create_app():
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
mcp_app = mcp.streamable_http_app()
|
mcp_app = mcp.streamable_http_app()
|
||||||
|
|||||||
@@ -468,6 +468,35 @@ class TestNotes:
|
|||||||
listing = tool_call(self.PORT, token, "list_notes", {"notebook": self.TEST_NOTEBOOK})
|
listing = tool_call(self.PORT, token, "list_notes", {"notebook": self.TEST_NOTEBOOK})
|
||||||
assert note_id in listing or f"Test Note {tag}" in listing
|
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)
|
# File Type Tests (read_file auf vorbereiteten /testdata-Dateien)
|
||||||
|
|||||||
Reference in New Issue
Block a user