手把手教程:在Dify中通過MCP連接ClickHouse,解鎖AI數據分析新能力 原創
在本教程中,我將詳細介紹如何在 Dify 平臺中通過 MCP 協議連接 ClickHouse 數據庫,實現數據分析和可視化功能。這個集成方案可以幫助開發者在 AI 應用中輕松訪問和分析大規模數據,提升 LLM 應用的數據處理能力。
Dify 環境配置
Dify 安裝
本文采用 Dify 社區版的 Docker Compose 部署方式。版本演進較快,可能與當前記錄方式有差異。以官方手冊為準。
Docker Compose 部署: https://docs.dify.ai/zh-hans/getting-started/install-self-hosted/docker-compose
安裝 Dify 之前, 請確保你的機器已滿足最低安裝要求:CPU >= 2 Core & RAM >= 4 GiB
(1)克隆 Dify 源代碼至本地環境。
git clone [https://github.com/langgenius/dify.git](https://github.com/langgenius/dify.git) --branch 0.15.3(2)啟動 Dify。
cd dify/docker
cp .env.example .env
# 啟動 Docker 容器,根據你系統上的 Docker Compose 版本,選擇合適的命令來啟動容器。
docker-compose up -d注意:由于 Dify 的 Docker 容器內容較多,如果像本文一樣,搭建 RagFlow 后又搭建 Dify,很容易出現沖突情況,需要再 Docker 容器上加上命名空間。
docker-compose -p dify up -d??docker-compose -p?? 命令用于在 Docker Compose 中指定項目的名稱,覆蓋默認的目錄名作為項目名,基本用法。
docker-compose -p <項目名稱> [其他命令]
例如:# 啟動項目并指定名稱
docker-compose -p myproject up -d(3)配置 Dify
如果需要 Dify 啟動內容,.env 還需要看一下的,里面可以配置的內容還是比較豐富的。
(4)檢查安裝
檢查是否所有容器都正常運行:
docker compose ps如果有問題,需要單獨查看報錯情況,目前兩次安裝的不同版本都可以直接啟動成功。
(5)訪問系統
# 本地環境
http://localhost/install
# 本地環境
http://localhost默認是 80 端口,如果端口沖突,需要配置端口號,在 .env 配置;
模型配置
大模型使用的是遠端模型,測試的 LLM 為:
- 阿里云百煉:https://bailian.console.aliyun.com/
- DeepSeek 開放平臺:https://platform.deepseek.com/
申請過程和本文主要內容關聯不大,不進行展開。
MCP 配置
本文在 Dify 中連接 MCP Server,使用的 MCP 工具是 Dify 市場的:MCP SSE 插件。

Dify市場中的MCP SSE插件界面
MCP SSE:通過 HTTP with SSE 傳輸使用 MCP 協議來發現和調用工具。
插件主頁:https://marketplace.dify.ai/plugins/junjiem/mcp_sse

MCP SSE插件詳細配置頁面
插件支持 sse 方式訪問 MCP Server,不支持本地方式。
http://localhost:8000/sseMCP服務配置,支持多個MCP服務。例如:
{
"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 MCP
ClickHouse 官方 Github 提供了一個 MCP Server 版本:ClickHouse MCP Server
地址:https://github.com/ClickHouse/mcp-clickhouse
(1)工具支持
1.run_select_query:在您的Clickhouse群集上執行SQL查詢。使用說明:
a.輸入:sql(字符串):執行的SQL查詢。
b.所有ClickHouse查詢均使用readonly = 1運行,以確保它們安全。
2.list_databases:在您的Clickhouse群集上列出所有數據庫。
3.list_tables:在數據庫中列出所有表。使用說明
4.輸入:database(字符串):數據庫的名稱。
(2)配置說明
將以下變量添加到存儲庫根的??.env??文件中。
CLICKHOUSE_HOST=localhost
CLICKHOUSE_PORT=8123
CLICKHOUSE_USER=default
CLICKHOUSE_PASSWORD=clickhouse(3)安裝依賴
運行??uv sync??以安裝依賴項(Python 相關環境),值得注意的是,Python 版本是:3.13

MCP服務配置示例和多服務器設置界面
版本低的話,需要升級 Python 版本,最新版本是 3.13.2,
注意:3.13 的先行版本會報錯,如果使用是 pyenv 或者 brew 安裝,同樣需要升級到較新版本。
(4)運行服務
項目根目錄使用如下命令運行服務
mcp dev mcp_clickhouse/mcp_server.py其中:mcp 需要 node 環境。
啟動后,安裝命令行中提示的地址進行連通性驗證,如果 list_databases 工具可用,證明連通性已經可以。
MCP Proxy
由于 ClickHouse MCP Server 當前所用版本(20250406)無 SSE 協議支持,所以只能使用代理進行協議轉換。
- 代理工具:mcp-proxy
- 項目介紹:Connect to MCP servers that run on SSE transport, or expose stdio servers as an SSE server using the MCP Proxy server.
- 項目地址:https://github.com/sparfenyuk/mcp-proxy
工具支持兩種模式:
- stdio to SSE
- SSE to stdio
本文使用的是:SSE to stdio,原理示意圖如下:
ClickHouse數據庫連接配置界面。
安裝環境后,通過命令啟動。
mcp-proxy --sse-host=0.0.0.0 --sse-port=8080 --pass-environment -- mcp run mcp_clickhouse/mcp_server.pyDify 通過 UI 進行配置。

ClickHouse查詢示例和數據結構展示
RAG 配置
為了讓大模型更好的理解 SQL,可以將 ClickHouse 的建表語句導入到了知識庫中。

ClickHouse與MCP集成的數據流程圖
官方手冊地址:https://docs.dify.ai/zh-hans/guides/knowledge-base
由于知識庫支持的格式為:
- 長文本內容(TXT、Markdown、DOCX、HTML、JSON 甚至是 PDF)
- 結構化數據(CSV、Excel 等)
- 在線數據源(網頁爬蟲、Notion 等)
導入 ClickHouse DDL 時,可以使用 TXT 格式。
ECharts 圖表插件
ECharts圖表生成是一個用于生成可視化ECharts圖表的工具,你可以通過它來生成柱狀圖、折線圖、餅圖等各類圖表。
項目地址:https://marketplace.dify.ai/plugins/langgenius/echarts
工作流配置
工作流基本配置
Dify 工作流分為兩種類型:
- Chatflow:面向對話類情景,包括客戶服務、語義搜索、以及其他需要在構建響應時進行多步邏輯的對話式應用程序。
- Workflow:面向自動化和批處理情景,適合高質量翻譯、數據分析、內容生成、電子郵件自動化等應用程序。
本文使用的方式是 Workflow,界面截圖參考官網如下:

Dify應用中集成ClickHouse的實際效果展示
工作流配置文檔較多,使用方式也較為靈活,如下是本文在配置過程中參考的一些文檔,可以對工作流有大體的了解。
如果需要對工作流中的每一個組件有進一步理解,可以參考官方手冊。
- 工作流官方手冊: https://docs.dify.ai/zh-hans/guides/workflow
本文使用的 SQL Prompt 是
你是一名專業的數據分析師和DBA。
理解用戶的原始需求,用戶需求是:{{#sys.query#}}
結合知識庫中提供的數據表結構信息
輸出標準的查詢 ClickHouse SQL 語句,供 ClickHouse 直接執行。
要求:
1. 返回內容:僅返回標準查詢 SQL 語句;
2. 返回內容:不要添加其他任何內容,不要添加格式內容,如:query、SQL 等;
3. 查詢語句的表名前,需要添加數據庫名稱。協議兼容
在 Dify 和 ClickHouse 進行互通時,經常出現格式不兼容的情況。可以使用以下工具代碼進行格式轉換:
(1)工具代碼:去掉 Markdown SQL 格式
去掉大模型返回的 markdown。
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
}(2)工具代碼:JSON 轉 ECharts
ClickHouse MCP 返回格式為 JSON 格式,如果想給 ECharts 插件使用,同樣需要進行轉換。本文試用了使用大模型進行格式轉換,效果不佳,速度還慢,固定格式還是代碼來的快。
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("cnotallow=[") + len("cnotallow=[")
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]
elifnot 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對象,提取字段名稱
ifnot 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"]
}其他工作流編排內容,和業務緊密關聯,可以使用工作流的提供的預覽與測試工具進行能力驗證。
應用發布
調試完成之后點擊右上角的發布,可以將該工作流保存并快速發布成為不同類型的應用。

數據可視化結果展示-ECharts圖表
效果驗證
本文使用 Dify 的對話框,通過自然語言的方式,輸入用戶需求,大模型理解后,輸出 ClickHouse 的 SQL 語句,Dify 通過 MCP 協議訪問 ClickHouse 并得到查詢結果,根據結果生成圖表和數據情況分析。

數據分析結果和統計信息展示
本文轉載自??AI 博物院?? 作者:longyunfeigu

















