table_overview
Retrieve a table's column details, up to three sample rows, and total row count. Use refresh=true to bypass cached results.
Instructions
Get an overview of a specific table: columns, sample rows (up to 3), and total row count. Uses cache unless refresh=true
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| table | Yes | Table name, optionally prefixed with database name (e.g., 'db_name.table_name'). If database is omitted, uses the default database. | |
| refresh | No | Set to true to force refresh, ignoring cache. Defaults to false. |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- Helper function _get_table_details that fetches table overview: row count, columns (DESCRIBE), and sample rows. Used by table_overview tool.
def _get_table_details(db_name, table_name, limit=None): """ Helper function to get description, sample rows, and count for a table. Returns a formatted string. Handles DB errors internally and returns error messages. """ global global_table_overview_cache logger.debug(f"Fetching table details for {db_name}.{table_name}") output_lines = [] full_table_name = f"`{table_name}`" if db_name: full_table_name = f"`{db_name}`.`{table_name}`" else: output_lines.append( f"Warning: Database name missing for table '{table_name}'. Using potentially incorrect context.") logger.warning(f"Database name missing for table '{table_name}'") count = 0 output_lines.append(f"--- Overview for {full_table_name} ---") # 1. Get Row Count query = f"SELECT COUNT(*) FROM {full_table_name}" count_result = db_client.execute(query, db=db_name) if count_result.success and count_result.rows: count = count_result.rows[0][0] output_lines.append(f"\nTotal rows: {count}") logger.debug(f"Table {full_table_name} has {count} rows") else: output_lines.append(f"\nCould not determine total row count.") if not count_result.success: output_lines.append(f"Error: {count_result.error_message}") logger.error(f"Failed to get row count for {full_table_name}: {count_result.error_message}") # 2. Get Columns (DESCRIBE) if count > 0: query = f"DESCRIBE {full_table_name}" desc_result = db_client.execute(query, db=db_name) if desc_result.success and desc_result.column_names and desc_result.rows: output_lines.append(f"\nColumns:") output_lines.append(desc_result.to_string(limit=limit)) else: output_lines.append("(Could not retrieve column information or table has no columns).") if not desc_result.success: output_lines.append(f"Error getting columns for {full_table_name}: {desc_result.error_message}") return "\n".join(output_lines) # 3. Get Sample Rows (LIMIT 3) query = f"SELECT * FROM {full_table_name} LIMIT 3" sample_result = db_client.execute(query, db=db_name) if sample_result.success and sample_result.column_names and sample_result.rows: output_lines.append(f"\nSample rows (limit 3):") output_lines.append(sample_result.to_string(limit=limit)) else: output_lines.append(f"(No rows found in {full_table_name}).") if not sample_result.success: output_lines.append(f"Error getting sample rows for {full_table_name}: {sample_result.error_message}") overview_string = "\n".join(output_lines) # Update cache even if there were partial errors, so we cache the error message too cache_key = (db_name, table_name) global_table_overview_cache[cache_key] = overview_string return overview_string - Main tool handler for 'table_overview' - accepts table name (optionally db-qualified) and refresh flag. Parses input, checks cache, calls _get_table_details.
@mcp.tool(description="Get an overview of a specific table: columns, sample rows (up to 3), and total row count. Uses cache unless refresh=true" + description_suffix) def table_overview( table: Annotated[str, Field( description="Table name, optionally prefixed with database name (e.g., 'db_name.table_name'). If database is omitted, uses the default database.")], refresh: Annotated[ bool, Field(description="Set to true to force refresh, ignoring cache. Defaults to false.")] = False ) -> str: try: logger.info(f"Getting table overview for: {table}, refresh={refresh}") if not table: logger.error("Table overview called without table name") return "Error: Missing 'table' argument." # Parse table argument: [db.]<table> parts = table.split('.', 1) db_name = None table_name = None if len(parts) == 2: db_name, table_name = parts[0], parts[1] elif len(parts) == 1: table_name = parts[0] db_name = db_client.default_database # Use default if only table name is given if not table_name: # Should not happen if table_arg exists, but check logger.error(f"Invalid table name format: {table}") return f"Error: Invalid table name format '{table}'." if not db_name: logger.error(f"No database specified for table {table_name}") return f"Error: Database name not specified for table '{table_name}' and no default database is set." cache_key = (db_name, table_name) # Check cache if not refresh and cache_key in global_table_overview_cache: logger.debug(f"Using cached overview for {db_name}.{table_name}") return global_table_overview_cache[cache_key] logger.debug(f"Fetching fresh overview for {db_name}.{table_name}") # Fetch details (will also update cache) overview_text = _get_table_details(db_name, table_name, limit=overview_length_limit) return overview_text except Exception as e: # Reset connections on unexpected errors logger.exception(f"Unexpected error in table_overview for {table}") reset_db_connections() stack_trace = traceback.format_exc() return f"Unexpected Error executing tool 'table_overview': {type(e).__name__}: {e}\nStack Trace:\n{stack_trace}" - src/mcp_server_starrocks/server.py:406-412 (registration)Registration via @mcp.tool decorator on the table_overview function, with description about cache and refresh behavior.
@mcp.tool(description="Get an overview of a specific table: columns, sample rows (up to 3), and total row count. Uses cache unless refresh=true" + description_suffix) def table_overview( table: Annotated[str, Field( description="Table name, optionally prefixed with database name (e.g., 'db_name.table_name'). If database is omitted, uses the default database.")], refresh: Annotated[ bool, Field(description="Set to true to force refresh, ignoring cache. Defaults to false.")] = False ) -> str: - Input schema for table_overview: 'table' (string, optionally db.table format) and 'refresh' (bool, defaults to False).
def table_overview( table: Annotated[str, Field( description="Table name, optionally prefixed with database name (e.g., 'db_name.table_name'). If database is omitted, uses the default database.")], refresh: Annotated[ bool, Field(description="Set to true to force refresh, ignoring cache. Defaults to false.")] = False ) -> str: - Global cache dict for table overviews keyed by (db_name, table_name) tuple, and overview_length_limit configuration.
overview_length_limit = int(os.getenv('STARROCKS_OVERVIEW_LIMIT', str(20000))) # Global cache for table overviews: {(db_name, table_name): overview_string} global_table_overview_cache = {}