z230

mcp_mongo.py

@@ -262,6 +262,197 @@ def distinct_values(db: str, collection: str, field: str, filter_json: Optional[
         raise
 
 
+@mcp.tool()
+def preview_update(
+    db: str,
+    collection: str,
+    filter_json: Union[str, dict],
+    update_json: Union[str, dict],
+) -> dict:
+    """Preview what update_documents will change — shows affected count and sample of matching docs
+    BEFORE any write. Always call this first, then present the summary to the user for confirmation.
+    filter_json: MongoDB filter, e.g. '{}' for all documents.
+    update_json: MongoDB update operator, e.g. '{"$set": {"Zdroj": "Iluminátor"}}'.
+    """
+    try:
+        col = client[db][collection]
+        filt = parse_filter(filter_json)
+        affected = col.count_documents(filt)
+        sample = [serialize(doc) for doc in col.find(filt).limit(3)]
+        upd = parse_filter(update_json)
+        return {
+            "db": db,
+            "collection": collection,
+            "filter": filt,
+            "update": upd,
+            "affected_count": affected,
+            "sample_docs": sample,
+            "note": "No changes made yet. Present this summary to the user and ask for confirmation before calling update_documents.",
+        }
+    except Exception:
+        log(f"preview_update error: {traceback.format_exc()}")
+        raise
+
+
+@mcp.tool()
+def update_documents(
+    db: str,
+    collection: str,
+    filter_json: Union[str, dict],
+    update_json: Union[str, dict],
+    confirmed: bool = False,
+) -> dict:
+    """Update documents matching filter_json using update_json (MongoDB update operators).
+    REQUIRES confirmed=True — only set this after presenting preview_update output to the user
+    and receiving explicit approval.
+    filter_json example: '{"zeme": "Czech Republic"}'
+    update_json example: '{"$set": {"Zdroj": "Iluminátor"}}'
+    """
+    if not confirmed:
+        return {
+            "status": "aborted",
+            "reason": "confirmed=False. Call preview_update first, show the user what will change, and only proceed with confirmed=True after explicit approval.",
+        }
+    try:
+        col = client[db][collection]
+        filt = parse_filter(filter_json)
+        upd = parse_filter(update_json)
+        result = col.update_many(filt, upd)
+        log(f"update_documents: matched={result.matched_count} modified={result.modified_count}")
+        return {
+            "status": "ok",
+            "db": db,
+            "collection": collection,
+            "filter": filt,
+            "update": upd,
+            "matched_count": result.matched_count,
+            "modified_count": result.modified_count,
+        }
+    except Exception:
+        log(f"update_documents error: {traceback.format_exc()}")
+        raise
+
+
+@mcp.tool()
+def preview_insert(
+    db: str,
+    collection: str,
+    documents_json: Union[str, list],
+) -> dict:
+    """Preview what insert_documents will insert — shows count and the documents themselves
+    BEFORE any write. Always call this first, then present the summary to the user for confirmation.
+    documents_json: JSON array of documents, e.g. '[{"name": "Test", "value": 1}]'.
+    """
+    try:
+        docs = json.loads(documents_json) if isinstance(documents_json, str) else documents_json
+        return {
+            "db": db,
+            "collection": collection,
+            "insert_count": len(docs),
+            "documents": docs,
+            "note": "No changes made yet. Present this summary to the user and ask for confirmation before calling insert_documents.",
+        }
+    except Exception:
+        log(f"preview_insert error: {traceback.format_exc()}")
+        raise
+
+
+@mcp.tool()
+def insert_documents(
+    db: str,
+    collection: str,
+    documents_json: Union[str, list],
+    confirmed: bool = False,
+) -> dict:
+    """Insert one or more documents into a collection.
+    REQUIRES confirmed=True — only set this after presenting preview_insert output to the user
+    and receiving explicit approval.
+    documents_json: JSON array of documents.
+    """
+    if not confirmed:
+        return {
+            "status": "aborted",
+            "reason": "confirmed=False. Call preview_insert first, show the user what will be inserted, and only proceed with confirmed=True after explicit approval.",
+        }
+    try:
+        col = client[db][collection]
+        docs = json.loads(documents_json) if isinstance(documents_json, str) else documents_json
+        result = col.insert_many(docs)
+        log(f"insert_documents: inserted={len(result.inserted_ids)}")
+        return {
+            "status": "ok",
+            "db": db,
+            "collection": collection,
+            "inserted_count": len(result.inserted_ids),
+            "inserted_ids": [str(i) for i in result.inserted_ids],
+        }
+    except Exception:
+        log(f"insert_documents error: {traceback.format_exc()}")
+        raise
+
+
+@mcp.tool()
+def preview_delete(
+    db: str,
+    collection: str,
+    filter_json: Union[str, dict],
+) -> dict:
+    """Preview what delete_documents will remove — shows affected count and sample docs
+    BEFORE any write. Always call this first, then present the summary to the user for confirmation.
+    filter_json: MongoDB filter. Be careful with '{}' — it matches all documents.
+    """
+    try:
+        col = client[db][collection]
+        filt = parse_filter(filter_json)
+        affected = col.count_documents(filt)
+        sample = [serialize(doc) for doc in col.find(filt).limit(3)]
+        return {
+            "db": db,
+            "collection": collection,
+            "filter": filt,
+            "affected_count": affected,
+            "sample_docs": sample,
+            "note": "No changes made yet. Present this summary to the user and ask for confirmation before calling delete_documents.",
+        }
+    except Exception:
+        log(f"preview_delete error: {traceback.format_exc()}")
+        raise
+
+
+@mcp.tool()
+def delete_documents(
+    db: str,
+    collection: str,
+    filter_json: Union[str, dict],
+    confirmed: bool = False,
+) -> dict:
+    """Delete documents matching filter_json.
+    REQUIRES confirmed=True — only set this after presenting preview_delete output to the user
+    and receiving explicit approval.
+    WARNING: '{}' as filter will delete ALL documents in the collection.
+    """
+    if not confirmed:
+        return {
+            "status": "aborted",
+            "reason": "confirmed=False. Call preview_delete first, show the user what will be deleted, and only proceed with confirmed=True after explicit approval.",
+        }
+    try:
+        col = client[db][collection]
+        filt = parse_filter(filter_json)
+        result = col.delete_many(filt)
+        log(f"delete_documents: deleted={result.deleted_count}")
+        return {
+            "status": "ok",
+            "db": db,
+            "collection": collection,
+            "filter": filt,
+            "deleted_count": result.deleted_count,
+        }
+    except Exception:
+        log(f"delete_documents error: {traceback.format_exc()}")
+        raise
+
+
 if __name__ == "__main__":
     log("MCP MongoDB server started (FastMCP)")
     mcp.run()