diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index 0bcb64de2..2aa0f9c71 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -1783,6 +1783,20 @@ async fn agent_object_statistics_query( .await } +#[cfg(feature = "mq-admin")] +fn message_queue_topic_tables(topics: Vec) -> Vec { + topics + .into_iter() + .map(|topic| db::TableInfo { + name: topic.name, + table_type: "TOPIC".to_string(), + comment: None, + parent_schema: topic.namespace, + parent_name: None, + }) + .collect() +} + async fn list_tables_once( state: &AppState, connection_id: &str, @@ -1799,6 +1813,25 @@ async fn list_tables_once( state.get_or_create_metadata_pool_for_session(connection_id, Some(database), client_session_id).await?; let db_config = connection_config(state, connection_id).await; + #[cfg(feature = "mq-admin")] + if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::MessageQueue) { + let topics = crate::mq::service::mq_list_topics_core( + state, + connection_id, + crate::mq::NamespaceRef { tenant: database.to_string(), namespace: schema.to_string() }, + crate::mq::ListTopicsOpts::default(), + ) + .await?; + return Ok(filter_table_infos( + message_queue_topic_tables(topics), + filter, + limit, + offset, + object_types, + table_name_filter, + )); + } + { let connections = state.connections.read().await; if let Some(PoolKind::ExternalDriver { config, session, .. }) = connections.get(&pool_key) { @@ -3167,6 +3200,26 @@ for line in sys.stdin: } } + #[cfg(feature = "mq-admin")] + #[test] + fn message_queue_topics_are_exposed_as_table_metadata() { + let tables = super::message_queue_topic_tables(vec![crate::mq::TopicInfo { + name: "orders".to_string(), + short_name: "orders".to_string(), + partitioned: true, + partitions: Some(3), + persistent: true, + internal: false, + message_type: None, + namespace: Some("default".to_string()), + }]); + + assert_eq!(tables.len(), 1); + assert_eq!(tables[0].name, "orders"); + assert_eq!(tables[0].table_type, "TOPIC"); + assert_eq!(tables[0].parent_schema.as_deref(), Some("default")); + } + fn test_object_info(name: &str, object_type: &str) -> super::db::ObjectInfo { super::db::ObjectInfo { name: name.to_string(), diff --git a/crates/dbx-mcp/Cargo.toml b/crates/dbx-mcp/Cargo.toml index 34ef453a1..e572bb486 100644 --- a/crates/dbx-mcp/Cargo.toml +++ b/crates/dbx-mcp/Cargo.toml @@ -5,8 +5,9 @@ edition = "2021" license = "Apache-2.0" [features] -default = ["duckdb-sidecar"] +default = ["duckdb-sidecar", "mq-admin"] duckdb-sidecar = ["dbx-core/duckdb-sidecar"] +mq-admin = ["dbx-core/mq-admin"] [[bin]] name = "dbx-mcp" diff --git a/crates/dbx-mcp/tests/local.rs b/crates/dbx-mcp/tests/local.rs index 01ec14a62..0cc3ca9fa 100644 --- a/crates/dbx-mcp/tests/local.rs +++ b/crates/dbx-mcp/tests/local.rs @@ -259,6 +259,12 @@ async fn executes_mongo_shell_commands_without_desktop_process() { server_task.abort(); } +#[test] +#[cfg(feature = "mq-admin")] +fn mcp_default_features_include_message_queue_admin() { + assert_eq!(dbx_core::mq::MqSystemKind::Kafka.as_str(), "kafka"); +} + async fn call_query(client: &rmcp::service::RunningService, sql: &str) -> String { let arguments = json!({ "connection_id": "mongo-e2e", diff --git a/packages/mcp-server/README.md b/packages/mcp-server/README.md index e4afaaedd..63b2ccd45 100644 --- a/packages/mcp-server/README.md +++ b/packages/mcp-server/README.md @@ -144,7 +144,7 @@ Ask the MCP client to: | `dbx_list_connections` | List connections visible to the MCP session | | `dbx_add_connection` | Add a connection to DBX storage | | `dbx_remove_connection` | Remove a connection from DBX storage | -| `dbx_list_tables` | List tables, views, or collections | +| `dbx_list_tables` | List tables, views, collections, or message queue topics | | `dbx_describe_table` | Return columns and table metadata | | `dbx_get_schema_context` | Return compact schema context suitable for an AI model | | `dbx_execute_query` | Execute SQL or a supported MongoDB shell command, returning at most 100 rows | @@ -455,7 +455,7 @@ MCP 配置: | `dbx_list_connections` | 列出当前 MCP 会话可见的连接 | | `dbx_add_connection` | 添加 DBX 连接配置 | | `dbx_remove_connection` | 删除 DBX 连接配置 | -| `dbx_list_tables` | 列出表、视图或集合 | +| `dbx_list_tables` | 列出表、视图、集合或消息队列 Topic | | `dbx_describe_table` | 获取字段和表结构 | | `dbx_get_schema_context` | 获取适合 AI 使用的紧凑 Schema 上下文 | | `dbx_execute_query` | 执行 SQL 或支持的 MongoDB Shell 命令,最多返回 100 行 |