From 5c95f2c00af410746feb26a893d5d6bb352a34f8 Mon Sep 17 00:00:00 2001 From: "C.J.S" <37900422+jasonchen8068@users.noreply.github.com> Date: Thu, 10 Sep 2026 18:04:16 +0800 Subject: [PATCH] feat(mcp): support read-only Kafka message browsing --- apps/desktop/src/i18n/locales/az.ts | 1 + apps/desktop/src/i18n/locales/en.ts | 1 + apps/desktop/src/i18n/locales/es.ts | 1 + apps/desktop/src/i18n/locales/it.ts | 1 + apps/desktop/src/i18n/locales/ja.ts | 1 + apps/desktop/src/i18n/locales/ko.ts | 1 + apps/desktop/src/i18n/locales/pt-BR.ts | 1 + apps/desktop/src/i18n/locales/tr.ts | 1 + apps/desktop/src/i18n/locales/zh-CN.ts | 1 + apps/desktop/src/i18n/locales/zh-TW.ts | 1 + .../__tests__/mcp/mcpPolicySelection.spec.ts | 7 + .../desktop/src/lib/mcp/mcpPolicySelection.ts | 1 + crates/dbx-mcp/src/backend.rs | 120 ++++++++ crates/dbx-mcp/src/server.rs | 291 +++++++++++++++++- crates/dbx-mcp/tests/protocol.rs | 70 ++++- docs/content/docs/mcp.cn.mdx | 25 +- docs/content/docs/mcp.mdx | 25 +- docs/content/docs/mcp.tr.mdx | 3 +- packages/mcp-server/README.md | 6 + 19 files changed, 551 insertions(+), 7 deletions(-) diff --git a/apps/desktop/src/i18n/locales/az.ts b/apps/desktop/src/i18n/locales/az.ts index cd57bac214..a679ac491e 100644 --- a/apps/desktop/src/i18n/locales/az.ts +++ b/apps/desktop/src/i18n/locales/az.ts @@ -7576,6 +7576,7 @@ export default withEnglishFallback({ mcpToolOpenSession: "Sorğu sessiyasını aç", mcpToolCloseSession: "Sorğu sessiyasını bağla", mcpToolExecuteRedisCommand: "Redis əmrini icra et", + mcpToolPeekMessages: "Kafka mesajlarını oxu", mcpToolSendMessage: "Mesaj növbəsinə mesaj göndər", mcpToolAddConnection: "Əlaqə əlavə et", mcpToolDuplicateConnection: "Əlaqənin surətini yarat", diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index 5ab25f75c7..93378ee00c 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -7573,6 +7573,7 @@ export default { mcpToolOpenSession: "Open query session", mcpToolCloseSession: "Close query session", mcpToolExecuteRedisCommand: "Execute Redis command", + mcpToolPeekMessages: "Read Kafka messages", mcpToolSendMessage: "Send message queue message", mcpToolAddConnection: "Add connection", mcpToolDuplicateConnection: "Duplicate connection", diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts index 5f42f8fc5d..5853130973 100644 --- a/apps/desktop/src/i18n/locales/es.ts +++ b/apps/desktop/src/i18n/locales/es.ts @@ -7088,6 +7088,7 @@ export default withEnglishFallback({ mcpToolOpenSession: "Abrir sesión de consulta", mcpToolCloseSession: "Cerrar sesión de consulta", mcpToolExecuteRedisCommand: "Ejecutar comando Redis", + mcpToolPeekMessages: "Leer mensajes de Kafka", mcpToolSendMessage: "Enviar mensaje de cola de mensajes", mcpToolAddConnection: "Añadir conexión", mcpToolDuplicateConnection: "Duplicar conexión", diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts index 9014837c88..f7e900e2be 100644 --- a/apps/desktop/src/i18n/locales/it.ts +++ b/apps/desktop/src/i18n/locales/it.ts @@ -7088,6 +7088,7 @@ export default withEnglishFallback({ mcpToolOpenSession: "Apri sessione di query", mcpToolCloseSession: "Chiudi sessione di query", mcpToolExecuteRedisCommand: "Esegui comando Redis", + mcpToolPeekMessages: "Leggi messaggi Kafka", mcpToolSendMessage: "Invia messaggio in coda", mcpToolAddConnection: "Aggiungi connessione", mcpToolDuplicateConnection: "Duplica connessione", diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts index d7f4b60b7b..7af37e4495 100644 --- a/apps/desktop/src/i18n/locales/ja.ts +++ b/apps/desktop/src/i18n/locales/ja.ts @@ -7095,6 +7095,7 @@ export default withEnglishFallback({ mcpToolOpenSession: "クエリセッションを開く", mcpToolCloseSession: "クエリセッションを閉じる", mcpToolExecuteRedisCommand: "Redis コマンドを実行", + mcpToolPeekMessages: "Kafka メッセージを読み取る", mcpToolSendMessage: "メッセージキューのメッセージを送信", mcpToolAddConnection: "接続を追加", mcpToolDuplicateConnection: "接続を複製", diff --git a/apps/desktop/src/i18n/locales/ko.ts b/apps/desktop/src/i18n/locales/ko.ts index c2ed1919d8..802985ae66 100644 --- a/apps/desktop/src/i18n/locales/ko.ts +++ b/apps/desktop/src/i18n/locales/ko.ts @@ -6849,6 +6849,7 @@ export default withEnglishFallback({ mcpToolOpenSession: "쿼리 세션 열기", mcpToolCloseSession: "쿼리 세션 닫기", mcpToolExecuteRedisCommand: "Redis 명령 실행", + mcpToolPeekMessages: "Kafka 메시지 읽기", mcpToolSendMessage: "메시지 큐 메시지 전송", mcpToolAddConnection: "연결 추가", mcpToolDuplicateConnection: "연결 복제", diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts index 58e10e0729..bba554a6ad 100644 --- a/apps/desktop/src/i18n/locales/pt-BR.ts +++ b/apps/desktop/src/i18n/locales/pt-BR.ts @@ -7090,6 +7090,7 @@ export default withEnglishFallback({ mcpToolOpenSession: "Abrir sessão de consulta", mcpToolCloseSession: "Fechar sessão de consulta", mcpToolExecuteRedisCommand: "Executar comando Redis", + mcpToolPeekMessages: "Ler mensagens Kafka", mcpToolSendMessage: "Enviar mensagem da fila", mcpToolAddConnection: "Adicionar conexão", mcpToolDuplicateConnection: "Duplicar conexão", diff --git a/apps/desktop/src/i18n/locales/tr.ts b/apps/desktop/src/i18n/locales/tr.ts index 35c807732f..c96dd3fa4a 100644 --- a/apps/desktop/src/i18n/locales/tr.ts +++ b/apps/desktop/src/i18n/locales/tr.ts @@ -7456,6 +7456,7 @@ export default withEnglishFallback({ mcpToolOpenSession: "Sorgu oturumu aç", mcpToolCloseSession: "Sorgu oturumunu kapat", mcpToolExecuteRedisCommand: "Redis komutunu çalıştır", + mcpToolPeekMessages: "Kafka mesajlarını oku", mcpToolSendMessage: "Mesaj kuyruğu mesajı gönder", mcpToolAddConnection: "Bağlantı ekle", mcpToolDuplicateConnection: "Bağlantıyı çoğalt", diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index 0ea9bd98d7..94a969e06b 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/apps/desktop/src/i18n/locales/zh-CN.ts @@ -7558,6 +7558,7 @@ export default withEnglishFallback({ mcpToolOpenSession: "打开查询会话", mcpToolCloseSession: "关闭查询会话", mcpToolExecuteRedisCommand: "执行 Redis 命令", + mcpToolPeekMessages: "读取 Kafka 消息", mcpToolSendMessage: "发送消息队列消息", mcpToolAddConnection: "新增连接", mcpToolDuplicateConnection: "复制连接", diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index 790a132ef9..2d98eeac8e 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/apps/desktop/src/i18n/locales/zh-TW.ts @@ -6449,6 +6449,7 @@ export default withEnglishFallback({ mcpToolOpenSession: "開啟查詢工作階段", mcpToolCloseSession: "關閉查詢工作階段", mcpToolExecuteRedisCommand: "執行 Redis 命令", + mcpToolPeekMessages: "讀取 Kafka 訊息", mcpToolSendMessage: "傳送訊息佇列訊息", mcpToolAddConnection: "新增連線", mcpToolDuplicateConnection: "複製連線", diff --git a/apps/desktop/src/lib/__tests__/mcp/mcpPolicySelection.spec.ts b/apps/desktop/src/lib/__tests__/mcp/mcpPolicySelection.spec.ts index ad0769dd18..92ef184626 100644 --- a/apps/desktop/src/lib/__tests__/mcp/mcpPolicySelection.spec.ts +++ b/apps/desktop/src/lib/__tests__/mcp/mcpPolicySelection.spec.ts @@ -97,6 +97,13 @@ describe("MCP tool permission selection", () => { expect(next).not.toContain("dbx_send_message"); }); + it("keeps Kafka reading independent from message sending", () => { + const next = toggleMcpAllowedToolName(null, "dbx_send_message", false); + expect(next).toContain("dbx_peek_messages"); + expect(toggleMcpAllowedToolName(next, "dbx_peek_messages", false)).not.toContain("dbx_peek_messages"); + expect(toggleMcpAllowedToolName([], "dbx_peek_messages", true)).toEqual(["dbx_peek_messages"]); + }); + it("lets batch execution be enabled and disabled independently", () => { expect(toggleMcpAllowedToolName(["dbx_execute_query"], "dbx_execute_batch", true)).toEqual(["dbx_execute_query", "dbx_execute_batch"]); expect(toggleMcpAllowedToolName(["dbx_execute_query", "dbx_execute_batch"], "dbx_execute_batch", false)).toEqual(["dbx_execute_query"]); diff --git a/apps/desktop/src/lib/mcp/mcpPolicySelection.ts b/apps/desktop/src/lib/mcp/mcpPolicySelection.ts index c99590e74f..b2cb830f20 100644 --- a/apps/desktop/src/lib/mcp/mcpPolicySelection.ts +++ b/apps/desktop/src/lib/mcp/mcpPolicySelection.ts @@ -34,6 +34,7 @@ export const MCP_TOOL_OPTIONS = [ { name: "dbx_open_session", labelKey: "settings.mcpToolOpenSession" }, { name: "dbx_close_session", labelKey: "settings.mcpToolCloseSession" }, { name: "dbx_execute_redis_command", labelKey: "settings.mcpToolExecuteRedisCommand" }, + { name: "dbx_peek_messages", labelKey: "settings.mcpToolPeekMessages" }, { name: "dbx_send_message", labelKey: "settings.mcpToolSendMessage" }, { name: "dbx_add_connection", labelKey: "settings.mcpToolAddConnection" }, { name: "dbx_duplicate_connection", labelKey: "settings.mcpToolDuplicateConnection" }, diff --git a/crates/dbx-mcp/src/backend.rs b/crates/dbx-mcp/src/backend.rs index 6c040f0196..ef1eba7499 100644 --- a/crates/dbx-mcp/src/backend.rs +++ b/crates/dbx-mcp/src/backend.rs @@ -215,6 +215,17 @@ pub trait DbxBackend: Send + Sync { let _ = (connection, request); Err("Message queue sending is not supported by this backend.".to_string()) } + #[cfg(feature = "mq-admin")] + async fn peek_messages( + &self, + connection: &ConnectionConfig, + topic: dbx_core::mq::TopicRef, + count: u32, + options: dbx_core::mq::PeekMessagesOptions, + ) -> Result { + let _ = (connection, topic, count, options); + Err("Message queue reading is not supported by this backend.".to_string()) + } async fn execute_query( &self, connection: &ConnectionConfig, @@ -766,6 +777,25 @@ impl DbxBackend for LocalBackend { dbx_core::mq::service::mq_send_message_core(&self.state, &connection.id, request).await } + #[cfg(feature = "mq-admin")] + async fn peek_messages( + &self, + connection: &ConnectionConfig, + topic: dbx_core::mq::TopicRef, + count: u32, + options: dbx_core::mq::PeekMessagesOptions, + ) -> Result { + dbx_core::mq::service::mq_peek_messages_core( + &self.state, + &connection.id, + topic, + "__dbx_kafka_viewer__".into(), + count, + Some(options), + ) + .await + } + async fn execute_query( &self, connection: &ConnectionConfig, @@ -1174,6 +1204,21 @@ impl DbxBackend for WebBackend { .map_err(|error| format!("Invalid message send response: {error}")) } + #[cfg(feature = "mq-admin")] + async fn peek_messages( + &self, + connection: &ConnectionConfig, + topic: dbx_core::mq::TopicRef, + count: u32, + options: dbx_core::mq::PeekMessagesOptions, + ) -> Result { + self.request( + reqwest::Method::POST, + "/api/mq/subscriptions/peek-messages", + Some(json!({ "connectionId": connection.id, "topic": topic, "sub": "__dbx_kafka_viewer__", "count": count, "options": options })), + ).await?.json().await.map_err(|error| format!("Invalid message peek response: {error}")) + } + async fn execute_query( &self, connection: &ConnectionConfig, @@ -2538,6 +2583,81 @@ mod tests { assert_eq!(second_request["timeoutSecs"], 300); } + #[cfg(feature = "mq-admin")] + #[tokio::test] + async fn web_peek_messages_forwards_kafka_options_and_preserves_partial_results() { + use dbx_core::mq::{PeekMessagesOptions, PeekStartPosition, TopicRef}; + use std::io::BufRead; + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + stream.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + let mut reader = std::io::BufReader::new(&mut stream); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + assert_eq!(line.trim(), "POST /api/mq/subscriptions/peek-messages HTTP/1.1"); + let mut content_length = 0; + loop { + line.clear(); + assert!(reader.read_line(&mut line).unwrap() > 0); + if line == "\r\n" { + break; + } + if let Some((name, value)) = line.split_once(':') { + if name.eq_ignore_ascii_case("content-length") { + content_length = value.trim().parse::().unwrap(); + } + } + } + let mut body = vec![0; content_length]; + reader.read_exact(&mut body).unwrap(); + let body: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body["connectionId"], "kafka-peek"); + assert_eq!(body["topic"]["topic"], "events"); + assert_eq!(body["sub"], "__dbx_kafka_viewer__"); + assert_eq!(body["count"], 7); + assert_eq!(body["options"], json!({"startPosition":"offset", "partition":2, "offset":17})); + let response = r#"{"messages":[{"position":1,"messageId":"2:17","payloadBase64":"/w==","headers":{"type":"binary"}}],"incomplete":true}"#; + write!(stream, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", response.len(), response).unwrap(); + }); + let backend = + WebBackend::new_with_config(format!("http://{address}"), String::new(), None, None, None, false, None) + .unwrap(); + backend.auth.lock().await.checked = true; + let connection = new_connection_config( + "kafka-peek".into(), + "Kafka".into(), + DatabaseType::MessageQueue, + "localhost".into(), + 9092, + String::new(), + String::new(), + None, + false, + None, + ) + .unwrap(); + let result = backend + .peek_messages( + &connection, + TopicRef { topic: "events".into(), ..Default::default() }, + 7, + PeekMessagesOptions { + start_position: Some(PeekStartPosition::Offset), + partition: Some(2), + offset: Some(17), + }, + ) + .await + .unwrap(); + assert!(result.incomplete); + assert_eq!(result.messages[0].payload_base64, "/w=="); + assert_eq!(result.messages[0].message_id.as_deref(), Some("2:17")); + assert_eq!(result.messages[0].headers.get("type").map(String::as_str), Some("binary")); + server.join().unwrap(); + } + #[tokio::test] async fn web_execute_batch_hits_execute_multi_and_decodes_statement_results() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); diff --git a/crates/dbx-mcp/src/server.rs b/crates/dbx-mcp/src/server.rs index cb3d5a1e91..f684aeeeac 100644 --- a/crates/dbx-mcp/src/server.rs +++ b/crates/dbx-mcp/src/server.rs @@ -197,6 +197,32 @@ pub struct ExecuteRedisCommandRequest { pub command: String, } +#[derive(Debug, Default, serde::Serialize, Deserialize, schemars::JsonSchema)] +#[serde(rename_all = "lowercase")] +pub enum PeekStartPosition { + #[default] + Latest, + Earliest, + Offset, +} + +#[derive(Debug, Deserialize, schemars::JsonSchema)] +pub struct PeekMessagesRequest { + #[serde(flatten)] + pub selector: ConnectionSelector, + #[schemars(description = "Kafka topic name")] + pub topic: String, + #[schemars(description = "Number of messages, 1 to 100 (default 20)", extend("type" = "integer"))] + pub count: Option, + #[serde(default)] + #[schemars(description = "Read latest messages (default), earliest messages, or from an offset")] + pub start_position: PeekStartPosition, + #[schemars(description = "Non-negative partition; omit to read across partitions", extend("type" = "integer"))] + pub partition: Option, + #[schemars(description = "Non-negative offset, required only for start_position=offset", extend("type" = "integer"))] + pub offset: Option, +} + #[derive(Debug, Deserialize, schemars::JsonSchema)] pub struct SendMessageRequest { #[serde(flatten)] @@ -382,7 +408,10 @@ impl DbxMcpServer { tool_router.disable_route("dbx_execute_and_show"); } #[cfg(not(feature = "mq-admin"))] - tool_router.disable_route("dbx_send_message"); + { + tool_router.disable_route("dbx_send_message"); + tool_router.disable_route("dbx_peek_messages"); + } Self { backend, scope, sessions: McpSessionStore::new(), tool_router } } @@ -1059,6 +1088,68 @@ impl DbxMcpServer { } } + #[tool( + name = "dbx_peek_messages", + description = "Read up to 100 Kafka messages without committing consumer offsets. Defaults to the latest 20 messages. Returns JSON with base64 payloads, UTF-8 previews, metadata, incomplete (partial broker read), and outputTruncated (256 KiB message output budget). Only Kafka is supported." + )] + async fn peek_messages(&self, Parameters(request): Parameters) -> CallToolResult { + if let Err(error) = self.ensure_tool_allowed("dbx_peek_messages").await { + return error; + } + #[cfg(not(feature = "mq-admin"))] + { + let _ = request; + tool_error("MQ_UNSUPPORTED", "Message queue support is not compiled into this DBX MCP build.") + } + #[cfg(feature = "mq-admin")] + { + use dbx_core::mq::{ + config::MqAdminConfig, MqSystemKind, PeekMessagesOptions, PeekStartPosition as Start, TopicRef, + }; + let resolved = match self.resolve_connection(&request.selector).await { + Ok(resolved) => resolved, + Err(error) => return error, + }; + match MqAdminConfig::from_connection(&resolved.connection) { + Ok(config) if config.system_kind == MqSystemKind::Kafka => {} + _ => { + return tool_error("MQ_UNSUPPORTED", "Message reading through MCP supports Kafka connections only.") + } + } + let count = request.count.unwrap_or(20); + let start_position = match request.start_position { + PeekStartPosition::Latest => Start::Latest, + PeekStartPosition::Earliest => Start::Earliest, + PeekStartPosition::Offset => Start::Offset, + }; + if request.topic.trim().is_empty() { + return tool_error("MESSAGE_TOPIC_REQUIRED", "Message topic must not be empty."); + } + if !(1..=100).contains(&count) + || request.partition.is_some_and(|value| value < 0) + || request.offset.is_some_and(|value| value < 0) + || (start_position == Start::Offset) != request.offset.is_some() + { + return tool_error("INVALID_PEEK_OPTIONS", "count must be 1..100; partition and offset must be non-negative; offset is required only for start_position=offset."); + } + let topic = TopicRef { + tenant: "_kafka".into(), + namespace: "default".into(), + topic: request.topic.trim().into(), + ..Default::default() + }; + let options = PeekMessagesOptions { + start_position: Some(start_position), + partition: request.partition, + offset: request.offset, + }; + match self.backend.peek_messages(&resolved.connection, topic, count, options).await { + Ok(result) => text(format_peek_messages_result(result, count as usize)), + Err(error) => backend_tool_error("MESSAGE_PEEK_ERROR", error), + } + } + } + #[tool( name = "dbx_send_message", description = "Send a base64-encoded message to a Kafka, RocketMQ, or RabbitMQ topic/queue" @@ -2127,6 +2218,23 @@ fn non_empty_env(name: &str) -> Option { std::env::var(name).ok().map(|value| value.trim().to_string()).filter(|value| !value.is_empty()) } +#[cfg(feature = "mq-admin")] +fn format_peek_messages_result(result: dbx_core::mq::PeekMessagesResult, count: usize) -> String { + let mut messages = Vec::new(); + let mut bytes = 0; + let mut truncated = result.messages.len() > count; + for message in result.messages.into_iter().take(count) { + let value = serde_json::to_value(message).expect("message is JSON serializable"); + bytes += value.to_string().len(); + if bytes > 256 * 1024 { + truncated = true; + break; + } + messages.push(value); + } + json!({ "messages": messages, "incomplete": result.incomplete, "outputTruncated": truncated }).to_string() +} + #[cfg(feature = "mq-admin")] fn format_send_message_result( system_kind: &dbx_core::mq::MqSystemKind, @@ -2391,6 +2499,28 @@ mod tests { } } + #[cfg(feature = "mq-admin")] + async fn peek_messages( + &self, + connection: &ConnectionConfig, + topic: dbx_core::mq::TopicRef, + count: u32, + options: dbx_core::mq::PeekMessagesOptions, + ) -> Result { + self.recorded_arguments.lock().unwrap().push(( + "peek_messages".into(), + json!({"connectionId": connection.id, "topic": topic, "count": count, "options": options}), + )); + Ok(dbx_core::mq::PeekMessagesResult { + messages: vec![dbx_core::mq::PeekedMessage { + payload_base64: "aGk=".into(), + payload_text: Some("hi".into()), + ..Default::default() + }], + incomplete: true, + }) + } + async fn close_client_session( &self, _connection_id: &str, @@ -2480,6 +2610,155 @@ mod tests { } } + #[cfg(feature = "mq-admin")] + fn kafka_connection() -> ConnectionConfig { + let mut conn = connection("kafka", "Kafka", "mq", ""); + conn.read_only = true; + conn.is_production = true; + conn.external_config = Some( + json!({"systemKind":"kafka", "adminUrl":"", "auth":{"kind":"none"}, "extra":{"bootstrapServers":"localhost:9092"}}), + ); + conn + } + + #[cfg(feature = "mq-admin")] + #[tokio::test] + async fn peek_messages_reads_with_read_only_policy_and_forwards_positions() { + let backend = Arc::new(FakeBackend { connections: vec![kafka_connection()], ..Default::default() }); + let server = DbxMcpServer::with_runtime_options(backend.clone(), McpScope::default(), false); + for (extra, expected) in [ + (json!({}), json!({"startPosition":"latest"})), + (json!({"start_position":"earliest", "partition":0}), json!({"startPosition":"earliest", "partition":0})), + ( + json!({"start_position":"offset", "partition":2, "offset":17}), + json!({"startPosition":"offset", "partition":2, "offset":17}), + ), + (json!({"start_position":"offset", "offset":0}), json!({"startPosition":"offset", "offset":0})), + ] { + let mut request = json!({"connection_id":"kafka", "topic":" events "}); + request.as_object_mut().unwrap().extend(extra.as_object().unwrap().clone()); + let result = server.peek_messages(Parameters(serde_json::from_value(request).unwrap())).await; + assert_ne!(result.is_error, Some(true), "{}", result_text(&result)); + let response: serde_json::Value = serde_json::from_str(result_text(&result)).unwrap(); + assert_eq!(response["messages"][0]["payloadBase64"], "aGk="); + assert_eq!(response["incomplete"], true); + assert_eq!(response["outputTruncated"], false); + let calls = backend.recorded_arguments.lock().unwrap(); + let args = &calls.last().unwrap().1; + assert_eq!(args["options"], expected); + assert_eq!(args["count"], 20); + assert_eq!(args["topic"]["topic"], "events"); + } + } + + #[cfg(feature = "mq-admin")] + #[tokio::test] + async fn peek_messages_rejects_invalid_options_before_backend_call() { + let backend = Arc::new(FakeBackend { connections: vec![kafka_connection()], ..Default::default() }); + let server = DbxMcpServer::with_runtime_options(backend.clone(), McpScope::default(), false); + for extra in [ + json!({"count":0}), + json!({"count":101}), + json!({"partition":-1}), + json!({"start_position":"offset"}), + json!({"offset":1}), + json!({"start_position":"earliest", "offset":0}), + json!({"start_position":"offset", "offset":-1}), + json!({"topic":" "}), + ] { + let mut request = json!({"connection_id":"kafka", "topic":"events"}); + request.as_object_mut().unwrap().extend(extra.as_object().unwrap().clone()); + let result = server.peek_messages(Parameters(serde_json::from_value(request).unwrap())).await; + assert_eq!(result.is_error, Some(true), "{}", result_text(&result)); + } + assert!(backend.recorded_arguments.lock().unwrap().is_empty()); + } + + #[cfg(feature = "mq-admin")] + #[tokio::test] + async fn peek_messages_enforces_tool_connection_scope_and_kafka_only() { + for (policy, conn, expected) in [ + ( + McpGlobalPolicy { allowed_tool_names: Some(vec![]), ..Default::default() }, + kafka_connection(), + "TOOL_OUT_OF_SCOPE", + ), + ( + McpGlobalPolicy { allowed_connection_ids: Some(vec![]), ..Default::default() }, + kafka_connection(), + "CONNECTION_OUT_OF_SCOPE", + ), + (McpGlobalPolicy::default(), connection("kafka", "redis", "redis", "0"), "MQ_UNSUPPORTED"), + ] { + let backend = Arc::new(FakeBackend { policy, connections: vec![conn], ..Default::default() }); + let server = DbxMcpServer::with_runtime_options(backend.clone(), McpScope::default(), false); + let result = server + .peek_messages(Parameters( + serde_json::from_value(json!({"connection_id":"kafka", "topic":"events"})).unwrap(), + )) + .await; + assert_eq!(result.is_error, Some(true)); + assert!(result_text(&result).contains(expected), "{}", result_text(&result)); + assert!(backend.recorded_arguments.lock().unwrap().is_empty()); + } + } + + #[cfg(feature = "mq-admin")] + #[tokio::test] + async fn peek_messages_respects_session_scope_and_rejects_other_queues() { + let mut rabbit = kafka_connection(); + rabbit.external_config.as_mut().unwrap()["systemKind"] = json!("rabbitmq"); + for (conn, scope, expected) in [ + ( + kafka_connection(), + McpScope { connection_ids: vec!["other".into()], ..Default::default() }, + "CONNECTION_OUT_OF_SCOPE", + ), + (rabbit, McpScope::default(), "MQ_UNSUPPORTED"), + ] { + let backend = Arc::new(FakeBackend { connections: vec![conn], ..Default::default() }); + let server = DbxMcpServer::with_runtime_options(backend.clone(), scope, false); + let result = server + .peek_messages(Parameters( + serde_json::from_value(json!({"connection_id":"kafka", "topic":"events"})).unwrap(), + )) + .await; + assert_eq!(result.is_error, Some(true)); + assert!(result_text(&result).contains(expected), "{}", result_text(&result)); + assert!(backend.recorded_arguments.lock().unwrap().is_empty()); + } + } + + #[cfg(feature = "mq-admin")] + #[test] + fn peek_messages_output_preserves_metadata_and_bounds_whole_messages() { + use dbx_core::mq::{PeekMessagesResult, PeekedMessage}; + let message = PeekedMessage { + payload_base64: "/w==".into(), + message_id: Some("2:17".into()), + headers: [("kind".into(), "event".into())].into(), + ..Default::default() + }; + let output = format_peek_messages_result(PeekMessagesResult::complete(vec![message.clone(), message]), 1); + let result: serde_json::Value = serde_json::from_str(&output).unwrap(); + assert_eq!(result["messages"].as_array().unwrap().len(), 1); + assert_eq!(result["messages"][0]["messageId"], "2:17"); + assert_eq!(result["messages"][0]["headers"]["kind"], "event"); + assert_eq!(result["outputTruncated"], true); + assert_eq!(result["incomplete"], false); + let output = format_peek_messages_result( + PeekMessagesResult::complete(vec![PeekedMessage { + payload_base64: "a".repeat(300_000), + ..Default::default() + }]), + 20, + ); + let result: serde_json::Value = serde_json::from_str(&output).unwrap(); + assert!(result["messages"].as_array().unwrap().is_empty()); + assert_eq!(result["outputTruncated"], true); + assert!(output.len() < 256 * 1024); + } + #[test] fn connection_table_escapes_markdown_cells() { let output = format_connections(&[ConnectionSummary { @@ -2502,9 +2781,13 @@ mod tests { let tools = server.tool_router.list_all(); let names = tools.iter().map(|tool| tool.name.as_ref()).collect::>(); #[cfg(feature = "mq-admin")] - assert_eq!(tools.len(), 18); + assert_eq!(tools.len(), 19); #[cfg(not(feature = "mq-admin"))] assert_eq!(tools.len(), 17); + #[cfg(feature = "mq-admin")] + assert!(names.contains(&"dbx_peek_messages")); + #[cfg(not(feature = "mq-admin"))] + assert!(!names.contains(&"dbx_peek_messages")); assert!(names.contains(&"dbx_list_connections")); assert!(names.contains(&"dbx_list_databases")); assert!(names.contains(&"dbx_list_tables")); @@ -2638,6 +2921,8 @@ mod tests { checks .push(("dbx_send_message", &["key", "payload_text", "partition", "exchange", "routing_key", "namespace"])); + #[cfg(feature = "mq-admin")] + checks.push(("dbx_peek_messages", &["count", "partition", "offset"])); for (tool_name, fields) in checks { let tool = tools.iter().find(|tool| tool.name == *tool_name).expect("tool should be registered"); let properties = tool @@ -2748,7 +3033,7 @@ mod tests { ); let names = server.tool_router.list_all().into_iter().map(|tool| tool.name).collect::>(); #[cfg(feature = "mq-admin")] - assert_eq!(names.len(), 13); + assert_eq!(names.len(), 14); #[cfg(not(feature = "mq-admin"))] assert_eq!(names.len(), 12); assert!(!names.iter().any(|name| name == "dbx_add_connection")); diff --git a/crates/dbx-mcp/tests/protocol.rs b/crates/dbx-mcp/tests/protocol.rs index 3ca7b550e0..23320a8dbb 100644 --- a/crates/dbx-mcp/tests/protocol.rs +++ b/crates/dbx-mcp/tests/protocol.rs @@ -121,6 +121,25 @@ struct CapturingBackend { #[async_trait] impl DbxBackend for CapturingBackend { + #[cfg(feature = "mq-admin")] + async fn peek_messages( + &self, + connection: &ConnectionConfig, + topic: dbx_core::mq::TopicRef, + count: u32, + options: dbx_core::mq::PeekMessagesOptions, + ) -> Result { + self.calls + .lock() + .unwrap() + .push(json!({"connection":connection.id, "topic":topic.topic, "count":count, "options":options})); + Ok(dbx_core::mq::PeekMessagesResult::complete(vec![dbx_core::mq::PeekedMessage { + payload_base64: "aGk=".into(), + payload_text: Some("hi".into()), + ..Default::default() + }])) + } + async fn load_mcp_global_policy(&self) -> Result { Ok(self.policy.clone()) } @@ -285,9 +304,13 @@ async fn initializes_lists_tools_and_calls_a_tool() { let tools = client.peer().list_tools(None).await.expect("list tools"); let names = tools.tools.iter().map(|tool| tool.name.as_ref()).collect::>(); #[cfg(feature = "mq-admin")] - assert_eq!(names.len(), 18); + assert_eq!(names.len(), 19); #[cfg(not(feature = "mq-admin"))] assert_eq!(names.len(), 17); + #[cfg(feature = "mq-admin")] + assert!(names.contains(&"dbx_peek_messages")); + #[cfg(not(feature = "mq-admin"))] + assert!(!names.contains(&"dbx_peek_messages")); assert!(names.contains(&"dbx_list_connections")); assert!(names.contains(&"dbx_list_databases")); assert!(names.contains(&"dbx_duplicate_connection")); @@ -309,6 +332,51 @@ async fn initializes_lists_tools_and_calls_a_tool() { server_task.abort(); } +#[cfg(feature = "mq-admin")] +#[tokio::test] +async fn kafka_peek_round_trips_over_mcp_in_local_and_web_modes() { + let mut connection = test_connection("kafka", "Kafka"); + connection.db_type = dbx_core::models::connection::DatabaseType::MessageQueue; + connection.read_only = true; + connection.is_production = true; + connection.external_config = Some(json!({"systemKind":"kafka", "adminUrl":"", "auth":{"kind":"none"}})); + for web_mode in [false, true] { + let backend = Arc::new(CapturingBackend { + policy: McpGlobalPolicy::default(), + connections: vec![connection.clone()], + calls: Mutex::new(vec![]), + }); + let (server_transport, client_transport) = tokio::io::duplex(16 * 1024); + let server = DbxMcpServer::with_runtime_options(backend.clone(), McpScope::default(), web_mode); + let server_task = tokio::spawn(async move { server.serve(server_transport).await }); + let client = ().serve(client_transport).await.unwrap(); + let result = client.peer().call_tool(CallToolRequestParams::new("dbx_peek_messages").with_arguments(json!({"connection_name":"Kafka", "topic":"events", "count":100, "start_position":"offset", "partition":0, "offset":120}).as_object().unwrap().clone())).await.unwrap(); + assert_ne!(result.is_error, Some(true), "{result:?}"); + let body: Value = serde_json::from_str(&result.content[0].as_text().unwrap().text).unwrap(); + assert_eq!(body["messages"][0]["payloadText"], "hi"); + assert_eq!(body["incomplete"], false); + assert_eq!( + backend.calls.lock().unwrap()[0], + json!({"connection":"kafka", "topic":"events", "count":100, "options":{"startPosition":"offset", "partition":0, "offset":120}}) + ); + let invalid = client + .peer() + .call_tool( + CallToolRequestParams::new("dbx_peek_messages").with_arguments( + json!({"connection_id":"kafka", "topic":"events", "start_position":"invalid"}) + .as_object() + .unwrap() + .clone(), + ), + ) + .await; + assert!(invalid.is_err() || invalid.unwrap().is_error == Some(true)); + assert_eq!(backend.calls.lock().unwrap().len(), 1); + client.cancel().await.unwrap(); + server_task.abort(); + } +} + #[tokio::test] async fn remove_connection_respects_global_connection_scope() { let backend = PolicyBackend { diff --git a/docs/content/docs/mcp.cn.mdx b/docs/content/docs/mcp.cn.mdx index ea70d84991..900c259693 100644 --- a/docs/content/docs/mcp.cn.mdx +++ b/docs/content/docs/mcp.cn.mdx @@ -87,7 +87,7 @@ DeepSeek Harness 通过 Cordis 插件条目加载 MCP Server，不读取 `mcpSer ## 工具列表 -DBX MCP 当前提供 18 个工具： +DBX MCP 当前提供以下工具（具体可用工具取决于构建功能及权限设置）： | 工具 | 说明 | | --- | --- | @@ -106,12 +106,35 @@ DBX MCP 当前提供 18 个工具： | `dbx_open_session` | 为 SQL 连接打开固定后端连接的有状态查询会话 | | `dbx_close_session` | 关闭会话并释放固定连接资源 | | `dbx_execute_redis_command` | 执行 Redis 命令 | +| `dbx_peek_messages` | 读取 Kafka 消息，不提交消费位点 | | `dbx_send_message` | 向支持的消息队列 Topic 或队列发送消息 | | `dbx_open_table` | 在运行中的 DBX 桌面端打开表 | | `dbx_execute_and_show` | 执行查询并在 DBX 中展示结果 | 启用连接作用域后，修改连接和桌面 UI 工具会被隐藏。 +## 读取 Kafka 消息 + +包含 `mq-admin` 的构建在本地和 Web 模式下均提供 `dbx_peek_messages`。通过 `connection_id` 或 `connection_name` 选择已保存的 Kafka 连接，并指定 `topic`。工具遵守 MCP 连接范围及工具白名单，允许读取只读连接和生产连接，不提交消费位点。 + +```json +{ + "connection_name": "Kafka", + "topic": "events", + "count": 20, + "start_position": "offset", + "partition": 0, + "offset": 120 +} +``` + +- `count`：1–100，默认 20。 +- `start_position`：`latest`（默认）、`earliest` 或 `offset`。仅 `offset` 模式必须且允许传入非负 `offset`。 +- `partition`：可选的非负分区号。不传时跨分区读取；offset 模式下从各分区的该位点开始读取。 +- JSON 结果包含 `messages`，保留 base64 消息体、可无损解码时的 UTF-8 预览、消息 ID 和元数据。`incomplete` 表示读取达到超时或扫描上限；`outputTruncated` 表示为满足 256 KiB 消息输出预算而省略了整条消息。如果首条消息过大，可能返回空列表并标记 `outputTruncated: true`，不会静默截断消息体。 + +该工具读取有限条消息快照，不提供持续订阅或全 Topic 文本搜索。其他消息队列类型会被拒绝。 + ## 批量 SQL 执行 `dbx_execute_batch` 在单次调用内执行包含多条语句的 SQL 脚本。默认（自动提交）模式下按语句返回结果，因此调用方可以清楚看到哪条失败、每条影响多少行、各自返回什么；使用 `use_transaction` 且脚本包含多条语句时，只返回一个合并的事务结果。 diff --git a/docs/content/docs/mcp.mdx b/docs/content/docs/mcp.mdx index ad8ec8213d..5411b1b89a 100644 --- a/docs/content/docs/mcp.mdx +++ b/docs/content/docs/mcp.mdx @@ -87,7 +87,7 @@ Run `dsh web --dump-config` to verify the composed configuration, then restart D ## Tools -DBX MCP currently provides 18 tools: +DBX MCP provides the following tools (availability depends on build features and access settings): | Tool | Description | | --- | --- | @@ -106,12 +106,35 @@ DBX MCP currently provides 18 tools: | `dbx_open_session` | Open a stateful SQL query session pinned to one backend connection | | `dbx_close_session` | Close a session and release its pinned connection resources | | `dbx_execute_redis_command` | Execute a Redis command | +| `dbx_peek_messages` | Read Kafka messages without committing consumer offsets | | `dbx_send_message` | Send a message to a supported message queue topic or queue | | `dbx_open_table` | Open a table in the running DBX desktop app | | `dbx_execute_and_show` | Execute a query and display the result in DBX | Connection-scoped sessions hide connection-mutating and desktop UI tools. +## Read Kafka Messages + +`dbx_peek_messages` is available in builds with `mq-admin`, in both local and Web modes. Select a saved Kafka connection by `connection_id` or `connection_name` and provide `topic`. It respects MCP connection scopes and tool allowlists and works with read-only and production connections without committing consumer offsets. + +```json +{ + "connection_name": "Kafka", + "topic": "events", + "count": 20, + "start_position": "offset", + "partition": 0, + "offset": 120 +} +``` + +- `count`: 1–100, defaults to 20. +- `start_position`: `latest` (default), `earliest`, or `offset`. Supply a non-negative `offset` only with `offset` mode. +- `partition`: optional non-negative partition number. If omitted, reads across topic partitions; in offset mode, the offset applies to every partition. +- The JSON response contains `messages` with base64 payloads, lossless UTF-8 previews when available, message IDs and metadata. `incomplete` means the broker read reached a deadline or scan limit. `outputTruncated` means whole messages were omitted to fit the 256 KiB message output budget. A single oversized message may therefore yield no messages with `outputTruncated: true`; payloads are never silently clipped. + +This is a bounded snapshot, not a continuous consumer or a topic-wide text search. Other message queue types are rejected. + ## Batch SQL Execution `dbx_execute_batch` runs a SQL script containing multiple statements in a single call. In auto-commit mode (the default) it returns one result per statement, so callers see exactly which statement failed, how many rows each affected, and what each returned. With `use_transaction` and a script containing multiple statements it returns a single merged transaction outcome instead. diff --git a/docs/content/docs/mcp.tr.mdx b/docs/content/docs/mcp.tr.mdx index d643302e48..c73aa908d9 100644 --- a/docs/content/docs/mcp.tr.mdx +++ b/docs/content/docs/mcp.tr.mdx @@ -87,7 +87,7 @@ Oluşan yapılandırmayı doğrulamak için `dsh web --dump-config` çalıştır ## Araçlar -DBX MCP şu anda 17 araç sunar: +DBX MCP aşağıdaki araçları sunar: | Araç | Açıklama | | --- | --- | @@ -105,6 +105,7 @@ DBX MCP şu anda 17 araç sunar: | `dbx_open_session` | Tek bir arka uç bağlantısına sabitlenmiş durumlu bir SQL sorgu oturumu açar | | `dbx_close_session` | Bir oturumu kapatır ve sabitlenmiş bağlantı kaynaklarını serbest bırakır | | `dbx_execute_redis_command` | Bir Redis komutu çalıştırır | +| `dbx_peek_messages` | Tüketici ofsetlerini kaydetmeden Kafka mesajlarını okur | | `dbx_send_message` | Desteklenen bir mesaj kuyruğu konusuna ya da kuyruğuna mesaj gönderir | | `dbx_open_table` | Çalışan DBX masaüstü uygulamasında bir tablo açar | | `dbx_execute_and_show` | Bir sorgu çalıştırır ve sonucu DBX'te gösterir | diff --git a/packages/mcp-server/README.md b/packages/mcp-server/README.md index 9148471f6b..b28d3f3949 100644 --- a/packages/mcp-server/README.md +++ b/packages/mcp-server/README.md @@ -157,12 +157,15 @@ Ask the MCP client to: | `dbx_open_session` | Open a stateful SQL query session pinned to one backend connection | | `dbx_close_session` | Close a session and release its pinned connection resources | | `dbx_execute_redis_command` | Execute a Redis command | +| `dbx_peek_messages` | Read Kafka messages without committing consumer offsets | | `dbx_send_message` | Send a message to a supported message queue topic | | `dbx_open_table` | Open a table in the running DBX desktop application | | `dbx_execute_and_show` | Execute a query and display the result in the DBX desktop application | When connection scoping is enabled, mutating connection tools and desktop UI tools are hidden. +`dbx_peek_messages` reads a Kafka topic in local or Web mode when `mq-admin` is enabled. Pass `connection_id` or `connection_name`, `topic`, optional `count` (1–100, default 20), `start_position` (`latest` by default, `earliest`, or `offset`), and optional non-negative `partition`. A non-negative `offset` is required only in offset mode; without a partition it applies to all partitions. The JSON response preserves base64 payloads and metadata, reports broker partial reads via `incomplete`, and reports whole-message omissions under a 256 KiB output budget via `outputTruncated`. It respects connection/tool scopes and permits read-only and production reads without committing consumer offsets. It does not support other MQ types or continuous subscriptions. + `dbx_list_databases` returns only database names allowed by the selected connection's MCP database scope. `dbx_send_message` is available when message-queue support is included in the server build. ## Execution Modes @@ -531,10 +534,13 @@ MCP 配置： | `dbx_open_session` | 为 SQL 连接打开固定后端连接的有状态查询会话 | | `dbx_close_session` | 关闭会话并释放固定连接资源 | | `dbx_execute_redis_command` | 执行 Redis 命令 | +| `dbx_peek_messages` | 读取 Kafka 消息，不提交消费位点 | | `dbx_send_message` | 向支持的消息队列 Topic 发送消息 | | `dbx_open_table` | 在 DBX 桌面端打开表 | | `dbx_execute_and_show` | 执行查询并在 DBX 桌面端展示结果 | +启用 `mq-admin` 后，`dbx_peek_messages` 可在本地和 Web 模式下读取 Kafka Topic。参数为 `connection_id` 或 `connection_name`、`topic`、可选 `count`（1–100，默认 20）、`start_position`（默认 `latest`，也支持 `earliest`、`offset`）及可选的非负 `partition`。仅 offset 模式必须且允许指定非负 `offset`，未指定分区时该位点应用于所有分区。JSON 结果保留 base64 消息体和元数据，通过 `incomplete` 标记底层不完整读取，通过 `outputTruncated` 标记因 256 KiB 输出预算而省略整条消息。工具遵守连接和工具范围，允许只读与生产连接读取，不提交消费位点，不支持其他 MQ 类型或持续订阅。 + `dbx_list_databases` 只返回该连接 MCP 数据库范围内允许访问的名称。`dbx_send_message` 仅在 Server 构建时包含消息队列支持时可用。 ### 本地数据目录