ClickHouse MCP Server

Implementation Reference

• mcp_clickhouse/mcp_server.py:37-81 (handler)

  The handler function for the 'list_tables' tool, registered via @mcp.tool(). It executes SHOW TABLES, enriches results with DESCRIBE TABLE and SHOW CREATE TABLE for each table, returning structured data including database, name, columns, and create query.

  @mcp.tool()
  def list_tables(database: str, like: str = None):
      logger.info(f"Listing tables in database '{database}'")
      client = create_clickhouse_client()
      query = f"SHOW TABLES FROM {database}"
      if like:
          query += f" LIKE '{like}'"
      result = client.command(query)
  
      def get_table_info(table):
          logger.info(f"Getting schema info for table {database}.{table}")
          schema_query = f"DESCRIBE TABLE {database}.`{table}`"
          schema_result = client.query(schema_query)
  
          columns = []
          column_names = schema_result.column_names
          for row in schema_result.result_rows:
              column_dict = {}
              for i, col_name in enumerate(column_names):
                  column_dict[col_name] = row[i]
              columns.append(column_dict)
  
          create_table_query = f"SHOW CREATE TABLE {database}.`{table}`"
          create_table_result = client.command(create_table_query)
  
          return {
              "database": database,
              "name": table,
              "columns": columns,
              "create_table_query": create_table_result,
          }
  
      tables = []
      if isinstance(result, str):
          # Single table result
          for table in (t.strip() for t in result.split()):
              if table:
                  tables.append(get_table_info(table))
      elif isinstance(result, Sequence):
          # Multiple table results
          for table in result:
              tables.append(get_table_info(table))
  
      logger.info(f"Found {len(tables)} tables")
      return tables

• mcp_clickhouse/mcp_server.py:104-115 (helper)

  Helper function used by list_tables (and other tools) to establish a ClickHouse client connection from environment variables.

  def create_clickhouse_client():
      host = os.getenv("CLICKHOUSE_HOST")
      port = os.getenv("CLICKHOUSE_PORT")
      username = os.getenv("CLICKHOUSE_USER")
      logger.info(f"Creating ClickHouse client connection to {host}:{port} as {username}")
      return clickhouse_connect.get_client(
          host=host,
          port=port,
          username=username,
          password=os.getenv("CLICKHOUSE_PASSWORD"),
      )

• mcp_clickhouse/mcp_server.py:37-37 (registration)

  The @mcp.tool() decorator registers the list_tables function with the FastMCP server.

  @mcp.tool()