手把手教程：在Dify中通過(guò)MCP連接ClickHouse，解鎖AI數(shù)據(jù)分析新能力 原創(chuàng)
在本教程中，我將詳細(xì)介紹如何在 Dify 平臺(tái)中通過(guò) MCP 協(xié)議連接 ClickHouse 數(shù)據(jù)庫(kù)，實(shí)現(xiàn)數(shù)據(jù)分析和可視化功能。這個(gè)集成方案可以幫助開(kāi)發(fā)者在 AI 應(yīng)用中輕松訪問(wèn)和分析大規(guī)模數(shù)據(jù)，提升 LLM 應(yīng)用的數(shù)據(jù)處理能力。
Dify 環(huán)境配置
Dify 安裝
本文采用 Dify 社區(qū)版的 Docker Compose 部署方式。版本演進(jìn)較快，可能與當(dāng)前記錄方式有差異。以官方手冊(cè)為準(zhǔn)。
Docker Compose 部署： https://docs.dify.ai/zh-hans/getting-started/install-self-hosted/docker-compose
安裝 Dify 之前, 請(qǐng)確保你的機(jī)器已滿足最低安裝要求：CPU >= 2 Core & RAM >= 4 GiB
（1）克隆 Dify 源代碼至本地環(huán)境。
git clone [https://github.com/langgenius/dify.git](https://github.com/langgenius/dify.git) --branch 0.15.3（2）啟動(dòng) Dify。
cd dify/docker
cp .env.example .env
# 啟動(dòng) Docker 容器，根據(jù)你系統(tǒng)上的 Docker Compose 版本，選擇合適的命令來(lái)啟動(dòng)容器。
docker-compose up -d注意：由于 Dify 的 Docker 容器內(nèi)容較多，如果像本文一樣，搭建 RagFlow 后又搭建 Dify，很容易出現(xiàn)沖突情況，需要再 Docker 容器上加上命名空間。
docker-compose -p dify up -d??docker-compose -p?? 命令用于在 Docker Compose 中指定項(xiàng)目的名稱，覆蓋默認(rèn)的目錄名作為項(xiàng)目名，基本用法。
docker-compose -p <項(xiàng)目名稱> [其他命令]
例如：# 啟動(dòng)項(xiàng)目并指定名稱
docker-compose -p myproject up -d（3）配置 Dify
如果需要 Dify 啟動(dòng)內(nèi)容，.env 還需要看一下的，里面可以配置的內(nèi)容還是比較豐富的。
（4）檢查安裝
檢查是否所有容器都正常運(yùn)行：
docker compose ps如果有問(wèn)題，需要單獨(dú)查看報(bào)錯(cuò)情況，目前兩次安裝的不同版本都可以直接啟動(dòng)成功。
（5）訪問(wèn)系統(tǒng)
# 本地環(huán)境
http://localhost/install
# 本地環(huán)境
http://localhost默認(rèn)是 80 端口，如果端口沖突，需要配置端口號(hào)，在 .env 配置；
模型配置
大模型使用的是遠(yuǎn)端模型，測(cè)試的 LLM 為：
- 阿里云百煉：https://bailian.console.aliyun.com/
- DeepSeek 開(kāi)放平臺(tái)：https://platform.deepseek.com/
申請(qǐng)過(guò)程和本文主要內(nèi)容關(guān)聯(lián)不大，不進(jìn)行展開(kāi)。
MCP 配置
本文在 Dify 中連接 MCP Server，使用的 MCP 工具是 Dify 市場(chǎng)的：MCP SSE 插件。
Dify市場(chǎng)中的MCP SSE插件界面
MCP SSE：通過(guò) HTTP with SSE 傳輸使用 MCP 協(xié)議來(lái)發(fā)現(xiàn)和調(diào)用工具。
插件主頁(yè)：https://marketplace.dify.ai/plugins/junjiem/mcp_sse
MCP SSE插件詳細(xì)配置頁(yè)面
插件支持 sse 方式訪問(wèn) MCP Server，不支持本地方式。
http://localhost:8000/sseMCP服務(wù)配置，支持多個(gè)MCP服務(wù)。例如：
{
"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"
}
}版本演進(jìn)較快，如果有版本差異，可以查看新版本的使用說(shuō)明。
ClickHouse MCP
ClickHouse 官方 Github 提供了一個(gè) MCP Server 版本：ClickHouse MCP Server
地址：https://github.com/ClickHouse/mcp-clickhouse
（1）工具支持
1.run_select_query：在您的Clickhouse群集上執(zhí)行SQL查詢。使用說(shuō)明：
a.輸入：sql（字符串）：執(zhí)行的SQL查詢。
b.所有ClickHouse查詢均使用readonly = 1運(yùn)行，以確保它們安全。
2.list_databases：在您的Clickhouse群集上列出所有數(shù)據(jù)庫(kù)。
3.list_tables：在數(shù)據(jù)庫(kù)中列出所有表。使用說(shuō)明
4.輸入：database（字符串）：數(shù)據(jù)庫(kù)的名稱。
（2）配置說(shuō)明
將以下變量添加到存儲(chǔ)庫(kù)根的??.env??文件中。
CLICKHOUSE_HOST=localhost
CLICKHOUSE_PORT=8123
CLICKHOUSE_USER=default
CLICKHOUSE_PASSWORD=clickhouse（3）安裝依賴
運(yùn)行??uv sync??以安裝依賴項(xiàng)（Python 相關(guān)環(huán)境），值得注意的是，Python 版本是：3.13
MCP服務(wù)配置示例和多服務(wù)器設(shè)置界面
版本低的話，需要升級(jí) Python 版本，最新版本是 3.13.2，
注意：3.13 的先行版本會(huì)報(bào)錯(cuò)，如果使用是 pyenv 或者 brew 安裝，同樣需要升級(jí)到較新版本。
（4）運(yùn)行服務(wù)
項(xiàng)目根目錄使用如下命令運(yùn)行服務(wù)
mcp dev mcp_clickhouse/mcp_server.py其中：mcp 需要 node 環(huán)境。
啟動(dòng)后，安裝命令行中提示的地址進(jìn)行連通性驗(yàn)證，如果 list_databases 工具可用，證明連通性已經(jīng)可以。
MCP Proxy
由于 ClickHouse MCP Server 當(dāng)前所用版本（20250406）無(wú) SSE 協(xié)議支持，所以只能使用代理進(jìn)行協(xié)議轉(zhuǎn)換。
- 代理工具：mcp-proxy
- 項(xiàng)目介紹：Connect to MCP servers that run on SSE transport, or expose stdio servers as an SSE server using the MCP Proxy server.
- 項(xiàng)目地址：https://github.com/sparfenyuk/mcp-proxy
工具支持兩種模式：
- stdio to SSE
- SSE to stdio
本文使用的是：SSE to stdio，原理示意圖如下：
ClickHouse數(shù)據(jù)庫(kù)連接配置界面。
安裝環(huán)境后，通過(guò)命令啟動(dòng)。
mcp-proxy --sse-host=0.0.0.0 --sse-port=8080 --pass-environment -- mcp run mcp_clickhouse/mcp_server.pyDify 通過(guò) UI 進(jìn)行配置。
ClickHouse查詢示例和數(shù)據(jù)結(jié)構(gòu)展示
RAG 配置
為了讓大模型更好的理解 SQL，可以將 ClickHouse 的建表語(yǔ)句導(dǎo)入到了知識(shí)庫(kù)中。
ClickHouse與MCP集成的數(shù)據(jù)流程圖
官方手冊(cè)地址：https://docs.dify.ai/zh-hans/guides/knowledge-base
由于知識(shí)庫(kù)支持的格式為：
- 長(zhǎng)文本內(nèi)容（TXT、Markdown、DOCX、HTML、JSON 甚至是 PDF）
- 結(jié)構(gòu)化數(shù)據(jù)（CSV、Excel 等）
- 在線數(shù)據(jù)源（網(wǎng)頁(yè)爬蟲(chóng)、Notion 等）
導(dǎo)入 ClickHouse DDL 時(shí)，可以使用 TXT 格式。
ECharts 圖表插件
ECharts圖表生成是一個(gè)用于生成可視化ECharts圖表的工具，你可以通過(guò)它來(lái)生成柱狀圖、折線圖、餅圖等各類圖表。
項(xiàng)目地址：https://marketplace.dify.ai/plugins/langgenius/echarts
工作流配置
工作流基本配置
Dify 工作流分為兩種類型：
- Chatflow：面向?qū)υ掝惽榫?，包括客戶服?wù)、語(yǔ)義搜索、以及其他需要在構(gòu)建響應(yīng)時(shí)進(jìn)行多步邏輯的對(duì)話式應(yīng)用程序。
- Workflow：面向自動(dòng)化和批處理情景，適合高質(zhì)量翻譯、數(shù)據(jù)分析、內(nèi)容生成、電子郵件自動(dòng)化等應(yīng)用程序。
本文使用的方式是 Workflow，界面截圖參考官網(wǎng)如下：
Dify應(yīng)用中集成ClickHouse的實(shí)際效果展示
工作流配置文檔較多，使用方式也較為靈活，如下是本文在配置過(guò)程中參考的一些文檔，可以對(duì)工作流有大體的了解。
如果需要對(duì)工作流中的每一個(gè)組件有進(jìn)一步理解，可以參考官方手冊(cè)。
- 工作流官方手冊(cè)： https://docs.dify.ai/zh-hans/guides/workflow
本文使用的 SQL Prompt 是
你是一名專業(yè)的數(shù)據(jù)分析師和DBA。
理解用戶的原始需求，用戶需求是：{{#sys.query#}}
結(jié)合知識(shí)庫(kù)中提供的數(shù)據(jù)表結(jié)構(gòu)信息
輸出標(biāo)準(zhǔn)的查詢 ClickHouse SQL 語(yǔ)句，供 ClickHouse 直接執(zhí)行。
要求：
1. 返回內(nèi)容：僅返回標(biāo)準(zhǔn)查詢 SQL 語(yǔ)句；
2. 返回內(nèi)容：不要添加其他任何內(nèi)容，不要添加格式內(nèi)容，如：query、SQL 等；
3. 查詢語(yǔ)句的表名前，需要添加數(shù)據(jù)庫(kù)名稱。協(xié)議兼容
在 Dify 和 ClickHouse 進(jìn)行互通時(shí)，經(jīng)常出現(xiàn)格式不兼容的情況?？梢允褂靡韵鹿ぞ叽a進(jìn)行格式轉(zhuǎn)換：
（1）工具代碼：去掉 Markdown SQL 格式
去掉大模型返回的 markdown。
def main(arg1: str) -> dict:
# 去掉開(kāi)頭的 ```sql
if arg1.startswith("```sql"):
arg1 = arg1[len("```sql"):]
# 去掉結(jié)尾的 ```
if arg1.endswith("```"):
arg1 = arg1[:-len("```")]
# 將所有的 \n 替換為空格
arg1 = arg1.replace("\n", " ")
# 去掉可能的前后空格
arg1 = arg1.strip()
return {
"result": arg1
}（2）工具代碼：JSON 轉(zhuǎn) ECharts
ClickHouse MCP 返回格式為 JSON 格式，如果想給 ECharts 插件使用，同樣需要進(jìn)行轉(zhuǎn)換。本文試用了使用大模型進(jìn)行格式轉(zhuǎn)換，效果不佳，速度還慢，固定格式還是代碼來(lái)的快。
import json
def main(arg1: str) -> dict:
# 結(jié)果存儲(chǔ)
result = {
"result-1": "", # 存儲(chǔ) trip_mile_group 的值
"result-2": "" # 存儲(chǔ) vin_count 的值
}
# 解析文本內(nèi)容
text_content = arg1
# 提取content部分
content_start = text_content.find("cnotallow=[") + len("cnotallow=[")
content_end = text_content.find("] isError=")
content_str = text_content[content_start:content_end]
# 分割content中的每個(gè)TextContent
text_contents = content_str.split("), ")
# 提取每個(gè)TextContent中的text字段并解析JSON
field_values = {}
field_names = []
# 處理第一個(gè)JSON對(duì)象時(shí)提取字段名稱
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='"):]
# 如果字符串以 ')結(jié)尾，去掉這部分
if json_str.endswith("')"):
json_str = json_str[:-2]
elifnot tc.endswith(")"): # 處理最后一個(gè)元素，它可能沒(méi)有結(jié)尾的 )
json_str = json_str
# 處理轉(zhuǎn)義字符
json_str = json_str.encode().decode('unicode_escape')
# 移除末尾的單引號(hào)（如果有）
if json_str.endswith("'"):
json_str = json_str[:-1]
try:
item = json.loads(json_str)
# 如果是第一個(gè)JSON對(duì)象，提取字段名稱
ifnot first_json_processed:
field_names = list(item.keys())
# 初始化每個(gè)字段的值列表
for field in field_names:
field_values[field] = []
first_json_processed = True
# 將每個(gè)字段的值添加到相應(yīng)的列表中
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}")
# 將提取的值用分號(hào)連接并存入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"]
}其他工作流編排內(nèi)容，和業(yè)務(wù)緊密關(guān)聯(lián)，可以使用工作流的提供的預(yù)覽與測(cè)試工具進(jìn)行能力驗(yàn)證。
應(yīng)用發(fā)布
調(diào)試完成之后點(diǎn)擊右上角的發(fā)布，可以將該工作流保存并快速發(fā)布成為不同類型的應(yīng)用。
數(shù)據(jù)可視化結(jié)果展示-ECharts圖表
效果驗(yàn)證
本文使用 Dify 的對(duì)話框，通過(guò)自然語(yǔ)言的方式，輸入用戶需求，大模型理解后，輸出 ClickHouse 的 SQL 語(yǔ)句，Dify 通過(guò) MCP 協(xié)議訪問(wèn) ClickHouse 并得到查詢結(jié)果，根據(jù)結(jié)果生成圖表和數(shù)據(jù)情況分析。
數(shù)據(jù)分析結(jié)果和統(tǒng)計(jì)信息展示
本文轉(zhuǎn)載自??AI 博物院?? 作者：longyunfeigu
