feat(mongodb): 支持聚合查询、批量写入和 MCP 集成

- 桌面端：解析 db.<集合>.aggregate([...]) shell 语法，新增 mongoAggregateDocuments
  接口（Tauri + HTTP）；mongo 入口统一补 ensureConnected；ObjectId(...) 自动
  转换为 {"\$oid":"..."} 扩展 JSON 表示。
- dbx-core：新增 aggregate_documents、insert/update/delete_documents bulk
  操作；扩展 JSON 解析识别 \$oid；schema list_tables_core 补 MongoDB/Elasticsearch
  分支返回 COLLECTION/INDEX。
- dbx-web / Tauri 命令 / MCP bridge：注册 4 个新 mongo 路由
  (aggregate/insert/update/delete documents)；MCP bridge 新增 mongo HTTP 入口
  和 lazy 连接复用。
- node-core (MCP 后端)：mongodb 走专属 executeQuery / listTables /
  describeTable；read-only 默认 + DBX_MCP_ALLOW_WRITES /
  DBX_MCP_ALLOW_DANGEROUS_SQL 两级开关；aggregate 含 \$out/\$merge 走同样开关；
  describeTable 抽样 20 条文档推断字段类型，_id 自动标记主键。
- mcp-server：mongodb 跳过 SQL safety evaluator（mongo shell 与 SQL 语法
  不兼容），改由 backend 做命令级 gating；附注释。
- AiAssistant 同时 emit replaceSql + executeSql 让编辑器和结果保持一致；
  App.vue ensureQueryTab 在无上下文时回退到连接默认库；aiMessageRender
  识别 mongodb / mongo 代码块。
- Review 修复：合并两份重复的 evaluateMongo*Safety 实现（同时修复 web-backend
  用字符串比较导致空 filter 检查可被空格绕过的 bug）；ObjectId 正则兼容
  单/双引号；mongo_ops 错误文案 "MCP" 改为 "bulk"；补 aggregate 解析测试用例。

测试：pnpm test 911 通过；vue-tsc 通过；cargo check -p dbx-core 通过。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

apps/desktop/src/App.vue

@@ -622,7 +622,7 @@ function ensureQueryTab(): string {
   const tab = activeTab.value;
   if (tab && tab.mode === "query") return tab.id;
   const connId = connectionStore.activeConnectionId || connectionStore.connections[0]?.id || "";
-  const db = tab?.database || "";
+  const db = tab?.database || connectionStore.getConfig(connId)?.database || "";
   return queryStore.createTab(connId, db, undefined, "query");
 }
 

apps/desktop/src/components/editor/AiAssistant.vue

@@ -556,6 +556,7 @@ function applySql(code: string) {
 }
 
 function executeSql(code: string) {
+  emit("replaceSql", code);
   emit("executeSql", code);
 }
 

apps/desktop/src/lib/aiMessageRender.ts

@@ -35,6 +35,8 @@ const SQL_LANGUAGES = new Map([
   ["sqlite", "SQLITE"],
   ["tsql", "TSQL"],
   ["clickhouse", "CLICKHOUSE"],
+  ["mongodb", "MONGODB"],
+  ["mongo", "MONGODB"],
 ]);
 const SHELL_LANGUAGES = new Map([
   ["bash", "BASH"],

apps/desktop/src/lib/api.ts

@@ -208,6 +208,7 @@ export const redisLoadMore = forward("redisLoadMore");
 export const mongoListDatabases = forward("mongoListDatabases");
 export const mongoListCollections = forward("mongoListCollections");
 export const mongoFindDocuments = forward("mongoFindDocuments");
+export const mongoAggregateDocuments = forward("mongoAggregateDocuments");
 export const mongoInsertDocument = forward("mongoInsertDocument");
 export const mongoUpdateDocument = forward("mongoUpdateDocument");
 export const mongoDeleteDocument = forward("mongoDeleteDocument");

apps/desktop/src/lib/http.ts

@@ -1267,6 +1267,15 @@ export async function mongoFindDocuments(
   return post("/api/mongo/find-documents", { connectionId, database, collection, skip, limit, filter, sort });
 }
 
+export async function mongoAggregateDocuments(
+  connectionId: string,
+  database: string,
+  collection: string,
+  pipelineJson: string,
+): Promise<MongoDocumentResult> {
+  return post("/api/mongo/aggregate-documents", { connectionId, database, collection, pipelineJson });
+}
+
 export async function mongoInsertDocument(
   connectionId: string,
   database: string,

apps/desktop/src/lib/mongoShellCommand.ts

@@ -13,6 +13,11 @@ export interface MongoCountDocumentsCommand {
   filter: string;
 }
 
+export interface MongoAggregateCommand {
+  collection: string;
+  pipeline: string;
+}
+
 const DEFAULT_LIMIT = 100;
 
 export function parseMongoFindCommand(input: string): MongoFindCommand | null {
@@ -72,6 +77,31 @@ export function parseMongoCountDocumentsCommand(input: string): MongoCountDocume
   };
 }
 
+export function parseMongoAggregateCommand(input: string): MongoAggregateCommand | null {
+  const source = input.trim().replace(/;$/, "").trim();
+  const target = parseCollectionMethodTarget(source, "aggregate");
+  if (!target) return null;
+
+  const openIndex = source.indexOf("(", target.methodCallIndex);
+  const closeIndex = findMatchingParen(source, openIndex);
+  if (closeIndex < 0 || source.slice(closeIndex + 1).trim()) return null;
+
+  const args = splitTopLevel(source.slice(openIndex + 1, closeIndex));
+  if (args.length !== 1) return null;
+  const pipeline = normalizeJsonArgument(args[0]);
+  if (!pipeline) return null;
+  try {
+    if (!Array.isArray(JSON.parse(pipeline))) return null;
+  } catch {
+    return null;
+  }
+
+  return {
+    collection: target.collection,
+    pipeline,
+  };
+}
+
 export function mongoDocumentsToQueryResult(documents: unknown[], executionTimeMs: number, total: number): QueryResult {
   const columns: string[] = [];
 
@@ -146,9 +176,10 @@ function parseCollectionMethodTarget(
 function normalizeJsonArgument(value: string): string | null {
   const trimmed = value.trim();
   if (!trimmed) return "{}";
+  const preprocessed = trimmed.replace(/ObjectId\s*\(\s*["']([^"']+)["']\s*\)/g, '{"$oid":"$1"}');
   try {
-    JSON.parse(trimmed);
-    return trimmed;
+    JSON.parse(preprocessed);
+    return preprocessed;
   } catch {
     return null;
   }

apps/desktop/src/lib/tauri.ts

@@ -1098,6 +1098,15 @@ export async function mongoFindDocuments(
   return invoke("mongo_find_documents", { connectionId, database, collection, skip, limit, filter, sort });
 }
 
+export async function mongoAggregateDocuments(
+  connectionId: string,
+  database: string,
+  collection: string,
+  pipelineJson: string,
+): Promise<MongoDocumentResult> {
+  return invoke("mongo_aggregate_documents", { connectionId, database, collection, pipelineJson });
+}
+
 export async function mongoInsertDocument(
   connectionId: string,
   database: string,

apps/desktop/src/stores/queryStore.ts

@@ -12,6 +12,7 @@ import { restoreOpenTabsState, serializeOpenTabs } from "@/lib/openTabsPersisten
 import {
   mongoCountToQueryResult,
   mongoDocumentsToQueryResult,
+  parseMongoAggregateCommand,
   parseMongoCountDocumentsCommand,
   parseMongoFindCommand,
 } from "@/lib/mongoShellCommand";
@@ -589,6 +590,7 @@ export const useQueryStore = defineStore("query", () => {
       }
       const mongoFind = conn?.db_type === "mongodb" ? parseMongoFindCommand(sql) : null;
       if (mongoFind) {
+        await connStore.ensureConnected(tab.connectionId);
         console.info("[DBX][executeTabSql:mongo-find:start]", { traceId, collection: mongoFind.collection });
         const result = await api.mongoFindDocuments(
           tab.connectionId,
@@ -621,6 +623,7 @@ export const useQueryStore = defineStore("query", () => {
       }
       const mongoCount = conn?.db_type === "mongodb" ? parseMongoCountDocumentsCommand(sql) : null;
       if (mongoCount) {
+        await connStore.ensureConnected(tab.connectionId);
         console.info("[DBX][executeTabSql:mongo-count:start]", { traceId, collection: mongoCount.collection });
         const result = await api.mongoFindDocuments(
           tab.connectionId,
@@ -650,6 +653,37 @@ export const useQueryStore = defineStore("query", () => {
         return;
       }
 
+      const mongoAggregate = conn?.db_type === "mongodb" ? parseMongoAggregateCommand(sql) : null;
+      if (mongoAggregate) {
+        await connStore.ensureConnected(tab.connectionId);
+        console.info("[DBX][executeTabSql:mongo-aggregate:start]", { traceId, collection: mongoAggregate.collection });
+        const result = await api.mongoAggregateDocuments(
+          tab.connectionId,
+          tab.database,
+          mongoAggregate.collection,
+          mongoAggregate.pipeline,
+        );
+        console.info("[DBX][executeTabSql:mongo-aggregate:done]", {
+          traceId,
+          rowCount: result.documents.length,
+          total: result.total,
+          elapsed: elapsed(),
+        });
+        const current = tabs.value.find((t) => t.id === id);
+        if (current?.executionId === executionId) {
+          current.results = undefined;
+          current.activeResultIndex = undefined;
+          current.result = mongoDocumentsToQueryResult(result.documents, performance.now() - startedAt, result.total);
+          current.queryAnalysis = undefined;
+          current.querySourceColumns = undefined;
+          current.queryEditabilityReason = undefined;
+          current.tableMeta = undefined;
+          current.resultBaseSql = options?.resultBaseSql ?? sql;
+          current.resultSortedSql = options?.resultSortedSql;
+        }
+        return;
+      }
+
       console.info("[DBX][executeTabSql:execute-multi:start]", { traceId, elapsed: elapsed() });
       const executionOptions = {
         ...(typeof pageLimit === "number"

crates/dbx-core/src/db/mongo_driver.rs

@@ -1,5 +1,5 @@
 use mongodb::{
-    bson::{doc, Bson, Document},
+    bson::{doc, oid::ObjectId, Bson, Document},
     Client,
 };
 use serde::{Deserialize, Serialize};
@@ -50,7 +50,7 @@ pub async fn find_documents(
     let filter_doc: Document = match filter {
         Some(f) if !f.trim().is_empty() => {
             let json: serde_json::Value = serde_json::from_str(f).map_err(|e| format!("Invalid filter JSON: {e}"))?;
-            mongodb::bson::to_document(&json).map_err(|e| format!("Invalid filter: {e}"))?
+            json_object_to_document(&json)?
         }
         _ => doc! {},
     };
@@ -61,7 +61,7 @@ pub async fn find_documents(
     if let Some(s) = sort {
         if !s.trim().is_empty() {
             let json: serde_json::Value = serde_json::from_str(s).map_err(|e| format!("Invalid sort JSON: {e}"))?;
-            let sort_doc = mongodb::bson::to_document(&json).map_err(|e| format!("Invalid sort: {e}"))?;
+            let sort_doc = json_object_to_document(&json).map_err(|e| format!("Invalid sort: {e}"))?;
             find = find.sort(sort_doc);
         }
     }
@@ -78,6 +78,30 @@ pub async fn find_documents(
     Ok(MongoDocumentResult { documents, total })
 }
 
+pub async fn aggregate_documents(
+    client: &Client,
+    database: &str,
+    collection: &str,
+    pipeline_json: &str,
+) -> Result<MongoDocumentResult, String> {
+    let json: serde_json::Value =
+        serde_json::from_str(pipeline_json).map_err(|e| format!("Invalid pipeline JSON: {e}"))?;
+    let pipeline_values = json.as_array().ok_or_else(|| "Aggregate pipeline must be a JSON array".to_string())?;
+    let pipeline = pipeline_values
+        .iter()
+        .map(|value| json_object_to_document(value).map_err(|e| format!("Invalid pipeline stage: {e}")))
+        .collect::<Result<Vec<Document>, String>>()?;
+    let col = client.database(database).collection::<Document>(collection);
+    let mut cursor = col.aggregate(pipeline).await.map_err(|e| e.to_string())?;
+    let mut documents = Vec::new();
+    while cursor.advance().await.map_err(|e| e.to_string())? {
+        let doc = cursor.deserialize_current().map_err(|e| e.to_string())?;
+        documents.push(bson_to_json(&Bson::Document(doc)));
+    }
+    let total = documents.len() as u64;
+    Ok(MongoDocumentResult { documents, total })
+}
+
 pub async fn insert_document(
     client: &Client,
     database: &str,
@@ -90,6 +114,28 @@ pub async fn insert_document(
     Ok(format!("{}", result.inserted_id))
 }
 
+pub async fn insert_documents(
+    client: &Client,
+    database: &str,
+    collection: &str,
+    docs_json: &str,
+) -> Result<u64, String> {
+    let json: serde_json::Value = serde_json::from_str(docs_json).map_err(|e| format!("Invalid JSON: {e}"))?;
+    let docs = match json {
+        serde_json::Value::Array(values) => values
+            .into_iter()
+            .map(|value| json_object_to_document(&value).map_err(|e| format!("Invalid document: {e}")))
+            .collect::<Result<Vec<Document>, String>>()?,
+        value => vec![json_object_to_document(&value).map_err(|e| format!("Invalid document: {e}"))?],
+    };
+    if docs.is_empty() {
+        return Ok(0);
+    }
+    let col = client.database(database).collection::<Document>(collection);
+    let result = col.insert_many(docs).await.map_err(|e| e.to_string())?;
+    Ok(result.inserted_ids.len() as u64)
+}
+
 pub async fn update_document(
     client: &Client,
     database: &str,
@@ -105,6 +151,29 @@ pub async fn update_document(
     Ok(result.modified_count)
 }
 
+pub async fn update_documents(
+    client: &Client,
+    database: &str,
+    collection: &str,
+    filter_json: &str,
+    update_json: &str,
+    many: bool,
+) -> Result<u64, String> {
+    let filter_value: serde_json::Value =
+        serde_json::from_str(filter_json).map_err(|e| format!("Invalid filter JSON: {e}"))?;
+    let update_value: serde_json::Value =
+        serde_json::from_str(update_json).map_err(|e| format!("Invalid update JSON: {e}"))?;
+    let filter = json_object_to_document(&filter_value).map_err(|e| format!("Invalid filter: {e}"))?;
+    let update = json_object_to_document(&update_value).map_err(|e| format!("Invalid update: {e}"))?;
+    let col = client.database(database).collection::<Document>(collection);
+    let result = if many {
+        col.update_many(filter, update).await.map_err(|e| e.to_string())?
+    } else {
+        col.update_one(filter, update).await.map_err(|e| e.to_string())?
+    };
+    Ok(result.modified_count)
+}
+
 pub async fn delete_document(client: &Client, database: &str, collection: &str, id: &str) -> Result<u64, String> {
     let oid = mongodb::bson::oid::ObjectId::parse_str(id).map_err(|e| format!("Invalid ObjectId: {e}"))?;
     let col = client.database(database).collection::<Document>(collection);
@@ -112,6 +181,25 @@ pub async fn delete_document(client: &Client, database: &str, collection: &str,
     Ok(result.deleted_count)
 }
 
+pub async fn delete_documents(
+    client: &Client,
+    database: &str,
+    collection: &str,
+    filter_json: &str,
+    many: bool,
+) -> Result<u64, String> {
+    let filter_value: serde_json::Value =
+        serde_json::from_str(filter_json).map_err(|e| format!("Invalid filter JSON: {e}"))?;
+    let filter = json_object_to_document(&filter_value).map_err(|e| format!("Invalid filter: {e}"))?;
+    let col = client.database(database).collection::<Document>(collection);
+    let result = if many {
+        col.delete_many(filter).await.map_err(|e| e.to_string())?
+    } else {
+        col.delete_one(filter).await.map_err(|e| e.to_string())?
+    };
+    Ok(result.deleted_count)
+}
+
 fn bson_to_json(bson: &Bson) -> serde_json::Value {
     match bson {
         Bson::Double(v) => serde_json::json!(v),
@@ -133,3 +221,42 @@ fn bson_to_json(bson: &Bson) -> serde_json::Value {
         _ => serde_json::Value::String(format!("{bson}")),
     }
 }
+
+/// Convert a `serde_json::Value` (JSON object) to a BSON `Document`,
+/// handling MongoDB extended JSON conventions such as `{"$oid":"..."}`.
+pub fn json_object_to_document(value: &serde_json::Value) -> Result<Document, String> {
+    match json_value_to_bson(value) {
+        Bson::Document(doc) => Ok(doc),
+        other => Err(format!("Expected a JSON object, got {other:?}")),
+    }
+}
+
+fn json_value_to_bson(value: &serde_json::Value) -> Bson {
+    match value {
+        serde_json::Value::Null => Bson::Null,
+        serde_json::Value::Bool(b) => Bson::Boolean(*b),
+        serde_json::Value::Number(n) => {
+            if let Some(i) = n.as_i64() {
+                Bson::Int64(i)
+            } else if let Some(f) = n.as_f64() {
+                Bson::Double(f)
+            } else {
+                Bson::Null
+            }
+        }
+        serde_json::Value::String(s) => Bson::String(s.clone()),
+        serde_json::Value::Array(arr) => Bson::Array(arr.iter().map(json_value_to_bson).collect()),
+        serde_json::Value::Object(obj) => {
+            // Extended JSON: {"$oid":"..."} → BSON ObjectId
+            if obj.len() == 1 {
+                if let Some(serde_json::Value::String(hex)) = obj.get("$oid") {
+                    if let Ok(oid) = ObjectId::parse_str(hex) {
+                        return Bson::ObjectId(oid);
+                    }
+                }
+            }
+            let doc: Document = obj.iter().map(|(k, v)| (k.clone(), json_value_to_bson(v))).collect();
+            Bson::Document(doc)
+        }
+    }
+}

crates/dbx-core/src/mongo_ops.rs

@@ -71,6 +71,23 @@ pub async fn mongo_find_documents_core(
     }
 }
 
+pub async fn mongo_aggregate_documents_core(
+    state: &AppState,
+    connection_id: &str,
+    database: &str,
+    collection: &str,
+    pipeline_json: &str,
+) -> Result<MongoDocumentResult, String> {
+    let connections = state.connections.read().await;
+    match connections.get(connection_id).ok_or("Not found")? {
+        PoolKind::MongoDb(client) => {
+            mongo_driver::aggregate_documents(client, database, collection, pipeline_json).await
+        }
+        PoolKind::Agent(_) => Err("MongoDB legacy agent does not support aggregate".to_string()),
+        _ => Err("Not a MongoDB connection".to_string()),
+    }
+}
+
 pub async fn mongo_insert_document_core(
     state: &AppState,
     connection_id: &str,
@@ -101,6 +118,21 @@ pub async fn mongo_insert_document_core(
     }
 }
 
+pub async fn mongo_insert_documents_core(
+    state: &AppState,
+    connection_id: &str,
+    database: &str,
+    collection: &str,
+    docs_json: &str,
+) -> Result<u64, String> {
+    let connections = state.connections.read().await;
+    match connections.get(connection_id).ok_or("Not found")? {
+        PoolKind::MongoDb(client) => mongo_driver::insert_documents(client, database, collection, docs_json).await,
+        PoolKind::Agent(_) => Err("MongoDB legacy agent does not support bulk insertMany/insertOne writes".to_string()),
+        _ => Err("Not a MongoDB connection".to_string()),
+    }
+}
+
 pub async fn mongo_update_document_core(
     state: &AppState,
     connection_id: &str,
@@ -133,6 +165,25 @@ pub async fn mongo_update_document_core(
     }
 }
 
+pub async fn mongo_update_documents_core(
+    state: &AppState,
+    connection_id: &str,
+    database: &str,
+    collection: &str,
+    filter_json: &str,
+    update_json: &str,
+    many: bool,
+) -> Result<u64, String> {
+    let connections = state.connections.read().await;
+    match connections.get(connection_id).ok_or("Not found")? {
+        PoolKind::MongoDb(client) => {
+            mongo_driver::update_documents(client, database, collection, filter_json, update_json, many).await
+        }
+        PoolKind::Agent(_) => Err("MongoDB legacy agent does not support bulk updateOne/updateMany writes".to_string()),
+        _ => Err("Not a MongoDB connection".to_string()),
+    }
+}
+
 pub async fn mongo_delete_document_core(
     state: &AppState,
     connection_id: &str,
@@ -157,3 +208,21 @@ pub async fn mongo_delete_document_core(
         _ => Err("Not a MongoDB/Elasticsearch connection".to_string()),
     }
 }
+
+pub async fn mongo_delete_documents_core(
+    state: &AppState,
+    connection_id: &str,
+    database: &str,
+    collection: &str,
+    filter_json: &str,
+    many: bool,
+) -> Result<u64, String> {
+    let connections = state.connections.read().await;
+    match connections.get(connection_id).ok_or("Not found")? {
+        PoolKind::MongoDb(client) => {
+            mongo_driver::delete_documents(client, database, collection, filter_json, many).await
+        }
+        PoolKind::Agent(_) => Err("MongoDB legacy agent does not support bulk deleteOne/deleteMany writes".to_string()),
+        _ => Err("Not a MongoDB connection".to_string()),
+    }
+}

crates/dbx-core/src/schema.rs

@@ -386,10 +386,29 @@ pub async fn list_tables_core(
         PoolKind::Sqlite(p) => {
             db::sqlite::list_tables(p, schema).await.map(|tables| filter_table_infos(tables, filter, limit))
         }
+        PoolKind::MongoDb(client) => db::mongo_driver::list_collections(client, database)
+            .await
+            .map(|names| collection_names_to_tables(names, "COLLECTION"))
+            .map(|tables| filter_table_infos(tables, filter, limit)),
+        PoolKind::Elasticsearch(client) => db::elasticsearch_driver::list_indices(client)
+            .await
+            .map(|names| collection_names_to_tables(names, "INDEX"))
+            .map(|tables| filter_table_infos(tables, filter, limit)),
         _ => Ok(vec![]),
     }
 }
 
+fn collection_names_to_tables(names: Vec<String>, table_type: &str) -> Vec<db::TableInfo> {
+    names
+        .into_iter()
+        .map(|name| db::TableInfo {
+            name,
+            table_type: table_type.to_string(),
+            comment: None,
+        })
+        .collect()
+}
+
 fn filter_table_infos(tables: Vec<db::TableInfo>, filter: Option<&str>, limit: Option<usize>) -> Vec<db::TableInfo> {
     let filter = filter.unwrap_or("").to_lowercase();
     let limit = limit.unwrap_or(usize::MAX);

crates/dbx-web/src/main.rs

@@ -220,9 +220,13 @@ async fn main() {
         .route("/mongo/list-databases", post(routes::mongo::list_databases))
         .route("/mongo/list-collections", post(routes::mongo::list_collections))
         .route("/mongo/find-documents", post(routes::mongo::find_documents))
+        .route("/mongo/aggregate-documents", post(routes::mongo::aggregate_documents))
         .route("/mongo/insert-document", post(routes::mongo::insert_document))
+        .route("/mongo/insert-documents", post(routes::mongo::insert_documents))
         .route("/mongo/update-document", post(routes::mongo::update_document))
+        .route("/mongo/update-documents", post(routes::mongo::update_documents))
         .route("/mongo/delete-document", post(routes::mongo::delete_document))
+        .route("/mongo/delete-documents", post(routes::mongo::delete_documents))
         // History
         .route("/history", get(routes::history::load_history).delete(routes::history::clear_history))
         .route("/history/save", post(routes::history::save_history))

crates/dbx-web/src/routes/mongo.rs

@@ -32,6 +32,15 @@ pub struct MongoFindRequest {
     pub sort: Option<String>,
 }
 
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoAggregateRequest {
+    pub connection_id: String,
+    pub database: String,
+    pub collection: String,
+    pub pipeline_json: String,
+}
+
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct MongoInsertRequest {
@@ -41,6 +50,36 @@ pub struct MongoInsertRequest {
     pub doc_json: String,
 }
 
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoInsertDocumentsRequest {
+    pub connection_id: String,
+    pub database: String,
+    pub collection: String,
+    pub docs_json: String,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoUpdateDocumentsRequest {
+    pub connection_id: String,
+    pub database: String,
+    pub collection: String,
+    pub filter_json: String,
+    pub update_json: String,
+    pub many: bool,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MongoDeleteDocumentsRequest {
+    pub connection_id: String,
+    pub database: String,
+    pub collection: String,
+    pub filter_json: String,
+    pub many: bool,
+}
+
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct MongoUpdateRequest {
@@ -98,6 +137,22 @@ pub async fn find_documents(
     Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?))
 }
 
+pub async fn aggregate_documents(
+    State(state): State<Arc<WebState>>,
+    Json(req): Json<MongoAggregateRequest>,
+) -> Result<Json<serde_json::Value>, AppError> {
+    let result = dbx_core::mongo_ops::mongo_aggregate_documents_core(
+        &state.app,
+        &req.connection_id,
+        &req.database,
+        &req.collection,
+        &req.pipeline_json,
+    )
+    .await
+    .map_err(AppError)?;
+    Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?))
+}
+
 pub async fn insert_document(
     State(state): State<Arc<WebState>>,
     Json(req): Json<MongoInsertRequest>,
@@ -114,6 +169,22 @@ pub async fn insert_document(
     Ok(Json(result))
 }
 
+pub async fn insert_documents(
+    State(state): State<Arc<WebState>>,
+    Json(req): Json<MongoInsertDocumentsRequest>,
+) -> Result<Json<serde_json::Value>, AppError> {
+    let result = dbx_core::mongo_ops::mongo_insert_documents_core(
+        &state.app,
+        &req.connection_id,
+        &req.database,
+        &req.collection,
+        &req.docs_json,
+    )
+    .await
+    .map_err(AppError)?;
+    Ok(Json(serde_json::json!({ "affected_rows": result })))
+}
+
 pub async fn update_document(
     State(state): State<Arc<WebState>>,
     Json(req): Json<MongoUpdateRequest>,
@@ -131,6 +202,24 @@ pub async fn update_document(
     Ok(Json(result))
 }
 
+pub async fn update_documents(
+    State(state): State<Arc<WebState>>,
+    Json(req): Json<MongoUpdateDocumentsRequest>,
+) -> Result<Json<serde_json::Value>, AppError> {
+    let result = dbx_core::mongo_ops::mongo_update_documents_core(
+        &state.app,
+        &req.connection_id,
+        &req.database,
+        &req.collection,
+        &req.filter_json,
+        &req.update_json,
+        req.many,
+    )
+    .await
+    .map_err(AppError)?;
+    Ok(Json(serde_json::json!({ "affected_rows": result })))
+}
+
 pub async fn delete_document(
     State(state): State<Arc<WebState>>,
     Json(req): Json<MongoDeleteRequest>,
@@ -146,3 +235,20 @@ pub async fn delete_document(
     .map_err(AppError)?;
     Ok(Json(result))
 }
+
+pub async fn delete_documents(
+    State(state): State<Arc<WebState>>,
+    Json(req): Json<MongoDeleteDocumentsRequest>,
+) -> Result<Json<serde_json::Value>, AppError> {
+    let result = dbx_core::mongo_ops::mongo_delete_documents_core(
+        &state.app,
+        &req.connection_id,
+        &req.database,
+        &req.collection,
+        &req.filter_json,
+        req.many,
+    )
+    .await
+    .map_err(AppError)?;
+    Ok(Json(serde_json::json!({ "affected_rows": result })))
+}

packages/app-tests/mongoShellCommand.test.ts

@@ -3,6 +3,7 @@ import test from "node:test";
 import {
   mongoCountToQueryResult,
   mongoDocumentsToQueryResult,
+  parseMongoAggregateCommand,
   parseMongoCountDocumentsCommand,
   parseMongoFindCommand,
 } from "../../apps/desktop/src/lib/mongoShellCommand.ts";
@@ -43,6 +44,38 @@ test("parseMongoCountDocumentsCommand parses db collection countDocuments", () =
   });
 });
 
+test("parseMongoAggregateCommand parses db collection aggregate", () => {
+  assert.deepEqual(parseMongoAggregateCommand('db.products.aggregate([{"$match":{"active":true}},{"$count":"total"}])'), {
+    collection: "products",
+    pipeline: '[{"$match":{"active":true}},{"$count":"total"}]',
+  });
+});
+
+test("parseMongoAggregateCommand accepts an empty pipeline", () => {
+  assert.deepEqual(parseMongoAggregateCommand("db.products.aggregate([])"), {
+    collection: "products",
+    pipeline: "[]",
+  });
+});
+
+test("parseMongoAggregateCommand rejects non-array pipelines and extra arguments", () => {
+  assert.equal(parseMongoAggregateCommand('db.products.aggregate({"$match":{}})'), null);
+  assert.equal(parseMongoAggregateCommand("db.products.aggregate([], {})"), null);
+  assert.equal(parseMongoAggregateCommand("db.products.aggregate([]).limit(10)"), null);
+});
+
+test("parseMongoAggregateCommand normalises ObjectId arguments with either quote style", () => {
+  const oid = "507f1f77bcf86cd799439011";
+  for (const quote of ["\"", "'"]) {
+    const command = parseMongoAggregateCommand(
+      `db.orders.aggregate([{"$match":{"_id":ObjectId(${quote}${oid}${quote})}}])`,
+    );
+    assert.ok(command, `quote=${quote} should parse`);
+    assert.equal(command.collection, "orders");
+    assert.deepEqual(JSON.parse(command.pipeline), [{ "$match": { "_id": { "$oid": oid } } }]);
+  }
+});
+
 test("mongoCountToQueryResult returns a single count row", () => {
   assert.deepEqual(mongoCountToQueryResult(42, 5), {
     columns: ["count"],

packages/mcp-server/src/index.ts

@@ -102,8 +102,12 @@ export function createDbxMcpServer(backend: Backend, options: { isWebMode?: bool
   async ({ connection_name, database, sql }) => {
     const config = await backend.findConnection(connection_name);
     if (!config) return text(`Connection "${connection_name}" not found`);
-    const safety = evaluateSqlSafety(sql, sqlSafetyFromEnv());
-    if (!safety.allowed) return text(`Query blocked: ${safety.reason}`);
+    if (config.db_type !== "mongodb") {
+      const safety = evaluateSqlSafety(sql, sqlSafetyFromEnv());
+      if (!safety.allowed) return text(`Query blocked: ${safety.reason}`);
+    }
+    // MongoDB shell commands don't fit the SQL safety evaluator; the backend
+    // (node-core executeQuery) applies command-aware read/write gating.
     try {
       const result = await backend.executeQuery(withDatabase(config, database), sql);
       if (result.columns.length === 0) return text(`Query executed. ${result.row_count} row(s) affected.`);
@@ -211,8 +215,13 @@ export function createDbxMcpServer(backend: Backend, options: { isWebMode?: bool
       database: z.string().optional().describe("Database name"),
     },
     async ({ connection_name, sql, database }) => {
-      const safety = evaluateSqlSafety(sql, sqlSafetyFromEnv());
-      if (!safety.allowed) return text(`Query blocked: ${safety.reason}`);
+      const config = await backend.findConnection(connection_name);
+      if (config?.db_type !== "mongodb") {
+        const safety = evaluateSqlSafety(sql, sqlSafetyFromEnv());
+        if (!safety.allowed) return text(`Query blocked: ${safety.reason}`);
+      }
+      // MongoDB shell commands bypass the SQL safety evaluator; the desktop
+      // app's executor applies command-aware read/write gating.
       return bridgeRequest("/execute-query", { connection_name, sql, database }, "Query sent to DBX");
     },
     );

packages/mcp-server/tests/server.test.ts

@@ -53,3 +53,67 @@ test("execute query scopes the connection to the requested database", async () =
 
   assert.equal(usedDatabase, "stores_demo");
 });
+
+test("mongodb list tables returns collections from the selected database", async () => {
+  let usedDatabase = "";
+  const mongoConnection: ConnectionConfig = { ...connection, db_type: "mongodb", database: "admin" };
+  const scopedBackend: Backend = {
+    ...backend,
+    findConnection: async () => mongoConnection,
+    listTables: async (config) => {
+      usedDatabase = config.database || "";
+      return [{ name: "projects", type: "COLLECTION" }];
+    },
+  };
+  const server = createDbxMcpServer(scopedBackend, { isWebMode: true });
+
+  const result = await (server as any)._registeredTools.dbx_list_tables.handler({
+    connection_name: "local",
+    database: "pystrument",
+  });
+
+  assert.equal(usedDatabase, "pystrument");
+  assert.match(result.content[0].text, /projects/);
+  assert.match(result.content[0].text, /COLLECTION/);
+});
+
+test("mongodb describe table returns inferred document fields", async () => {
+  const mongoConnection: ConnectionConfig = { ...connection, db_type: "mongodb" };
+  const scopedBackend: Backend = {
+    ...backend,
+    findConnection: async () => mongoConnection,
+    describeTable: async () => [
+      { name: "_id", data_type: "object", is_nullable: false, column_default: null, is_primary_key: true, comment: null },
+      { name: "name", data_type: "string", is_nullable: false, column_default: null, is_primary_key: false, comment: null },
+    ],
+  };
+  const server = createDbxMcpServer(scopedBackend, { isWebMode: true });
+
+  const result = await (server as any)._registeredTools.dbx_describe_table.handler({
+    connection_name: "local",
+    database: "pystrument",
+    table: "projects",
+  });
+
+  assert.match(result.content[0].text, /_id \(PK\)/);
+  assert.match(result.content[0].text, /name/);
+});
+
+test("mongodb execute query formats shell-style find results", async () => {
+  const mongoConnection: ConnectionConfig = { ...connection, db_type: "mongodb" };
+  const scopedBackend: Backend = {
+    ...backend,
+    findConnection: async () => mongoConnection,
+    executeQuery: async () => ({ columns: ["_id", "name"], rows: [{ _id: "1", name: "demo" }], row_count: 1 }),
+  };
+  const server = createDbxMcpServer(scopedBackend, { isWebMode: true });
+
+  const result = await (server as any)._registeredTools.dbx_execute_query.handler({
+    connection_name: "local",
+    database: "pystrument",
+    sql: "db.projects.find({}).limit(1)",
+  });
+
+  assert.match(result.content[0].text, /demo/);
+  assert.match(result.content[0].text, /1 row\(s\)/);
+});

packages/node-core/src/database.ts

@@ -272,6 +272,11 @@ interface BridgeColumnInfo {
   comment: string | null;
 }
 
+interface MongoDocumentResult {
+  documents: unknown[];
+  total: number;
+}
+
 function bridgeAppDataDir(): string {
   const home = homedir();
   switch (platform()) {
@@ -407,6 +412,35 @@ function sqliteQuery(config: ConnectionConfig, sql: string, options?: QueryOptio
 }
 
 export async function executeQuery(config: ConnectionConfig, sql: string, options?: QueryOptions): Promise<QueryResult> {
+  if (config.db_type === "mongodb") {
+    const find = parseMongoFindCommand(sql);
+    if (find) {
+      const result = await withTimeout(mongoFindDocuments(config, find.collection, find.skip, find.limit, find.filter, find.sort), resolveTimeoutMs(options));
+      return mongoDocumentsToQueryResult(result.documents.slice(0, resolveMaxRows(options)), result.total);
+    }
+    const count = parseMongoCountDocumentsCommand(sql);
+    if (count) {
+      const result = await withTimeout(mongoFindDocuments(config, count.collection, 0, 1, count.filter), resolveTimeoutMs(options));
+      return { columns: ["count"], rows: [{ count: result.total }], row_count: 1 };
+    }
+    const aggregate = parseMongoAggregateCommand(sql);
+    if (aggregate) {
+      const safety = evaluateMongoAggregateSafety(aggregate, sqlSafetyFromEnv());
+      if (!safety.allowed) throw new Error(safety.reason);
+      const result = await withTimeout(mongoAggregateDocuments(config, aggregate.collection, aggregate.pipeline), resolveTimeoutMs(options));
+      return mongoDocumentsToQueryResult(result.documents.slice(0, resolveMaxRows(options)), result.total);
+    }
+    const write = parseMongoWriteCommand(sql);
+    if (write) {
+      const safety = evaluateMongoWriteSafety(write, sqlSafetyFromEnv());
+      if (!safety.allowed) throw new Error(safety.reason);
+      const affected = await withTimeout(executeMongoWrite(config, write), resolveTimeoutMs(options));
+      return { columns: [], rows: [], row_count: affected };
+    }
+    throw new Error(
+      "Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.projects.countDocuments({}), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})",
+    );
+  }
   if (isDirectType(config.db_type)) {
     return query(config, sql, undefined, options);
   }
@@ -419,6 +453,14 @@ export async function executeQuery(config: ConnectionConfig, sql: string, option
 }
 
 export async function listTables(config: ConnectionConfig, schema?: string): Promise<TableInfo[]> {
+  if (config.db_type === "mongodb") {
+    const collections = await bridgeDataRequest<string[]>("/data/mongo/list-collections", {
+      connection_name: config.name,
+      database: config.database || "",
+      schema: schema || "",
+    });
+    return collections.map((name) => ({ name, type: "COLLECTION" }));
+  }
   if (config.db_type === "sqlite") {
     const result = await query(
       config,
@@ -448,6 +490,10 @@ export async function listTables(config: ConnectionConfig, schema?: string): Pro
 }
 
 export async function describeTable(config: ConnectionConfig, table: string, schema?: string): Promise<ColumnInfo[]> {
+  if (config.db_type === "mongodb") {
+    const result = await mongoFindDocuments(config, table, 0, 20, "{}");
+    return inferMongoColumns(result.documents);
+  }
   if (config.db_type === "sqlite") {
     const result = await query(config, `PRAGMA table_info(${quoteSqliteIdentifier(table)})`);
     return result.rows.map((r) => ({
@@ -498,3 +544,393 @@ export async function describeTable(config: ConnectionConfig, table: string, sch
     comment: r.comment != null ? String(r.comment) : null,
   }));
 }
+
+async function mongoFindDocuments(
+  config: ConnectionConfig,
+  collection: string,
+  skip: number,
+  limit: number,
+  filter: string,
+  sort?: string,
+): Promise<MongoDocumentResult> {
+  return bridgeDataRequest<MongoDocumentResult>("/data/mongo/find-documents", {
+    connection_name: config.name,
+    database: config.database || "",
+    collection,
+    skip,
+    limit,
+    filter,
+    sort,
+  });
+}
+
+async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCommand): Promise<number> {
+  if (command.kind === "insert") {
+    const result = await bridgeDataRequest<{ affected_rows: number }>("/data/mongo/insert-documents", {
+      connection_name: config.name,
+      database: config.database || "",
+      collection: command.collection,
+      docs_json: command.docsJson,
+    });
+    return result.affected_rows;
+  }
+  if (command.kind === "update") {
+    const result = await bridgeDataRequest<{ affected_rows: number }>("/data/mongo/update-documents", {
+      connection_name: config.name,
+      database: config.database || "",
+      collection: command.collection,
+      filter_json: command.filter,
+      update_json: command.update,
+      many: command.many,
+    });
+    return result.affected_rows;
+  }
+  const result = await bridgeDataRequest<{ affected_rows: number }>("/data/mongo/delete-documents", {
+    connection_name: config.name,
+    database: config.database || "",
+    collection: command.collection,
+    filter_json: command.filter,
+    many: command.many,
+  });
+  return result.affected_rows;
+}
+
+async function mongoAggregateDocuments(
+  config: ConnectionConfig,
+  collection: string,
+  pipelineJson: string,
+): Promise<MongoDocumentResult> {
+  return bridgeDataRequest<MongoDocumentResult>("/data/mongo/aggregate-documents", {
+    connection_name: config.name,
+    database: config.database || "",
+    collection,
+    pipeline_json: pipelineJson,
+  });
+}
+
+export function mongoDocumentsToQueryResult(documents: unknown[], total: number): QueryResult {
+  const columns: string[] = [];
+  for (const doc of documents) {
+    if (isRecord(doc)) {
+      for (const key of Object.keys(doc)) {
+        if (!columns.includes(key)) columns.push(key);
+      }
+    } else if (!columns.includes("value")) {
+      columns.push("value");
+    }
+  }
+  const rows = documents.map((doc) => {
+    const row: Record<string, unknown> = {};
+    for (const column of columns) {
+      row[column] = isRecord(doc) ? toCellValue(doc[column]) : column === "value" ? toCellValue(doc) : null;
+    }
+    return row;
+  });
+  return { columns, rows, row_count: rows.length };
+}
+
+export function inferMongoColumns(documents: unknown[]): ColumnInfo[] {
+  const columns = new Map<string, { types: Set<string>; nullable: boolean }>();
+  for (const doc of documents) {
+    if (!isRecord(doc)) {
+      const entry = columns.get("value") ?? { types: new Set<string>(), nullable: false };
+      entry.types.add(mongoTypeName(doc));
+      columns.set("value", entry);
+      continue;
+    }
+    for (const [name, value] of Object.entries(doc)) {
+      const entry = columns.get(name) ?? { types: new Set<string>(), nullable: false };
+      entry.types.add(mongoTypeName(value));
+      if (value === null || value === undefined) entry.nullable = true;
+      columns.set(name, entry);
+    }
+  }
+  return Array.from(columns.entries()).map(([name, entry]) => ({
+    name,
+    data_type: Array.from(entry.types).sort().join(" | ") || "unknown",
+    is_nullable: entry.nullable,
+    column_default: null,
+    is_primary_key: name === "_id",
+    comment: null,
+  }));
+}
+
+interface MongoFindCommand {
+  collection: string;
+  filter: string;
+  skip: number;
+  limit: number;
+  sort?: string;
+}
+
+interface MongoCountDocumentsCommand {
+  collection: string;
+  filter: string;
+}
+
+interface MongoAggregateCommand {
+  collection: string;
+  pipeline: string;
+}
+
+export type MongoWriteCommand =
+  | { kind: "insert"; collection: string; docsJson: string }
+  | { kind: "update"; collection: string; filter: string; update: string; many: boolean }
+  | { kind: "delete"; collection: string; filter: string; many: boolean };
+
+export function parseMongoFindCommand(input: string): MongoFindCommand | null {
+  const source = input.trim().replace(/;$/, "").trim();
+  const target = parseCollectionMethodTarget(source, "find");
+  if (!target) return null;
+  const findOpenIndex = source.indexOf("(", target.methodCallIndex);
+  const findCloseIndex = findMatchingParen(source, findOpenIndex);
+  if (findCloseIndex < 0) return null;
+  const findArgs = splitTopLevel(source.slice(findOpenIndex + 1, findCloseIndex));
+  const filter = normalizeJsonArgument(findArgs[0] || "{}");
+  if (!filter) return null;
+  const chain = source.slice(findCloseIndex + 1).trim();
+  if (chain && !chain.startsWith(".")) return null;
+  const sortArg = readChainedCallArgument(chain, "sort");
+  let sort: string | undefined;
+  if (sortArg !== undefined) {
+    const parsedSort = normalizeJsonArgument(sortArg);
+    if (!parsedSort) return null;
+    sort = parsedSort;
+  }
+  const skip = readChainedIntegerArgument(chain, "skip", 0);
+  const limit = readChainedIntegerArgument(chain, "limit", MAX_ROWS);
+  if (skip === null || limit === null) return null;
+  return { collection: target.collection, filter, skip, limit, sort };
+}
+
+export function parseMongoCountDocumentsCommand(input: string): MongoCountDocumentsCommand | null {
+  const source = input.trim().replace(/;$/, "").trim();
+  const target = parseCollectionMethodTarget(source, "countDocuments");
+  if (!target) return null;
+  const openIndex = source.indexOf("(", target.methodCallIndex);
+  const closeIndex = findMatchingParen(source, openIndex);
+  if (closeIndex < 0 || source.slice(closeIndex + 1).trim()) return null;
+  const args = splitTopLevel(source.slice(openIndex + 1, closeIndex));
+  if (args.length > 1 && args.slice(1).some((arg) => arg.trim())) return null;
+  const filter = normalizeJsonArgument(args[0] || "{}");
+  return filter ? { collection: target.collection, filter } : null;
+}
+
+export function parseMongoAggregateCommand(input: string): MongoAggregateCommand | null {
+  const source = input.trim().replace(/;$/, "").trim();
+  const target = parseCollectionMethodTarget(source, "aggregate");
+  if (!target) return null;
+  const args = parseMethodArgs(source, target.methodCallIndex);
+  if (!args || args.length !== 1) return null;
+  const pipeline = normalizeJsonArgument(args[0]);
+  if (!pipeline) return null;
+  return Array.isArray(JSON.parse(pipeline)) ? { collection: target.collection, pipeline } : null;
+}
+
+export function mongoAggregateWriteStage(pipelineJson: string): "$out" | "$merge" | null {
+  try {
+    const pipeline = JSON.parse(pipelineJson);
+    if (!Array.isArray(pipeline)) return null;
+    for (const stage of pipeline) {
+      if (!isRecord(stage)) continue;
+      if (Object.prototype.hasOwnProperty.call(stage, "$out")) return "$out";
+      if (Object.prototype.hasOwnProperty.call(stage, "$merge")) return "$merge";
+    }
+    return null;
+  } catch {
+    return null;
+  }
+}
+
+export function parseMongoWriteCommand(input: string): MongoWriteCommand | null {
+  const source = input.trim().replace(/;$/, "").trim();
+  const insertOne = parseCollectionMethodTarget(source, "insertOne");
+  if (insertOne) {
+    const args = parseMethodArgs(source, insertOne.methodCallIndex);
+    if (!args || args.length !== 1) return null;
+    const doc = normalizeJsonArgument(args[0]);
+    return doc ? { kind: "insert", collection: insertOne.collection, docsJson: doc } : null;
+  }
+
+  const insertMany = parseCollectionMethodTarget(source, "insertMany");
+  if (insertMany) {
+    const args = parseMethodArgs(source, insertMany.methodCallIndex);
+    if (!args || args.length !== 1) return null;
+    const docs = normalizeJsonArgument(args[0]);
+    if (!docs) return null;
+    return Array.isArray(JSON.parse(docs)) ? { kind: "insert", collection: insertMany.collection, docsJson: docs } : null;
+  }
+
+  for (const method of ["updateOne", "updateMany"] as const) {
+    const target = parseCollectionMethodTarget(source, method);
+    if (!target) continue;
+    const args = parseMethodArgs(source, target.methodCallIndex);
+    if (!args || args.length !== 2) return null;
+    const filter = normalizeJsonArgument(args[0]);
+    const update = normalizeJsonArgument(args[1]);
+    if (!filter || !update) return null;
+    return { kind: "update", collection: target.collection, filter, update, many: method === "updateMany" };
+  }
+
+  for (const method of ["deleteOne", "deleteMany"] as const) {
+    const target = parseCollectionMethodTarget(source, method);
+    if (!target) continue;
+    const args = parseMethodArgs(source, target.methodCallIndex);
+    if (!args || args.length !== 1) return null;
+    const filter = normalizeJsonArgument(args[0]);
+    if (!filter) return null;
+    return { kind: "delete", collection: target.collection, filter, many: method === "deleteMany" };
+  }
+
+  return null;
+}
+
+export function evaluateMongoWriteSafety(
+  command: MongoWriteCommand,
+  options: { allowWrites?: boolean; allowDangerous?: boolean },
+): { allowed: boolean; reason?: string } {
+  if (!options.allowWrites) {
+    return {
+      allowed: false,
+      reason: "MCP MongoDB execution is read-only by default. Set DBX_MCP_ALLOW_WRITES=1 to allow write commands.",
+    };
+  }
+  if (!options.allowDangerous && (command.kind === "update" || command.kind === "delete") && isEmptyJsonObject(command.filter)) {
+    return {
+      allowed: false,
+      reason: "MongoDB update/delete commands must include a non-empty filter unless DBX_MCP_ALLOW_DANGEROUS_SQL=1 is set.",
+    };
+  }
+  return { allowed: true };
+}
+
+export function evaluateMongoAggregateSafety(
+  command: MongoAggregateCommand,
+  options: { allowWrites?: boolean; allowDangerous?: boolean },
+): { allowed: boolean; reason?: string } {
+  const writeStage = mongoAggregateWriteStage(command.pipeline);
+  if (!writeStage) return { allowed: true };
+  if (!options.allowWrites) {
+    return {
+      allowed: false,
+      reason: `MongoDB aggregate stage "${writeStage}" writes data. Set DBX_MCP_ALLOW_WRITES=1 to allow write commands.`,
+    };
+  }
+  if (!options.allowDangerous) {
+    return {
+      allowed: false,
+      reason: `MongoDB aggregate stage "${writeStage}" is dangerous. Set DBX_MCP_ALLOW_DANGEROUS_SQL=1 to allow it.`,
+    };
+  }
+  return { allowed: true };
+}
+
+function parseCollectionMethodTarget(source: string, method: string): { collection: string; methodCallIndex: number } | null {
+  const direct = new RegExp(`^db\\.([A-Za-z_$][\\w$]*)\\.${method}\\s*\\(`).exec(source);
+  if (direct) return { collection: direct[1], methodCallIndex: source.indexOf(`.${method}`) };
+  const quoted = new RegExp(`^db\\.getCollection\\(\\s*(['"])([^'"]+)\\1\\s*\\)\\.${method}\\s*\\(`).exec(source);
+  if (quoted) return { collection: quoted[2], methodCallIndex: source.indexOf(`.${method}`) };
+  return null;
+}
+
+function parseMethodArgs(source: string, methodCallIndex: number): string[] | null {
+  const openIndex = source.indexOf("(", methodCallIndex);
+  const closeIndex = findMatchingParen(source, openIndex);
+  if (closeIndex < 0 || source.slice(closeIndex + 1).trim()) return null;
+  return splitTopLevel(source.slice(openIndex + 1, closeIndex));
+}
+
+function readChainedCallArgument(chain: string, method: string): string | undefined {
+  const pattern = new RegExp(`\\.${method}\\s*\\(`, "g");
+  const match = pattern.exec(chain);
+  if (!match) return undefined;
+  const openIndex = match.index + match[0].lastIndexOf("(");
+  const closeIndex = findMatchingParen(chain, openIndex);
+  return closeIndex < 0 ? undefined : chain.slice(openIndex + 1, closeIndex);
+}
+
+function readChainedIntegerArgument(chain: string, method: string, fallback: number): number | null {
+  const arg = readChainedCallArgument(chain, method);
+  if (arg === undefined) return fallback;
+  if (!/^\d+$/.test(arg.trim())) return null;
+  return Number(arg.trim());
+}
+
+function normalizeJsonArgument(arg: string): string | null {
+  const value = (arg.trim() || "{}").replace(/ObjectId\s*\(\s*["']([^"']+)["']\s*\)/g, '{"$oid":"$1"}');
+  try {
+    JSON.parse(value);
+    return value;
+  } catch {
+    return null;
+  }
+}
+
+function isEmptyJsonObject(json: string): boolean {
+  try {
+    const parsed = JSON.parse(json);
+    return isRecord(parsed) && Object.keys(parsed).length === 0;
+  } catch {
+    return false;
+  }
+}
+
+function splitTopLevel(source: string): string[] {
+  const parts: string[] = [];
+  let depth = 0;
+  let start = 0;
+  let quote: string | null = null;
+  for (let i = 0; i < source.length; i += 1) {
+    const ch = source[i];
+    if (quote) {
+      if (ch === "\\" && i + 1 < source.length) i += 1;
+      else if (ch === quote) quote = null;
+      continue;
+    }
+    if (ch === "'" || ch === '"') quote = ch;
+    else if (ch === "{" || ch === "[" || ch === "(") depth += 1;
+    else if (ch === "}" || ch === "]" || ch === ")") depth -= 1;
+    else if (ch === "," && depth === 0) {
+      parts.push(source.slice(start, i));
+      start = i + 1;
+    }
+  }
+  parts.push(source.slice(start));
+  return parts;
+}
+
+function findMatchingParen(source: string, openIndex: number): number {
+  if (openIndex < 0 || source[openIndex] !== "(") return -1;
+  let depth = 0;
+  let quote: string | null = null;
+  for (let i = openIndex; i < source.length; i += 1) {
+    const ch = source[i];
+    if (quote) {
+      if (ch === "\\" && i + 1 < source.length) i += 1;
+      else if (ch === quote) quote = null;
+      continue;
+    }
+    if (ch === "'" || ch === '"') quote = ch;
+    else if (ch === "(") depth += 1;
+    else if (ch === ")") {
+      depth -= 1;
+      if (depth === 0) return i;
+    }
+  }
+  return -1;
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function mongoTypeName(value: unknown): string {
+  if (value === null || value === undefined) return "null";
+  if (Array.isArray(value)) return "array";
+  if (isRecord(value)) return "object";
+  return typeof value;
+}
+
+function toCellValue(value: unknown): unknown {
+  return typeof value === "object" && value !== null ? JSON.stringify(value) : value;
+}

packages/node-core/src/web-backend.ts

@@ -1,5 +1,17 @@
 import type { ConnectionConfig } from "./connections.js";
 import type { TableInfo, ColumnInfo, QueryOptions, QueryResult } from "./database.js";
+import {
+  evaluateMongoAggregateSafety,
+  evaluateMongoWriteSafety,
+  inferMongoColumns,
+  mongoDocumentsToQueryResult,
+  parseMongoAggregateCommand,
+  parseMongoCountDocumentsCommand,
+  parseMongoFindCommand,
+  parseMongoWriteCommand,
+  type MongoWriteCommand,
+} from "./database.js";
+import { sqlSafetyFromEnv } from "./sql-safety.js";
 
 const baseUrl = process.env.DBX_WEB_URL!.replace(/\/+$/, "");
 const password = process.env.DBX_WEB_PASSWORD || "";
@@ -86,6 +98,14 @@ async function ensureConnected(config: ConnectionConfig): Promise<void> {
 
 export async function listTables(config: ConnectionConfig, schema?: string): Promise<TableInfo[]> {
   await ensureConnected(config);
+  if (config.db_type === "mongodb") {
+    const res = await apiFetch("/api/mongo/list-collections", {
+      method: "POST",
+      body: JSON.stringify({ connectionId: config.id, database: config.database || "" }),
+    });
+    const collections = (await res.json()) as string[];
+    return collections.map((name) => ({ name, type: "COLLECTION" }));
+  }
   const params = new URLSearchParams({
     connection_id: config.id,
     database: config.database || "",
@@ -97,6 +117,14 @@ export async function listTables(config: ConnectionConfig, schema?: string): Pro
 
 export async function describeTable(config: ConnectionConfig, table: string, schema?: string): Promise<ColumnInfo[]> {
   await ensureConnected(config);
+  if (config.db_type === "mongodb") {
+    const res = await apiFetch("/api/mongo/find-documents", {
+      method: "POST",
+      body: JSON.stringify({ connectionId: config.id, database: config.database || "", collection: table, skip: 0, limit: 20, filter: "{}" }),
+    });
+    const result = (await res.json()) as { documents: unknown[]; total: number };
+    return inferMongoColumns(result.documents);
+  }
   const params = new URLSearchParams({
     connection_id: config.id,
     database: config.database || "",
@@ -109,6 +137,67 @@ export async function describeTable(config: ConnectionConfig, table: string, sch
 
 export async function executeQuery(config: ConnectionConfig, sql: string, options?: QueryOptions): Promise<QueryResult> {
   await ensureConnected(config);
+  if (config.db_type === "mongodb") {
+    const find = parseMongoFindCommand(sql);
+    if (find) {
+      const res = await apiFetch("/api/mongo/find-documents", {
+        method: "POST",
+        body: JSON.stringify({
+          connectionId: config.id,
+          database: config.database || "",
+          collection: find.collection,
+          skip: find.skip,
+          limit: find.limit,
+          filter: find.filter,
+          sort: find.sort,
+        }),
+      });
+      const result = (await res.json()) as { documents: unknown[]; total: number };
+      return mongoDocumentsToQueryResult(result.documents.slice(0, options?.maxRows ?? result.documents.length), result.total);
+    }
+    const count = parseMongoCountDocumentsCommand(sql);
+    if (count) {
+      const res = await apiFetch("/api/mongo/find-documents", {
+        method: "POST",
+        body: JSON.stringify({
+          connectionId: config.id,
+          database: config.database || "",
+          collection: count.collection,
+          skip: 0,
+          limit: 1,
+          filter: count.filter,
+        }),
+      });
+      const result = (await res.json()) as { documents: unknown[]; total: number };
+      return { columns: ["count"], rows: [{ count: result.total }], row_count: 1 };
+    }
+    const aggregate = parseMongoAggregateCommand(sql);
+    if (aggregate) {
+      const safety = evaluateMongoAggregateSafety(aggregate, sqlSafetyFromEnv());
+      if (!safety.allowed) throw new Error(safety.reason);
+      const res = await apiFetch("/api/mongo/aggregate-documents", {
+        method: "POST",
+        body: JSON.stringify({
+          connectionId: config.id,
+          database: config.database || "",
+          collection: aggregate.collection,
+          pipelineJson: aggregate.pipeline,
+        }),
+      });
+      const result = (await res.json()) as { documents: unknown[]; total: number };
+      return mongoDocumentsToQueryResult(result.documents.slice(0, options?.maxRows ?? result.documents.length), result.total);
+    }
+    const write = parseMongoWriteCommand(sql);
+    if (write) {
+      const safety = evaluateMongoWriteSafety(write, sqlSafetyFromEnv());
+      if (!safety.allowed) throw new Error(safety.reason);
+      const affected = await executeMongoWrite(config, write);
+      return { columns: [], rows: [], row_count: affected };
+    }
+    throw new Error(
+      "Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.projects.countDocuments({}), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})",
+    );
+  }
   const res = await apiFetch("/api/query/execute", {
     method: "POST",
     body: JSON.stringify({
@@ -128,3 +217,47 @@ export async function executeQuery(config: ConnectionConfig, sql: string, option
   const limitedRows = rows.slice(0, options?.maxRows ?? rows.length);
   return { columns: data.columns, rows: limitedRows, row_count: limitedRows.length };
 }
+
+async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCommand): Promise<number> {
+  if (command.kind === "insert") {
+    const res = await apiFetch("/api/mongo/insert-documents", {
+      method: "POST",
+      body: JSON.stringify({
+        connectionId: config.id,
+        database: config.database || "",
+        collection: command.collection,
+        docsJson: command.docsJson,
+      }),
+    });
+    const result = (await res.json()) as { affected_rows: number };
+    return result.affected_rows;
+  }
+  if (command.kind === "update") {
+    const res = await apiFetch("/api/mongo/update-documents", {
+      method: "POST",
+      body: JSON.stringify({
+        connectionId: config.id,
+        database: config.database || "",
+        collection: command.collection,
+        filterJson: command.filter,
+        updateJson: command.update,
+        many: command.many,
+      }),
+    });
+    const result = (await res.json()) as { affected_rows: number };
+    return result.affected_rows;
+  }
+  const res = await apiFetch("/api/mongo/delete-documents", {
+    method: "POST",
+    body: JSON.stringify({
+      connectionId: config.id,
+      database: config.database || "",
+      collection: command.collection,
+      filterJson: command.filter,
+      many: command.many,
+    }),
+  });
+  const result = (await res.json()) as { affected_rows: number };
+  return result.affected_rows;
+}
+

packages/node-core/tests/mongo-query.test.ts

@@ -0,0 +1,152 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import {
+  executeQuery,
+  inferMongoColumns,
+  mongoAggregateWriteStage,
+  mongoDocumentsToQueryResult,
+  parseMongoAggregateCommand,
+  parseMongoCountDocumentsCommand,
+  parseMongoFindCommand,
+  parseMongoWriteCommand,
+} from "../src/database.js";
+
+test("parseMongoFindCommand accepts shell-style find commands", () => {
+  assert.deepEqual(parseMongoFindCommand('db.getCollection("operation_logs").find({"level":"info"}).sort({"ts":-1}).skip(5).limit(10)'), {
+    collection: "operation_logs",
+    filter: '{"level":"info"}',
+    skip: 5,
+    limit: 10,
+    sort: '{"ts":-1}',
+  });
+});
+
+test("parseMongoCountDocumentsCommand accepts shell-style count commands", () => {
+  assert.deepEqual(parseMongoCountDocumentsCommand('db.projects.countDocuments({"active":true})'), {
+    collection: "projects",
+    filter: '{"active":true}',
+  });
+});
+
+test("parseMongoAggregateCommand accepts aggregate pipelines", () => {
+  assert.deepEqual(parseMongoAggregateCommand('db.projects.aggregate([{"$match":{"active":true}},{"$group":{"_id":"$owner","total":{"$sum":1}}}])'), {
+    collection: "projects",
+    pipeline: '[{"$match":{"active":true}},{"$group":{"_id":"$owner","total":{"$sum":1}}}]',
+  });
+});
+
+test("mongoAggregateWriteStage detects write stages", () => {
+  assert.equal(mongoAggregateWriteStage('[{"$match":{"active":true}}]'), null);
+  assert.equal(mongoAggregateWriteStage('[{"$match":{}},{"$out":"projects_dump"}]'), "$out");
+  assert.equal(mongoAggregateWriteStage('[{"$merge":{"into":"projects_dump"}}]'), "$merge");
+});
+
+test("mongodb executeQuery blocks aggregate write stages without both env flags", async () => {
+  const oldAllowWrites = process.env.DBX_MCP_ALLOW_WRITES;
+  const oldAllowDangerous = process.env.DBX_MCP_ALLOW_DANGEROUS_SQL;
+  delete process.env.DBX_MCP_ALLOW_WRITES;
+  delete process.env.DBX_MCP_ALLOW_DANGEROUS_SQL;
+  const config = {
+    id: "mongo",
+    name: "mongo",
+    db_type: "mongodb",
+    host: "127.0.0.1",
+    port: 27017,
+    username: "",
+    password: "",
+    database: "app",
+    ssh_enabled: false,
+    ssl: false,
+  } as const;
+
+  await assert.rejects(executeQuery(config, 'db.projects.aggregate([{"$out":"projects_dump"}])'), /DBX_MCP_ALLOW_WRITES=1/);
+
+  process.env.DBX_MCP_ALLOW_WRITES = "1";
+  await assert.rejects(
+    executeQuery(config, 'db.projects.aggregate([{"$merge":{"into":"projects_dump"}}])'),
+    /DBX_MCP_ALLOW_DANGEROUS_SQL=1/,
+  );
+
+  if (oldAllowWrites === undefined) delete process.env.DBX_MCP_ALLOW_WRITES;
+  else process.env.DBX_MCP_ALLOW_WRITES = oldAllowWrites;
+  if (oldAllowDangerous === undefined) delete process.env.DBX_MCP_ALLOW_DANGEROUS_SQL;
+  else process.env.DBX_MCP_ALLOW_DANGEROUS_SQL = oldAllowDangerous;
+});
+
+test("parseMongoWriteCommand accepts supported write commands", () => {
+  assert.deepEqual(parseMongoWriteCommand('db.projects.insertOne({"name":"demo"})'), {
+    kind: "insert",
+    collection: "projects",
+    docsJson: '{"name":"demo"}',
+  });
+  assert.deepEqual(parseMongoWriteCommand('db.projects.updateOne({"_id":"1"},{"$set":{"name":"next"}})'), {
+    kind: "update",
+    collection: "projects",
+    filter: '{"_id":"1"}',
+    update: '{"$set":{"name":"next"}}',
+    many: false,
+  });
+  assert.deepEqual(parseMongoWriteCommand('db.projects.deleteMany({"stale":true})'), {
+    kind: "delete",
+    collection: "projects",
+    filter: '{"stale":true}',
+    many: true,
+  });
+});
+
+test("mongodb executeQuery blocks writes without the write env flag", async () => {
+  const oldAllowWrites = process.env.DBX_MCP_ALLOW_WRITES;
+  delete process.env.DBX_MCP_ALLOW_WRITES;
+  await assert.rejects(
+    executeQuery(
+      {
+        id: "mongo",
+        name: "mongo",
+        db_type: "mongodb",
+        host: "127.0.0.1",
+        port: 27017,
+        username: "",
+        password: "",
+        database: "app",
+        ssh_enabled: false,
+        ssl: false,
+      },
+      'db.projects.insertOne({"name":"demo"})',
+    ),
+    /DBX_MCP_ALLOW_WRITES=1/,
+  );
+  if (oldAllowWrites === undefined) delete process.env.DBX_MCP_ALLOW_WRITES;
+  else process.env.DBX_MCP_ALLOW_WRITES = oldAllowWrites;
+});
+
+test("mongoDocumentsToQueryResult turns documents into rows", () => {
+  assert.deepEqual(mongoDocumentsToQueryResult([{ _id: "1", nested: { ok: true } }, { _id: "2", name: "demo" }], 2), {
+    columns: ["_id", "nested", "name"],
+    rows: [
+      { _id: "1", nested: '{"ok":true}', name: undefined },
+      { _id: "2", nested: undefined, name: "demo" },
+    ],
+    row_count: 2,
+  });
+});
+
+test("inferMongoColumns marks _id as primary and reports observed types", () => {
+  assert.deepEqual(inferMongoColumns([{ _id: "1", active: true }, { _id: "2", active: null }]), [
+    {
+      name: "_id",
+      data_type: "string",
+      is_nullable: false,
+      column_default: null,
+      is_primary_key: true,
+      comment: null,
+    },
+    {
+      name: "active",
+      data_type: "boolean | null",
+      is_nullable: true,
+      column_default: null,
+      is_primary_key: false,
+      comment: null,
+    },
+  ]);
+});

src-tauri/src/commands/mcp_bridge.rs

@@ -39,6 +39,52 @@ struct DescribeTableRequest {
     table: String,
 }
 
+#[derive(Deserialize)]
+struct MongoFindDocumentsRequest {
+    connection_name: String,
+    database: Option<String>,
+    collection: String,
+    skip: Option<u64>,
+    limit: Option<i64>,
+    filter: Option<String>,
+    sort: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct MongoAggregateDocumentsRequest {
+    connection_name: String,
+    database: Option<String>,
+    collection: String,
+    pipeline_json: String,
+}
+
+#[derive(Deserialize)]
+struct MongoInsertDocumentsRequest {
+    connection_name: String,
+    database: Option<String>,
+    collection: String,
+    docs_json: String,
+}
+
+#[derive(Deserialize)]
+struct MongoUpdateDocumentsRequest {
+    connection_name: String,
+    database: Option<String>,
+    collection: String,
+    filter_json: String,
+    update_json: String,
+    many: bool,
+}
+
+#[derive(Deserialize)]
+struct MongoDeleteDocumentsRequest {
+    connection_name: String,
+    database: Option<String>,
+    collection: String,
+    filter_json: String,
+    many: bool,
+}
+
 #[derive(Clone, Serialize)]
 pub struct McpOpenTableEvent {
     pub connection_id: String,
@@ -92,6 +138,18 @@ pub fn start(app_handle: AppHandle, state: Arc<AppState>) {
                     handle_list_tables_data(&st, body, &mut stream).await;
                 } else if first_line.starts_with("POST /data/describe-table") {
                     handle_describe_table_data(&st, body, &mut stream).await;
+                } else if first_line.starts_with("POST /data/mongo/list-collections") {
+                    handle_mongo_list_collections_data(&st, body, &mut stream).await;
+                } else if first_line.starts_with("POST /data/mongo/find-documents") {
+                    handle_mongo_find_documents_data(&st, body, &mut stream).await;
+                } else if first_line.starts_with("POST /data/mongo/aggregate-documents") {
+                    handle_mongo_aggregate_documents_data(&st, body, &mut stream).await;
+                } else if first_line.starts_with("POST /data/mongo/insert-documents") {
+                    handle_mongo_insert_documents_data(&st, body, &mut stream).await;
+                } else if first_line.starts_with("POST /data/mongo/update-documents") {
+                    handle_mongo_update_documents_data(&st, body, &mut stream).await;
+                } else if first_line.starts_with("POST /data/mongo/delete-documents") {
+                    handle_mongo_delete_documents_data(&st, body, &mut stream).await;
                 } else if first_line.starts_with("POST /data/execute-query") {
                     handle_execute_query_data(&st, body, &mut stream).await;
                 } else if first_line.starts_with("POST /execute-query") {
@@ -148,6 +206,30 @@ async fn resolve_connection(
     Ok(config)
 }
 
+async fn resolve_mongo_pool_key(
+    state: &Arc<AppState>,
+    connection_name: &str,
+    database: Option<String>,
+    stream: &mut tokio::net::TcpStream,
+) -> Option<(String, String)> {
+    let config = match resolve_connection(state, connection_name).await {
+        Ok(c) => c,
+        Err(e) => {
+            respond_error(stream, "404 Not Found", &e).await;
+            return None;
+        }
+    };
+    let database = database.unwrap_or_else(|| config.database.clone().unwrap_or_default());
+    let pool_key = match state.get_or_create_pool(&config.id, Some(&database)).await {
+        Ok(key) => key,
+        Err(e) => {
+            respond_error(stream, "500 Internal Server Error", &e).await;
+            return None;
+        }
+    };
+    Some((pool_key, database))
+}
+
 async fn handle_open_table(app: &AppHandle, state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
     let req: OpenTableRequest = match serde_json::from_str(body) {
         Ok(r) => r,
@@ -251,6 +333,154 @@ async fn handle_describe_table_data(state: &Arc<AppState>, body: &str, stream: &
     }
 }
 
+async fn handle_mongo_list_collections_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
+    let req: ListTablesRequest = match serde_json::from_str(body) {
+        Ok(r) => r,
+        Err(_) => {
+            respond_error(stream, "400 Bad Request", "Invalid JSON").await;
+            return;
+        }
+    };
+    let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await
+    else {
+        return;
+    };
+    match dbx_core::mongo_ops::mongo_list_collections_core(state, &pool_key, &database).await {
+        Ok(collections) => respond_json(stream, &collections).await,
+        Err(e) => respond_error(stream, "500 Internal Server Error", &e).await,
+    }
+}
+
+async fn handle_mongo_find_documents_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
+    let req: MongoFindDocumentsRequest = match serde_json::from_str(body) {
+        Ok(r) => r,
+        Err(_) => {
+            respond_error(stream, "400 Bad Request", "Invalid JSON").await;
+            return;
+        }
+    };
+    let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await
+    else {
+        return;
+    };
+    match dbx_core::mongo_ops::mongo_find_documents_core(
+        state,
+        &pool_key,
+        &database,
+        &req.collection,
+        req.skip.unwrap_or(0),
+        req.limit.unwrap_or(100),
+        req.filter.as_deref(),
+        req.sort.as_deref(),
+    )
+    .await
+    {
+        Ok(result) => respond_json(stream, &result).await,
+        Err(e) => respond_error(stream, "500 Internal Server Error", &e).await,
+    }
+}
+
+async fn handle_mongo_aggregate_documents_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
+    let req: MongoAggregateDocumentsRequest = match serde_json::from_str(body) {
+        Ok(r) => r,
+        Err(_) => {
+            respond_error(stream, "400 Bad Request", "Invalid JSON").await;
+            return;
+        }
+    };
+    let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await
+    else {
+        return;
+    };
+    match dbx_core::mongo_ops::mongo_aggregate_documents_core(
+        state,
+        &pool_key,
+        &database,
+        &req.collection,
+        &req.pipeline_json,
+    )
+    .await
+    {
+        Ok(result) => respond_json(stream, &result).await,
+        Err(e) => respond_error(stream, "500 Internal Server Error", &e).await,
+    }
+}
+
+async fn handle_mongo_insert_documents_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
+    let req: MongoInsertDocumentsRequest = match serde_json::from_str(body) {
+        Ok(r) => r,
+        Err(_) => {
+            respond_error(stream, "400 Bad Request", "Invalid JSON").await;
+            return;
+        }
+    };
+    let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await
+    else {
+        return;
+    };
+    match dbx_core::mongo_ops::mongo_insert_documents_core(state, &pool_key, &database, &req.collection, &req.docs_json)
+        .await
+    {
+        Ok(inserted) => respond_json(stream, &serde_json::json!({ "affected_rows": inserted })).await,
+        Err(e) => respond_error(stream, "500 Internal Server Error", &e).await,
+    }
+}
+
+async fn handle_mongo_update_documents_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
+    let req: MongoUpdateDocumentsRequest = match serde_json::from_str(body) {
+        Ok(r) => r,
+        Err(_) => {
+            respond_error(stream, "400 Bad Request", "Invalid JSON").await;
+            return;
+        }
+    };
+    let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await
+    else {
+        return;
+    };
+    match dbx_core::mongo_ops::mongo_update_documents_core(
+        state,
+        &pool_key,
+        &database,
+        &req.collection,
+        &req.filter_json,
+        &req.update_json,
+        req.many,
+    )
+    .await
+    {
+        Ok(modified) => respond_json(stream, &serde_json::json!({ "affected_rows": modified })).await,
+        Err(e) => respond_error(stream, "500 Internal Server Error", &e).await,
+    }
+}
+
+async fn handle_mongo_delete_documents_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
+    let req: MongoDeleteDocumentsRequest = match serde_json::from_str(body) {
+        Ok(r) => r,
+        Err(_) => {
+            respond_error(stream, "400 Bad Request", "Invalid JSON").await;
+            return;
+        }
+    };
+    let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await
+    else {
+        return;
+    };
+    match dbx_core::mongo_ops::mongo_delete_documents_core(
+        state,
+        &pool_key,
+        &database,
+        &req.collection,
+        &req.filter_json,
+        req.many,
+    )
+    .await
+    {
+        Ok(deleted) => respond_json(stream, &serde_json::json!({ "affected_rows": deleted })).await,
+        Err(e) => respond_error(stream, "500 Internal Server Error", &e).await,
+    }
+}
+
 async fn handle_execute_query_data(state: &Arc<AppState>, body: &str, stream: &mut tokio::net::TcpStream) {
     let req: ExecuteQueryRequest = match serde_json::from_str(body) {
         Ok(r) => r,

src-tauri/src/commands/mongo_cmd.rs

@@ -45,6 +45,18 @@ pub async fn mongo_find_documents(
     .await
 }
 
+#[tauri::command]
+pub async fn mongo_aggregate_documents(
+    state: State<'_, Arc<AppState>>,
+    connection_id: String,
+    database: String,
+    collection: String,
+    pipeline_json: String,
+) -> Result<MongoDocumentResult, String> {
+    dbx_core::mongo_ops::mongo_aggregate_documents_core(&state, &connection_id, &database, &collection, &pipeline_json)
+        .await
+}
+
 #[tauri::command]
 pub async fn mongo_insert_document(
     state: State<'_, Arc<AppState>>,

src-tauri/src/lib.rs

@@ -370,6 +370,7 @@ pub fn run() {
             commands::mongo_cmd::mongo_list_databases,
             commands::mongo_cmd::mongo_list_collections,
             commands::mongo_cmd::mongo_find_documents,
+            commands::mongo_cmd::mongo_aggregate_documents,
             commands::mongo_cmd::mongo_insert_document,
             commands::mongo_cmd::mongo_update_document,
             commands::mongo_cmd::mongo_delete_document,