Add query support to local MCP server

crates/tower-cmd/src/catalogs.rs

@@ -379,49 +379,10 @@ async fn fetch_catalog_tables(
 ) -> Result<QueryResult, String> {
     let mut spinner = out.spinner("Listing tables...");
 
-    let response = match api::vend_catalog_credentials(
-        config,
-        name,
-        env,
-        vend_catalog_credentials_body::Mode::Read,
-    )
-    .await
-    {
-        Ok(response) => response,
-        Err(err) => {
-            spinner.failure(out);
-            return Err(err.to_string());
-        }
-    };
-
-    let token = response.credentials.oauth_token.clone();
-
-    // `--full` needs every table's column schema. Going through DuckDB means a
-    // `DESCRIBE` per table, each of which fully opens the Iceberg table (reading
-    // manifests from object storage) — slow, and unbounded for tables with heavy
-    // metadata. The Iceberg REST catalog's `loadTable` returns the schema
-    // straight from table metadata with no manifest I/O, so `--full` talks to
-    // the catalog directly. The plain listing stays on DuckDB.
     let result = if full {
-        fetch_catalog_columns_via_rest(&response.credentials).await
+        list_catalog_columns(config, name, env).await
     } else {
-        let setup = attach_statements(
-            name,
-            &response.credentials,
-            vend_catalog_credentials_body::Mode::Read,
-        );
-        let db_name = name.to_string();
-        tokio::task::spawn_blocking(move || {
-            run_query(
-                &setup,
-                "SELECT \"schema\", name FROM (SHOW ALL TABLES) WHERE database = ? ORDER BY \"schema\", name",
-                params![db_name],
-                &Limits::none(),
-            )
-        })
-        .await
-        .map_err(|err| err.to_string())
-        .and_then(|inner| inner.map_err(|err| err.to_string()))
+        list_catalog_tables(config, name, env).await
     };
 
     match result {
@@ -431,11 +392,68 @@ async fn fetch_catalog_tables(
         }
         Err(err) => {
             spinner.failure(out);
-            Err(redact_token(&err, &token))
+            Err(err)
         }
     }
 }
 
+/// Attaches a storage catalog read-only and returns its (namespace, table)
+/// rows via `SHOW ALL TABLES`. Shared by the CLI `show` and the MCP server.
+/// Errors are returned with the OAuth token redacted.
+pub(crate) async fn list_catalog_tables(
+    config: &Config,
+    name: &str,
+    env: &str,
+) -> Result<QueryResult, String> {
+    let response =
+        api::vend_catalog_credentials(config, name, env, vend_catalog_credentials_body::Mode::Read)
+            .await
+            .map_err(|err| err.to_string())?;
+
+    let token = response.credentials.oauth_token.clone();
+    let setup = attach_statements(
+        name,
+        &response.credentials,
+        vend_catalog_credentials_body::Mode::Read,
+    );
+    let db_name = name.to_string();
+
+    tokio::task::spawn_blocking(move || {
+        run_query(
+            &setup,
+            "SELECT \"schema\", name FROM (SHOW ALL TABLES) WHERE database = ? ORDER BY \"schema\", name",
+            params![db_name],
+            &Limits::none(),
+        )
+    })
+    .await
+    .map_err(|err| err.to_string())
+    .and_then(|inner| inner.map_err(|err| err.to_string()))
+    .map_err(|err| redact_token(&err, &token))
+}
+
+/// The `--full` listing: every table's column schema. Going through DuckDB
+/// means a `DESCRIBE` per table, each of which fully opens the Iceberg table
+/// (reading manifests from object storage) — slow, and unbounded for tables
+/// with heavy metadata. The Iceberg REST catalog's `loadTable` returns the
+/// schema straight from table metadata with no manifest I/O, so this talks to
+/// the catalog directly. The plain listing stays on DuckDB.
+async fn list_catalog_columns(
+    config: &Config,
+    name: &str,
+    env: &str,
+) -> Result<QueryResult, String> {
+    let response =
+        api::vend_catalog_credentials(config, name, env, vend_catalog_credentials_body::Mode::Read)
+            .await
+            .map_err(|err| err.to_string())?;
+
+    let token = response.credentials.oauth_token.clone();
+    fetch_catalog_columns_via_rest(&response.credentials)
+        .await
+        .map_err(|err| redact_token(&err, &token))
+}
+
 /// DuckDB errors can echo the failing statement, and the setup batch contains
 /// the vended OAuth token — scrub it before the message reaches any output.
 fn redact_token(message: &str, token: &str) -> String {
@@ -608,6 +626,88 @@ async fn execute_catalog_query(
     }
 }
 
+/// Agent-facing counterpart of `do_query`, used by the MCP server: the same
+/// read-only gate, storage-type check, sandbox, and ceilings, but errors are
+/// returned rather than printed, and the caps are non-negotiable — an agent
+/// has no `--write` or `--max-rows` escape hatch.
+pub(crate) async fn query_catalog_for_agent(
+    config: &Config,
+    name: &str,
+    env: &str,
+    sql: String,
+) -> Result<QueryResult, String> {
+    let sql_to_check = sql.clone();
+    let verdict = tokio::task::spawn_blocking(move || guard::classify_read_only(&sql_to_check))
+        .await
+        .map_err(|err| format!("Could not validate the query: {err}"))?
+        .map_err(|err| format!("Could not validate the query: {err}"))?;
+    match verdict {
+        guard::ReadOnlyCheck::Allowed => {}
+        guard::ReadOnlyCheck::Empty => {
+            return Err("No SQL statement provided.".to_string());
+        }
+        guard::ReadOnlyCheck::Multiple => {
+            return Err("Only a single SQL statement can be run per query.".to_string());
+        }
+        guard::ReadOnlyCheck::NotReadOnly => {
+            return Err(
+                "This query is read-only; only a single SELECT statement is allowed.".to_string(),
+            );
+        }
+        guard::ReadOnlyCheck::DeniedFunction(function) => {
+            return Err(format!(
+                "'{function}' is not allowed in a read-only query: it changes engine state, runs SQL built at runtime, or reads outside the catalog."
+            ));
+        }
+        guard::ReadOnlyCheck::DeniedTableReference(reference) => {
+            return Err(format!(
+                "'{reference}' is a file or URL, not a table in this catalog. Read-only queries can only read the catalog's own tables."
+            ));
+        }
+    }
+
+    let response = api::describe_catalog(config, name, env)
+        .await
+        .map_err(|err| format!("Fetching catalog details failed: {err}"))?;
+    if !is_storage_catalog_type(Some(&response.catalog.r#type)) {
+        return Err(format!(
+            "Querying is only supported for {} catalogs; '{}' has type '{}'.",
+            STORAGE_CATALOG_TYPE, name, response.catalog.r#type
+        ));
+    }
+
+    let response =
+        api::vend_catalog_credentials(config, name, env, vend_catalog_credentials_body::Mode::Read)
+            .await
+            .map_err(|err| format!("Running query failed: {err}"))?;
+
+    let token = response.credentials.oauth_token.clone();
+    let setup = attach_statements(
+        name,
+        &response.credentials,
+        vend_catalog_credentials_body::Mode::Read,
+    );
+    let result = tokio::task::spawn_blocking(move || -> Result<QueryResult, tower_duckdb::Error> {
+        let session = Session::open()?;
+        session.run_setup(&setup)?;
+        session.harden(&Hardening::agent())?;
+        session.query(&sql, [], &Limits::agent())
+    })
+    .await;
+
+    match result {
+        Ok(Ok(query_result)) => Ok(query_result),
+        Ok(Err(err)) => Err(format!(
+            "Query failed: {}",
+            redact_token(&err.to_string(), &token)
+        )),
+        Err(err) => Err(format!(
+            "Query execution panicked: {}",
+            redact_token(&err.to_string(), &token)
+        )),
+    }
+}
+
 fn read_sql_from_stdin(out: &output::Out) -> String {
     let mut stdin = std::io::stdin();
     if stdin.is_terminal() {
@@ -1077,7 +1177,7 @@ fn json_array_to_strings(value: Option<&serde_json::Value>) -> Vec<String> {
     }
 }
 
-fn is_storage_catalog_type(catalog_type: Option<&str>) -> bool {
+pub(crate) fn is_storage_catalog_type(catalog_type: Option<&str>) -> bool {
     catalog_type == Some(STORAGE_CATALOG_TYPE)
 }
 

crates/tower-cmd/src/mcp.rs

@@ -199,6 +199,18 @@ struct ShowCatalogRequest {
     environment: Option<String>,
 }
 
+#[derive(Debug, Deserialize, JsonSchema)]
+struct QueryCatalogRequest {
+    /// Name of the Tower-managed storage catalog to query
+    name: String,
+    /// A single read-only SQL statement. Tables must be fully qualified as
+    /// <catalog>.<namespace>.<table> (there is no default schema). Writes, DDL,
+    /// and multiple statements are rejected.
+    sql: String,
+    /// The environment the catalog belongs to (defaults to "default")
+    environment: Option<String>,
+}
+
 pub fn mcp_cmd() -> Command {
     Command::new("mcp-server")
         .about("Runs an MCP server for LLM interaction")
@@ -719,7 +731,9 @@ impl TowerService {
         }
     }
 
-    #[tool(description = "Show details for a catalog, including its property names")]
+    #[tool(
+        description = "Show a catalog's details: its property names and, for Tower-managed storage catalogs, the namespaces and tables you can query."
+    )]
     async fn tower_catalogs_show(
         &self,
         Parameters(request): Parameters<ShowCatalogRequest>,
@@ -739,17 +753,88 @@ impl TowerService {
                         })
                     })
                     .collect();
+
+                // Only Tower-managed storage catalogs expose queryable tables;
+                // for anything else `tables` stays null. A listing failure is
+                // surfaced in `tables_error` without failing the whole call.
+                let (tables, tables_error) = if crate::catalogs::is_storage_catalog_type(Some(
+                    &catalog.r#type,
+                )) {
+                    match crate::catalogs::list_catalog_tables(
+                            &self.config,
+                            &request.name,
+                            environment,
+                        )
+                        .await
+                        {
+                            Ok(result) => (
+                                Value::Array(
+                                    result
+                                        .rows
+                                        .iter()
+                                        .map(|row| {
+                                            json!({
+                                                "namespace": row.first().cloned().unwrap_or(Value::Null),
+                                                "table": row.get(1).cloned().unwrap_or(Value::Null),
+                                            })
+                                        })
+                                        .collect(),
+                                ),
+                                Value::Null,
+                            ),
+                            Err(e) => (Value::Null, Value::String(e)),
+                        }
+                } else {
+                    (Value::Null, Value::Null)
+                };
+
                 Self::json_success(json!({
                     "name": catalog.name,
                     "type": catalog.r#type,
                     "environment": catalog.environment,
                     "properties": properties,
+                    "tables": tables,
+                    "tables_error": tables_error,
                 }))
             }
             Err(e) => Self::error_result("Failed to show catalog", e),
         }
     }
 
+    #[tool(
+        description = "Run one read-only SQL statement against a Tower-managed storage (Iceberg) catalog and return the columns plus rows as positional arrays (one value per column, in column order). Fully qualify tables as \"<catalog>\".\"<namespace>\".\"<table>\"; call tower_catalogs_show first to list a catalog's namespaces and tables. Only a single statement runs per call, and writes/DDL are rejected. Results are capped (1000 rows, 1 MiB, 60s) — when a cap was hit, the response sets \"truncated\": true, so narrow with WHERE/LIMIT or aggregate."
+    )]
+    async fn tower_catalogs_query(
+        &self,
+        Parameters(request): Parameters<QueryCatalogRequest>,
+    ) -> Result<CallToolResult, McpError> {
+        let environment = request.environment.as_deref().unwrap_or("default");
+        let sql = request.sql.trim().to_string();
+        if sql.is_empty() {
+            return Self::text_error("No SQL statement provided.".to_string());
+        }
+
+        // Agents get read-only, sandboxed, result-capped access; positional rows
+        // preserve duplicate column names (e.g. from joins) that an object keyed
+        // by name would silently collapse.
+        match crate::catalogs::query_catalog_for_agent(
+            &self.config,
+            &request.name,
+            environment,
+            sql,
+        )
+        .await
+        {
+            Ok(result) => Self::json_success(json!({
+                "columns": result.columns,
+                "rows": result.rows,
+                "row_count": result.rows.len(),
+                "truncated": result.is_truncated(),
+            })),
+            Err(e) => Self::text_error(e),
+        }
+    }
+
     #[tool(description = "List teams you belong to")]
     async fn tower_teams_list(&self) -> Result<CallToolResult, McpError> {
         if self.config.api_key.is_some() {

tests/integration/features/environment.py

@@ -16,6 +16,25 @@ def before_all(context):
 
 
 def before_scenario(context, scenario):
+    # Catalog scenarios hit the real attach -> Iceberg -> query path, which the
+    # mock server can't stand up. Skip them unless a storage catalog is
+    # configured (and rely on a real TOWER_URL + session being provided too).
+    if "catalogs" in scenario.effective_tags:
+        catalog = os.environ.get("TOWER_TEST_CATALOG")
+        if not catalog:
+            scenario.skip(
+                "set TOWER_TEST_CATALOG (and a real TOWER_URL) to run catalog tests"
+            )
+            return
+        context.test_catalog = catalog
+        context.test_catalog_env = os.environ.get("TOWER_TEST_CATALOG_ENV", "default")
+        context.test_catalog_table = os.environ.get("TOWER_TEST_CATALOG_TABLE")
+        if "catalog-data" in scenario.effective_tags and not context.test_catalog_table:
+            scenario.skip(
+                "set TOWER_TEST_CATALOG_TABLE to run the catalog data query test"
+            )
+            return
+
     # Create a temporary working directory for this scenario
     context.temp_dir = tempfile.mkdtemp(prefix="tower_test_")
     context.original_cwd = os.getcwd()