深入探究 ClickHouse 的 MCP（Model Context Protocol）实现，从零开始搭建到开源分享，带你系统掌握 ClickHouse 的数据处理与分析能力。

核心内容：

1. ClickHouse MCP 的实现背景与开发动机
2. 构建 ClickHouse MCP 的完整步骤与代码详细解析
3. 写入 ClickHouse 的模拟数据生成与表结构设计

背景

近期看到一篇关于 GreptimeDB 集成 MCP 的文章，其演示效果非常酷。受此启发，我尝试使用 ClickHouse 也实现一套类似的 MCP 服务器，前后大约花了几个小时。目前项目已在 GitHub 开源（dubin555/clickhouse_mcp_server）。尽管 GitHub 上已有两三个现成的实现，但当前版本在代码注释完整度和文档详尽程度上，应该算是做得比较充分的。

效果

写数据

首先向 ClickHouse 写入一些模拟数据，这里准备了一组销售数据（可随意替换为其他数据，利用豆包、元宝等工具生成假数据非常便捷）。

-- Create sales analysis table with comments
CREATE TABLE IF NOT EXISTS default.city_sales
(
    city String COMMENT 'Name of the city where the sale occurred',
    product_category Enum('Electronics' = 1, 'Apparel' = 2, 'Grocery' = 3) COMMENT 'Category of the product sold',
    sale_date Date COMMENT 'Date of the sales transaction',
    units_sold UInt32 COMMENT 'Number of units sold in the transaction',
    unit_price Float32 COMMENT 'Price per unit in USD',
    total_sales Float32 MATERIALIZED units_sold * unit_price COMMENT 'Calculated total sales amount'
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(sale_date)
ORDER BY (city, product_category, sale_date)
COMMENT 'Table storing city-wise product sales data for business analysis';

-- Generate 10,000 random sales records
INSERT INTO default.city_sales (city, product_category, sale_date, units_sold, unit_price)
SELECT
    ['New York', 'London', 'Tokyo', 'Paris', 'Singapore', 'Dubai'][rand() % 6 + 1] AS city,
    toInt16(rand() % 3 + 1) AS product_category,
    today() - rand() % 365 AS sale_date,
    rand() % 100 + 1 AS units_sold,      -- Units between 1-100
    randNormal(50, 15) AS unit_price     -- Normal distribution around $50
FROM numbers(10000);

表中的字段涵盖城市、销售品类、产品、销售额等多个维度，便于进行多角度分析。

提问

接下来通过自然语言进行查询（此处使用的客户端是 VSCode 的 Cline 插件）：

• 各城市的销售额分别是多少？
• 哪种商品最畅销？

LLM 的第一次调用：

实现

在 GitHub 上参考了两三个已有的实现，整体逻辑并不复杂。完整的代码可前往仓库查看，这里重点讲解最关键的一个文件——server.py。

整体

其中包含 5 个主要类：

• ClickHouseClient：负责创建 ClickHouse 连接并执行查询操作。
• TableMetadataManager：负责查询表结构的元数据，例如字段列表、注释等信息。
• ResourceManager：负责构造供 LLM 使用的资源提示，展示有哪些可访问的资源，内部会调用 TableMetadataManager。
• ToolManager：负责告知 LLM 有哪些可用的工具（Tool），并执行这些工具的调用，内部会调用 ClickHouseClient。
• DatabaseServer：整合上面 4 个类的功能，完成最终的 MCP 服务器。

具体实现

ClickHouseClient

class ClickHouseClient:
    """ClickHouse database client"""

    def __init__(self, config: Config, logger: Logger):
        self.logger = logger
        self.db_config = {
            "host": config.host,
            "port": int(config.port),
            "user": config.user,
            "password": config.password,
            "database": config.database
        }
        self._client = None

    def get_client(self):
        """Get ClickHouse client, singleton pattern"""
        if self._client is None:
            self._client = self._create_client()
        return self._client

    def _create_client(self):
        """Create a new ClickHouse client"""
        try:
            self.logger.debug(f"Creating ClickHouse client with config: {self.db_config}")
            client = clickhouse_connect.get_client(**self.db_config)
            version = client.server_version
            self.logger.info("ClickHouse client created successfully")
            return client
        except Exception as e:
            self.logger.error(f"Failed to create ClickHouse client: {e}")
            raise

    def execute_query(self, query: str, readonly: bool = True):
        """Execute a query against the ClickHouse database"""
        try:
            client = self.get_client()
            settings = {"readonly": 1} if readonly else {}
            res = client.query(query, settings=settings)

            # convert result to list of dicts
            rows = []
            for row in res.result_rows:
                row_dict = {}
                for i, col_name in enumerate(res.column_names):
                    row_dict[col_name] = row[i]
                rows.append(row_dict)
                
            self.logger.debug(f"Query executed successfully: {query}")
            return rows
        except Exception as e:
            self.logger.error(f"Failed to execute query: {e}")
            raise

ClickHouseClient 类采用单例模式管理数据库连接，并支持设置只读模式执行查询，确保数据安全。

TableMetadataManager

class TableMetadataManager:
    """Manage table metadata in ClickHouse"""
    def __init__(self, client: ClickHouseClient, logger: Logger):
        self.client = client
        self.logger = logger

    def get_table_list(self, database: str) -> List[str]:
        """Get list of tables in the database"""
        query = f"SHOW TABLES FROM {quote_identifier(database)}"
        result = self.client.execute_query(query)
        if not result:
            return []
        return [row[next(iter(row.keys()))] for row in result]

    def get_table_comments(self, database: str) -> Dict[str, str]:
        """Get comments for the tables in the database"""
        query = f"SELECT name, comment FROM system.tables WHERE database = {format_query_value(database)}"
        result = self.client.execute_query(query)
        return {row['name']: row['comment'] for row in result}

    def get_column_comments(self, database: str) -> Dict[str, Dict[str, str]]:
        """Get comments for the columns in the tables in the database"""
        query = f"SELECT table, name, comment FROM system.columns WHERE database = {format_query_value(database)}"
        result = self.client.execute_query(query)

        column_comments = {}
        for row in result:
            table, col_name, comment = row['table'], row['name'], row['comment']
            if table not in column_comments:
                column_comments[table] = {}
            column_comments[table][col_name] = comment
        return column_comments
    
    def format_table_description(self, table_name: str, table_comment: str, columns_info: Dict[str, str]) -> str:
        """Format table description for the model"""
        description = f"Table: {table_name}\n"
        if table_comment:
            description += f"Description: {table_comment}\n"
        else:
            description += "Description: No description provided\n"

        if columns_info:
            # Add column descriptions
            description += "Columns:\n"
            for col_name, col_comment in columns_info.items():
                if col_comment:
                    description += f"  - {col_name}: {col_comment}\n"
                else:
                    description += f"  - {col_name}: No description provided\n"

        return description

TableMetadataManager 负责从系统表中提取表名、表注释及字段注释，并将这些元数据格式化为 LLM 易于理解的描述文本。

ResourceManager

class ResourceManager:
    """MCP resource manager"""

    def __init__(self, client: ClickHouseClient, logger: Logger
                 , resource_prefix: str = DEFAULT_RESOURCE_PREFIX
                 , results_limit: int = DEFAULT_RESULTS_LIMIT):
        self.client = client
        self.logger = logger
        self.metadata_manager = TableMetadataManager(client, logger)
        self.resource_prefix = resource_prefix
        self.results_limit = results_limit

    async def list_resources(self) -> List[Resource]:
        """List all resources in the database"""
        self.logger.debug("Listing resources")
        database = self.client.db_config.get("database")
        
        try:
            # Get table list
            table_list = self.metadata_manager.get_table_list(database)
            if not table_list:
                return []

            # Get table comments and column comments
            table_comments = self.metadata_manager.get_table_comments(database)
            column_comments = self.metadata_manager.get_column_comments(database)

            # Format table descriptions
            resources = []
            for table_name in table_list:
                table_comment = table_comments.get(table_name, "")
                columns_info = column_comments.get(table_name, {})
                description = self.metadata_manager.format_table_description(table_name, table_comment, columns_info)

                # Create resources
                resource = Resource(
                    uri=f"{self.resource_prefix}/{table_name}/data",
                    name=f"Table: {table_name}",
                    mimeType="text/plain",
                    description=description,
                    type="table",
                    metadata = {
                        "columns": [
                            {
                                "name": col_name,
                                "description": col_comment
                            }
                            for col_name, col_comment in columns_info.items()
                        ]
                    }
                )
                resources.append(resource)
            self.logger.debug(f"Found {len(resources)} resources")
            return resources
        except Exception as e:
            self.logger.error(f"Failed to list resources: {e}")
            return []

    async def read_resource(self, uri: AnyUrl) -> str:
        """Read resource data"""
        self.logger.debug(f"Reading resource: {uri}")
        uri_str = str(uri)

        try:
            # Parse URI
            if not uri_str.startswith(self.resource_prefix):
                self.logger.error(f"Invalid resource URI: {uri}")
                return ""

                # get talbe name
                table_name = uri_str[len(self.resource_prefix):].split("/")[0]

                # get query
                query = f"SELECT * FROM {quote_identifier(table_name)} LIMIT {self.results_limit}"
                result = self.client.execute_query(query)

                # format result
                if not result:
                    return "No data found"
                return json.dumps(result, default=str , indent=2)
        except Exception as e:
            self.logger.error(f"Failed to read resource: {e}")
            return f"Error reading resource: {str(e)}"

ResourceManager 将数据库中的每张表封装为一个 MCP 资源，并附带其结构描述，供 LLM 理解和访问。

ToolManager

class ToolManager:
    """MCP tool manager"""

    def __init__(self, client: ClickHouseClient, logger: Logger):
        self.client = client
        self.logger = logger

    async def list_tools(self) -> List[Tool]:
        """List all tools"""
        self.logger.debug("Listing tools")
        return [
            Tool(
                name="execute_sql",
                description="Execute a query against the ClickHouse database",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "The SQL query to be executed"
                        }
                    },
                    "required": ["query"],
                }
            )
        ]

    async def call_tool(self, name: str, arguments: Dict[str, Any]) -> List[TextContent]:
        """Call a tool"""
        self.logger.debug(f"Calling tool: {name} with arguments: {arguments}")

        # Tool handler mapping
        tool_handlers = {
            "execute_sql": self._handle_execute_sql
        }

        # Get handler
        handler = tool_handlers.get(name)
        if not handler:
            self.logger.error(f"Tool not found: {name}")
            return []

        # Call handler
        return await handler(arguments)

    async def _handle_execute_sql(self, arguments: Dict[str, str]) -> List[TextContent]:
        """Handle execute_sql tool"""
        self.logger.debug("Handling execute_sql tool")
        # Get query
        query = arguments.get("query")
        if not query:
            self.logger.error("Query is required")
            return []

        # Check query
        is_dangerous, pattern = dangerous_check(query)
        if is_dangerous:
            self.logger.error(f"Dangerous query detected: {pattern}")
            return [TextContent(value=f"Error: Dangerous query detected: {pattern}")]

        try:
            # Execute query
            result = self.client.execute_query(query)
            json_result = json.dumps(result, default=str, indent=2)
            return [
                TextContent(
                    type='text',
                    text=json_result,
                    mimeType='application/json'
                )
            ]
        except Exception as e:
            self.logger.error(f"Failed to execute query: {e}")
            return [TextContent(type='text', text=f"Error executing query: {str(e)}")]

ToolManager 定义了一个 execute_sql 工具，支持 LLM 提交 SQL 查询并获取 JSON 格式的结果，同时通过危险检测机制保障数据库安全。

DatabaseServer

class DatabaseServer:
    """MCP database server"""
    def __init__(self, config: Config, logger: Logger):
        self.app = Server("clickhouse_mcp_server")
        self.logger = logger

        # create components
        self.client = ClickHouseClient(config, logger)
        self.resource_manager = ResourceManager(self.client, logger)
        self.tool_manager = ToolManager(self.client, logger)

        # register components
        self.app.list_resources()(self.resource_manager.list_resources)
        self.app.read_resource()(self.resource_manager.read_resource)
        self.app.list_tools()(self.tool_manager.list_tools)
        self.app.call_tool()(self.tool_manager.call_tool)

    async def run(self):
        """Run the server"""
        from mcp.server.stdio import stdio_server
        
        self.logger.info("Starting server")
        async with stdio_server() as (read_stream, write_stream):
            try:
                await self.app.run(
                    read_stream, 
                    write_stream,
                    self.app.create_initialization_options()
                )
            except Exception as e:
                self.logger.error(f"Server error: {e}")
                raise

通过上述几个模块的组合，一套完整的 ClickHouse MCP 服务器就搭建完成了。实际使用时，只需在 VSCode 的 Cline 或类似的 MCP 客户端中配置好连接，即可直接使用自然语言进行查询和分析数据，体验远比手动编写 SQL 更加流畅高效。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：一步步从零开始自己动手实现ClickHouse MCP的完整教程要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题