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)