diff --git a/crates/tower-cmd/src/catalogs.rs b/crates/tower-cmd/src/catalogs.rs index cebc682a..1bc6ea19 100644 --- a/crates/tower-cmd/src/catalogs.rs +++ b/crates/tower-cmd/src/catalogs.rs @@ -379,49 +379,10 @@ async fn fetch_catalog_tables( ) -> Result { 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 { + 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 { + 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 { + 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 { + 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 { } } -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) } diff --git a/crates/tower-cmd/src/mcp.rs b/crates/tower-cmd/src/mcp.rs index fc301457..7ccbce56 100644 --- a/crates/tower-cmd/src/mcp.rs +++ b/crates/tower-cmd/src/mcp.rs @@ -199,6 +199,18 @@ struct ShowCatalogRequest { environment: Option, } +#[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 + /// .. (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, +} + 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, @@ -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 \"\".\"\".\"
\"; 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, + ) -> Result { + 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 { if self.config.api_key.is_some() { diff --git a/tests/integration/features/environment.py b/tests/integration/features/environment.py index b6fb2d71..c3227b4b 100644 --- a/tests/integration/features/environment.py +++ b/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() diff --git a/tests/integration/features/mcp_catalogs.feature b/tests/integration/features/mcp_catalogs.feature new file mode 100644 index 00000000..97671112 --- /dev/null +++ b/tests/integration/features/mcp_catalogs.feature @@ -0,0 +1,35 @@ +@catalogs +Feature: MCP catalog querying + As a developer using the Tower MCP server with an agent + I want to query Tower-managed storage catalogs safely + So that an agent can explore data without write access or unbounded results + + # These scenarios exercise the real attach -> Iceberg -> query path, so they + # are skipped unless a storage catalog is configured (see environment.py): + # TOWER_TEST_CATALOG name of a tower-catalog storage catalog + # TOWER_URL a real server (not the mock) with a valid session + # TOWER_TEST_CATALOG_ENV environment the catalog lives in (default: default) + # TOWER_TEST_CATALOG_TABLE optional "cat"."ns"."tbl" for the data query below + + Scenario: Show a storage catalog lists its tables + When I show the test catalog via MCP + Then I should receive a success response + And the catalog response should list tables + + Scenario: Run a read-only query against the catalog + When I query the test catalog with SQL "SELECT 1 AS one" via MCP + Then I should receive query results + And the query result columns should include "one" + + Scenario: Write statements are rejected + When I query the test catalog with SQL "DROP TABLE does_not_exist" via MCP + Then I should receive an error response about a read-only query + + Scenario: Multiple statements are rejected + When I query the test catalog with SQL "SELECT 1; SELECT 2" via MCP + Then I should receive an error response about a single statement + + @catalog-data + Scenario: Query a configured table returns positional rows + When I query the configured catalog table via MCP + Then I should receive query results diff --git a/tests/integration/features/steps/mcp_steps.py b/tests/integration/features/steps/mcp_steps.py index 1fa07c3d..9a668c4b 100644 --- a/tests/integration/features/steps/mcp_steps.py +++ b/tests/integration/features/steps/mcp_steps.py @@ -904,3 +904,120 @@ def step_then_receive_workflow_help_stdio(context): @given('I have a simple hello world application named "{app_name}"') def step_create_hello_world_app_named(context, app_name): create_towerfile(context, app_name=app_name) + + +# --- Catalog querying (gated on TOWER_TEST_CATALOG; see environment.py) ------- + + +def _first_text_content(response): + """Return the first text block of an MCP response, or ''.""" + for item in response.get("content", []): + if isinstance(item, dict): + if item.get("type") == "text": + return item.get("text", "") + elif getattr(item, "type", None) == "text": + return getattr(item, "text", "") + return "" + + +def _parse_json_content(response): + """Parse the response's text block as JSON, or None if it isn't JSON.""" + try: + return json.loads(_first_text_content(response)) + except (ValueError, TypeError): + return None + + +@when("I show the test catalog via MCP") +@async_run_until_complete +async def step_show_test_catalog(context): + await call_mcp_tool( + context, + "tower_catalogs_show", + {"name": context.test_catalog, "environment": context.test_catalog_env}, + ) + + +@when('I query the test catalog with SQL "{sql}" via MCP') +@async_run_until_complete +async def step_query_test_catalog(context, sql): + await call_mcp_tool( + context, + "tower_catalogs_query", + { + "name": context.test_catalog, + "environment": context.test_catalog_env, + "sql": sql, + }, + ) + + +@when("I query the configured catalog table via MCP") +@async_run_until_complete +async def step_query_configured_table(context): + await call_mcp_tool( + context, + "tower_catalogs_query", + { + "name": context.test_catalog, + "environment": context.test_catalog_env, + "sql": f"SELECT * FROM {context.test_catalog_table} LIMIT 5", + }, + ) + + +@then("the catalog response should list tables") +def step_catalog_lists_tables(context): + data = _parse_json_content(context.mcp_response) + assert ( + data is not None + ), f"Expected JSON content, got: {context.mcp_response.get('content')}" + assert "tables" in data, f"Response should include a 'tables' field, got: {data}" + + +@then("I should receive query results") +def step_receive_query_results(context): + assert context.mcp_response.get( + "success", False + ), f"Expected a successful query, got: {context.mcp_response}" + data = _parse_json_content(context.mcp_response) + assert ( + data is not None + ), f"Expected JSON content, got: {context.mcp_response.get('content')}" + assert isinstance( + data.get("columns"), list + ), f"Expected a 'columns' list, got: {data}" + assert isinstance(data.get("rows"), list), f"Expected a 'rows' list, got: {data}" + # Rows are positional arrays (one value per column), not objects keyed by name. + for row in data["rows"]: + assert isinstance(row, list), f"Expected positional row arrays, got: {row}" + + +@then('the query result columns should include "{column}"') +def step_query_columns_include(context, column): + data = _parse_json_content(context.mcp_response) + assert data is not None and column in data.get( + "columns", [] + ), f"Expected column '{column}', got: {data}" + + +@then("I should receive an error response about a read-only query") +def step_error_read_only(context): + assert is_error_response( + context.mcp_response + ), f"Expected an error response, got: {context.mcp_response}" + text = str(context.mcp_response).lower() + assert ( + "read-only" in text or "not allowed" in text + ), f"Error should mention the read-only restriction, got: {context.mcp_response}" + + +@then("I should receive an error response about a single statement") +def step_error_single_statement(context): + assert is_error_response( + context.mcp_response + ), f"Expected an error response, got: {context.mcp_response}" + text = str(context.mcp_response).lower() + assert ( + "single" in text and "statement" in text + ), f"Error should mention the single-statement rule, got: {context.mcp_response}"