Dify 中使用 MCP 协议访问 ClickHouse 搭建完整教程

# Open Source License

Dify is licensed under a modified version of the Apache License 2.0, with the following additional conditions:

1. Dify may be utilized commercially, including as a backend service for other applications or as an application development platform for enterprises. Should the conditions below be met, a commercial license must be obtained from the producer:

a. Multi-tenant service: Unless explicitly authorized by Dify in writing, you may not use the Dify source code to operate a multi-tenant environment.
    - Tenant Definition: Within the context of Dify, one tenant corresponds to one workspace. The workspace provides a separated area for each tenant's data and configurations.

b. LOGO and copyright information: In the process of using Dify's frontend, you may not remove or modify the LOGO or copyright information in the Dify console or applications. This restriction is inapplicable to uses of Dify that do not involve its frontend.
    - Frontend Definition: For the purposes of this license, the "frontend" of Dify includes all components located in the `web/` directory when running Dify from the raw source code, or the "web" image when running Dify with Docker.

2. As a contributor, you should agree that:

a. The producer can adjust the open-source agreement to be more strict or relaxed as deemed necessary.
b. Your contributed code may be used for commercial purposes, including but not limited to its cloud business operations.

Apart from the specific conditions mentioned above, all other rights and restrictions follow the Apache License 2.0. Detailed information about the Apache License 2.0 can be found at http://www.apache.org/licenses/LICENSE-2.0.

The interactive design of this product is protected by appearance patent.

© 2025 LangGenius, Inc.

# 开源许可证

Dify 是根据 Apache License 2.0 的修改版本授权的，并附带以下附加条件：

1. Dify 可用于商业用途，包括作为其他应用程序的后端服务或企业应用程序开发平台。若满足以下条件，则必须从生产商处获得商业许可：

a. 多租户服务：除非获得 Dify 的明确书面授权，否则您不得使用 Dify 源代码来运营多租户环境。
    - 租户定义：在Dify环境下，一个租户对应一个工作区，工作区为每个租户的数据和配置提供一个独立的区域。

b.LOGO及版权信息：在使用Dify前端的过程中，您不得删除或修改Dify控制台或应用程序中的LOGO或版权信息。此限制不适用于不涉及Dify前端的使用。
    - 前端定义：为了本许可的目的，Dify 的“前端”包括从原始源代码运行 Dify 时位于“web/”目录中的所有组件，或使用 Docker 运行 Dify 时的“web”图像。

请通过电子邮件联系business@dify.ai询问许可事宜。

2. 作为贡献者，您应当同意：

a. 生产者可以根据需要对开源协议进行调整，使其更加严格或者宽松。
b.您贡献的代码可能会被用于商业用途，包括但不限于其云业务运营。

除上述特定条件外，其他所有权利和限制均遵循 Apache License 2.0。有关 Apache License 2.0 的详细信息，请访问 http://www.apache.org/licenses/LICENSE-2.0。

该产品的交互设计受外观专利保护。

© 2025 LangGenius, Inc.

git clone [https://github.com/langgenius/dify.git](https://github.com/langgenius/dify.git) --branch 0.15.3

cd dify/docker
cp .env.example .env

# 启动 Docker 容器，根据你系统上的 Docker Compose 版本，选择合适的命令来启动容器。
docker-compose up -d

docker-compose -p dify up -d

docker-compose -p <项目名称> [其他命令]
例如：# 启动项目并指定名称
docker-compose -p myproject up -d

docker compose ps

# 本地环境
http://localhost/install

# 本地环境
http://localhost

http://localhost:8000/sse

{
  "server_name1": {
    "url": "http://127.0.0.1:8000/sse",
    "headers": {},
    "timeout": 60,
    "sse_read_timeout": 300
  },
  "server_name2": {
    "url": "http://127.0.0.1:8001/sse"
  }
}

CLICKHOUSE_HOST=localhost
CLICKHOUSE_PORT=8123
CLICKHOUSE_USER=default
CLICKHOUSE_PASSWORD=clickhouse

mcp dev mcp_clickhouse/mcp_server.py

mcp-proxy --sse-host=0.0.0.0 --sse-port=8080 --pass-environment -- mcp run mcp_clickhouse/mcp_server.py

你是一名专业的数据分析师和DBA。
理解用户的原始需求，用户需求是：{{#sys.query#}}
结合知识库中提供的数据表结构信息
输出标准的查询 ClickHouse SQL 语句，供 ClickHouse 直接执行。

要求：
1. 返回内容：仅返回标准查询 SQL 语句；
2. 返回内容：不要添加其他任何内容，不要添加格式内容，如：query、SQL 等；
3. 查询语句的表名前，需要添加数据库名称。

def main(arg1: str) -> dict:
    # 去掉开头的 ```sql
    if arg1.startswith("```sql"):
        arg1 = arg1[len("```sql"):]
    # 去掉结尾的 ```
    if arg1.endswith("```"):
        arg1 = arg1[:-len("```")]
    # 将所有的 \n 替换为空格
    arg1 = arg1.replace("\n", " ")
    # 去掉可能的前后空格
    arg1 = arg1.strip()
    return {
        "result": arg1
    }

import json

def main(arg1: str) -> dict:

    # 结果存储
    result = {
    "result-1": "",  # 存储 trip_mile_group 的值
    "result-2": ""   # 存储 vin_count 的值
    }

    # 解析文本内容
    text_content = arg1
    # 提取content部分
    content_start = text_content.find("content=[") + len("content=[")
    content_end = text_content.find("] isError=")
    content_str = text_content[content_start:content_end]

    # 分割content中的每个TextContent
    text_contents = content_str.split("), ")

    # 提取每个TextContent中的text字段并解析JSON
    field_values = {}
    field_names = []

    # 处理第一个JSON对象时提取字段名称
    first_json_processed = False

    for tc in text_contents:
        if tc.startswith("TextContent(type='text', text='"):
            # 提取JSON字符串
            json_str = tc[len("TextContent(type='text', text='"):]
            
            # 如果字符串以 ')结尾，去掉这部分
            if json_str.endswith("')"):
                json_str = json_str[:-2]
            elif not tc.endswith(")"): # 处理最后一个元素，它可能没有结尾的 )
                json_str = json_str
            
            # 处理转义字符
            json_str = json_str.encode().decode('unicode_escape')
            
            # 移除末尾的单引号（如果有）
            if json_str.endswith("'"):
                json_str = json_str[:-1]
            
            try:
                item = json.loads(json_str)
                
                # 如果是第一个JSON对象，提取字段名称
                if not first_json_processed:
                    field_names = list(item.keys())
                    # 初始化每个字段的值列表
                    for field in field_names:
                        field_values[field] = []
                    first_json_processed = True
                
                # 将每个字段的值添加到相应的列表中
                for field in field_names:
                    if field in item:
                        field_values[field].append(str(item[field]))
                    else:
                        field_values[field].append("")
                        
            except json.JSONDecodeError as e:
                print(f"Error parsing JSON: {e}")
                print(f"Problematic JSON string: {json_str}")

    # 将提取的值用分号连接并存入result
    if len(field_names) >= 2:
        result["result-1"] = ";".join(field_values[field_names[0]])
        result["result-2"] = ";".join(field_values[field_names[1]])
    else:
        print("Warning: Expected at least 2 fields in JSON objects, but found", len(field_names))

    return {
        "result_1": result["result-1"],
        "result_2": result["result-2"]
    }