table_overview
View table details, including columns, sample rows, and row count. Use the refresh option to bypass cache and get updated data from the StarRocks database.
Instructions
Get an overview of a specific table: columns, sample rows (up to 5), and total row count. Uses cache unless refresh=true
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| refresh | No | Set to true to force refresh, ignoring cache. Defaults to false. | |
| table | Yes | Table name, optionally prefixed with database name (e.g., 'db_name.table_name'). If database is omitted, uses the default database. |
Implementation Reference
- Handler function for the 'table_overview' tool. Includes registration via @mcp.tool decorator, input schema via Annotated Fields, caching logic, parsing of table parameter, error handling, and invocation of helper function to retrieve table details.
@mcp.tool(description="Get an overview of a specific table: columns, sample rows (up to 5), 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}" - Pydantic input schema for 'table_overview' tool parameters: 'table' (string, table name possibly with db prefix) and 'refresh' (boolean, optional, default 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: - Supporting helper function _get_table_details(db_name, table_name, limit) that performs the core queries (COUNT(*), DESCRIBE, SELECT LIMIT 3), formats the overview string, and updates the global cache.
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 - src/mcp_server_starrocks/server.py:406-406 (registration)Registration of the 'table_overview' tool via FastMCP @mcp.tool decorator with description.
@mcp.tool(description="Get an overview of a specific table: columns, sample rows (up to 5), and total row count. Uses cache unless refresh=true" + description_suffix) - Global cache dictionary for storing table overview strings, keyed by (db_name, table_name). Used by table_overview and db_overview tools.
global_table_overview_cache = {}