mcp-clickhouse-long-running

Instructions

Implementation Reference

• mcp_clickhouse/mcp_server.py:100-132 (handler)

  The @mcp.tool()-decorated handler function that implements the list_tables tool. It queries system.tables for table metadata in the given database (optionally filtered by like/not_like), then for each table queries system.columns to fetch column details, constructs Table and Column dataclasses, and returns serialized dicts.

  @mcp.tool()
  def list_tables(
      database: str, like: Optional[str] = None, not_like: Optional[str] = None
  ):
      """List available ClickHouse tables in a database, including schema, comment,
      row count, and column count."""
      logger.info(f"Listing tables in database '{database}'")
      client = create_clickhouse_client()
      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)}"
      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)
  
      # Deserialize result as Table dataclass instances
      tables = result_to_table(result.column_names, result.result_rows)
  
      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 = [
              c
              for c in result_to_column(
                  column_data_query_result.column_names,
                  column_data_query_result.result_rows,
              )
          ]
  
      logger.info(f"Found {len(tables)} tables")
      return [asdict(table) for table in tables]

• mcp_clickhouse/mcp_server.py:27-45 (schema)

  Table dataclass defining the output schema for tables returned by list_tables.

  @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:16-25 (schema)

  Column dataclass defining the schema for table columns.

  @dataclass
  class Column:
      database: str
      table: str
      name: str
      column_type: str
      default_kind: Optional[str]
      default_expression: Optional[str]
      comment: Optional[str]

• mcp_clickhouse/mcp_server.py:72-73 (helper)

  Helper function to convert query results to List[Table].

  def result_to_table(query_columns, result) -> List[Table]:
      return [Table(**dict(zip(query_columns, row))) for row in result]

• mcp_clickhouse/mcp_server.py:76-77 (helper)

  Helper function to convert query results to List[Column].

  def result_to_column(query_columns, result) -> List[Column]:
      return [Column(**dict(zip(query_columns, row))) for row in result]