StarRocks MCP Server

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• src/mcp_server_starrocks/server.py:456-513 (handler)

  The handler function implementing the 'db_overview' tool. It lists all tables in the specified database (or default), fetches overviews for each using cache or _get_table_details helper, and compiles them into a string response. Note: The @mcp.tool decorator is commented out (line 455), so it may not be actively registered.
  def db_overview( db: Annotated[str, Field( description="Database name. Optional: uses the default database if not provided.")] = None, refresh: Annotated[ bool, Field(description="Set to true to force refresh, ignoring cache. Defaults to false.")] = False ) -> str: try: db_name = db if db else db_client.default_database logger.info(f"Getting database overview for: {db_name}, refresh={refresh}") if not db_name: logger.error("Database overview called without database name") return "Error: Database name not provided and no default database is set." # List tables in the database query = f"SHOW TABLES FROM `{db_name}`" result = db_client.execute(query, db=db_name) if not result.success: logger.error(f"Failed to list tables in database {db_name}: {result.error_message}") return f"Database Error listing tables in '{db_name}': {result.error_message}" if not result.rows: logger.info(f"No tables found in database {db_name}") return f"No tables found in database '{db_name}'." tables = [row[0] for row in result.rows] logger.info(f"Found {len(tables)} tables in database {db_name}") all_overviews = [f"--- Overview for Database: `{db_name}` ({len(tables)} tables) ---"] total_length = 0 limit_per_table = overview_length_limit * (math.log10(len(tables)) + 1) // len(tables) # Limit per table for table_name in tables: cache_key = (db_name, table_name) overview_text = None # Check cache first if not refresh and cache_key in global_table_overview_cache: logger.debug(f"Using cached overview for {db_name}.{table_name}") overview_text = global_table_overview_cache[cache_key] else: logger.debug(f"Fetching fresh overview for {db_name}.{table_name}") # Fetch details for this table (will update cache via _get_table_details) overview_text = _get_table_details(db_name, table_name, limit=limit_per_table) all_overviews.append(overview_text) all_overviews.append("\n") # Add separator total_length += len(overview_text) + 1 logger.info(f"Database overview completed for {db_name}, total length: {total_length}") return "\n".join(all_overviews) except Exception as e: # Catch any other unexpected errors during tool execution logger.exception(f"Unexpected error in db_overview for database {db}") reset_db_connections() stack_trace = traceback.format_exc() return f"Unexpected Error executing tool 'db_overview': {type(e).__name__}: {e}\nStack Trace:\n{stack_trace}"
• src/mcp_server_starrocks/server.py:125-187 (helper)

  Supporting helper function called by db_overview (and table_overview) to generate detailed overview for a single table: row count, DESCRIBE columns, and sample rows (LIMIT 3). 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:455-455 (registration)

  Commented-out registration decorator for the db_overview tool using @mcp.tool. Indicates it is not actively registered, preferring db_summary instead.
  #@mcp.tool(description="Get an overview (columns, sample rows, row count) for ALL tables in a database. Uses cache unless refresh=True" + description_suffix)