z230

mcp_mongo.py

@@ -48,15 +48,22 @@ def serialize(obj):
 
 
 def parse_filter(filter_json) -> dict:
-    """Parse JSON string or dict to dict. Returns {} on None/empty."""
+    """Parse JSON string or dict to dict. Returns {} on None/empty.
+    Automatically converts string _id values to ObjectId."""
     if not filter_json:
         return {}
-    if isinstance(filter_json, dict):
-        return filter_json
     if isinstance(filter_json, str):
         if not filter_json.strip():
             return {}
-        return json.loads(filter_json)
+        filter_json = json.loads(filter_json)
+    if isinstance(filter_json, dict):
+        if "_id" in filter_json and isinstance(filter_json["_id"], str):
+            try:
+                filter_json = dict(filter_json)
+                filter_json["_id"] = ObjectId(filter_json["_id"])
+            except Exception:
+                pass
+        return filter_json
     return {}
 
 
@@ -453,6 +460,50 @@ def delete_documents(
         raise
 
 
+@mcp.tool()
+def bulk_update(
+    db: str,
+    collection: str,
+    operations_json: Union[str, list],
+    confirmed: bool = False,
+) -> dict:
+    """Run multiple update operations in one call — each with its own filter and update.
+    operations_json: JSON array of {filter, update} objects, e.g.:
+      [{"filter": {"email": "a@b.com"}, "update": {"$set": {"excel": {...}}}}, ...]
+    Each operation runs as update_one. REQUIRES confirmed=True.
+    """
+    if not confirmed:
+        ops = json.loads(operations_json) if isinstance(operations_json, str) else operations_json
+        return {
+            "status": "preview",
+            "operation_count": len(ops),
+            "sample": ops[:3],
+            "note": "No changes made. Call again with confirmed=True to execute.",
+        }
+    try:
+        from pymongo import UpdateOne
+        col = client[db][collection]
+        ops = json.loads(operations_json) if isinstance(operations_json, str) else operations_json
+        requests = [
+            UpdateOne(parse_filter(op["filter"]), parse_filter(op["update"]))
+            for op in ops
+        ]
+        result = col.bulk_write(requests, ordered=False)
+        log(f"bulk_update: matched={result.matched_count} modified={result.modified_count} upserted={result.upserted_count}")
+        return {
+            "status": "ok",
+            "db": db,
+            "collection": collection,
+            "operation_count": len(requests),
+            "matched_count": result.matched_count,
+            "modified_count": result.modified_count,
+            "upserted_count": result.upserted_count,
+        }
+    except Exception:
+        log(f"bulk_update error: {traceback.format_exc()}")
+        raise
+
+
 if __name__ == "__main__":
     log("MCP MongoDB server started (FastMCP)")
     mcp.run()