mcp-clickhouse

Instructions

Implementation Reference

• mcp_clickhouse/mcp_server.py:266-384 (handler)

  Core handler function for the 'list_tables' tool. Implements table listing with support for LIKE/NOT LIKE filters, pagination (page_token, page_size), detailed column info, and caching for pagination state. Uses helper functions to query system.tables and system.columns.
  def list_tables( database: str, like: Optional[str] = None, not_like: Optional[str] = None, page_token: Optional[str] = None, page_size: int = 50, include_detailed_columns: bool = True, ) -> Dict[str, Any]: """List available ClickHouse tables in a database, including schema, comment, row count, and column count. Args: database: The database to list tables from like: Optional LIKE pattern to filter table names not_like: Optional NOT LIKE pattern to exclude table names page_token: Token for pagination, obtained from a previous call page_size: Number of tables to return per page (default: 50) include_detailed_columns: Whether to include detailed column metadata (default: True). When False, the columns array will be empty but create_table_query still contains all column information. This reduces payload size for large schemas. Returns: A dictionary containing: - tables: List of table information (as dictionaries) - next_page_token: Token for the next page, or None if no more pages - total_tables: Total number of tables matching the filters """ logger.info( "Listing tables in database '%s' with like=%s, not_like=%s, " "page_token=%s, page_size=%s, include_detailed_columns=%s", database, like, not_like, page_token, page_size, include_detailed_columns, ) client = create_clickhouse_client() if page_token and page_token in table_pagination_cache: cached_state = table_pagination_cache[page_token] cached_include_detailed = cached_state.get("include_detailed_columns", True) if ( cached_state["database"] != database or cached_state["like"] != like or cached_state["not_like"] != not_like or cached_include_detailed != include_detailed_columns ): logger.warning( "Page token %s is for a different database, filter, or metadata setting. " "Ignoring token and starting from beginning.", page_token, ) page_token = None else: table_names = cached_state["table_names"] start_idx = cached_state["start_idx"] tables, end_idx, has_more = get_paginated_table_data( client, database, table_names, start_idx, page_size, include_detailed_columns, ) next_page_token = None if has_more: next_page_token = create_page_token( database, like, not_like, table_names, end_idx, include_detailed_columns ) del table_pagination_cache[page_token] logger.info( "Returned page with %s tables (total: %s), next_page_token=%s", len(tables), len(table_names), next_page_token, ) return { "tables": [asdict(table) for table in tables], "next_page_token": next_page_token, "total_tables": len(table_names), } table_names = fetch_table_names_from_system(client, database, like, not_like) start_idx = 0 tables, end_idx, has_more = get_paginated_table_data( client, database, table_names, start_idx, page_size, include_detailed_columns, ) next_page_token = None if has_more: next_page_token = create_page_token( database, like, not_like, table_names, end_idx, include_detailed_columns ) logger.info( "Found %s tables, returning %s with next_page_token=%s", len(table_names), len(tables), next_page_token, ) return { "tables": [asdict(table) for table in tables], "next_page_token": next_page_token, "total_tables": len(table_names), }
• mcp_clickhouse/mcp_server.py:37-55 (schema)

  Dataclass defining the output Table structure used by list_tables, including metadata like engine, row counts, and nested columns.
  @dataclass class Table: database: str name: str engine: str create_table_query: str dependencies_database: str dependencies_table: str engine_full: str sorting_key: str primary_key: str total_rows: int total_bytes: int total_bytes_uncompressed: int parts: int active_parts: int total_marks: int comment: Optional[str] = None columns: List[Column] = field(default_factory=list)
• mcp_clickhouse/mcp_server.py:562-566 (registration)

  Registers the list_tables tool (along with others) to the FastMCP server instance when CLICKHOUSE_ENABLED is true.
  if os.getenv("CLICKHOUSE_ENABLED", "true").lower() == "true": mcp.add_tool(Tool.from_function(list_databases)) mcp.add_tool(Tool.from_function(list_tables)) mcp.add_tool(Tool.from_function(run_select_query)) logger.info("ClickHouse tools registered")
• mcp_clickhouse/mcp_server.py:145-172 (helper)

  Helper function to fetch filtered table names from system.tables, used by list_tables.
  def fetch_table_names_from_system( client, database: str, like: Optional[str] = None, not_like: Optional[str] = None, ) -> List[str]: """Get list of table names from system.tables. Args: client: ClickHouse client database: Database name like: Optional pattern to filter table names (LIKE) not_like: Optional pattern to filter out table names (NOT LIKE) Returns: List of table names """ query = f"SELECT name FROM system.tables WHERE database = {format_query_value(database)}" if like: query += f" AND name LIKE {format_query_value(like)}" if not_like: query += f" AND name NOT LIKE {format_query_value(not_like)}" result = client.query(query) table_names = [row[0] for row in result.result_rows] return table_names
• mcp_clickhouse/mcp_server.py:174-231 (helper)

  Helper function to retrieve detailed paginated table data including optional column details from system.tables and system.columns, used by list_tables.
  def get_paginated_table_data( client, database: str, table_names: List[str], start_idx: int, page_size: int, include_detailed_columns: bool = True, ) -> tuple[List[Table], int, bool]: """Get detailed information for a page of tables. Args: client: ClickHouse client database: Database name table_names: List of all table names to paginate start_idx: Starting index for pagination page_size: Number of tables per page include_detailed_columns: Whether to include detailed column metadata (default: True) Returns: Tuple of (list of Table objects, end index, has more pages) """ end_idx = min(start_idx + page_size, len(table_names)) current_page_table_names = table_names[start_idx:end_idx] if not current_page_table_names: return [], end_idx, False query = f""" SELECT database, name, engine, create_table_query, dependencies_database, dependencies_table, engine_full, sorting_key, primary_key, total_rows, total_bytes, total_bytes_uncompressed, parts, active_parts, total_marks, comment FROM system.tables WHERE database = {format_query_value(database)} AND name IN ({", ".join(format_query_value(name) for name in current_page_table_names)}) """ result = client.query(query) tables = result_to_table(result.column_names, result.result_rows) if include_detailed_columns: for table in tables: column_data_query = f""" SELECT database, table, name, type AS column_type, default_kind, default_expression, comment FROM system.columns WHERE database = {format_query_value(database)} AND table = {format_query_value(table.name)} """ column_data_query_result = client.query(column_data_query) table.columns = result_to_column( column_data_query_result.column_names, column_data_query_result.result_rows, ) else: for table in tables: table.columns = [] return tables, end_idx, end_idx < len(table_names)