ByteHouse MCP Server Skill
🔵 ByteHouse 品牌标识
「ByteHouse」—— 火山引擎云原生数据仓库，极速、稳定、安全、易用
本Skill基于ByteHouse官方MCP Server，提供完整的ByteHouse数据访问能力
描述
在本地拉起ByteHouse MCP Server并调用其tools的技能。
当以下情况时使用此 Skill:
(1) 需要连接ByteHouse数据库查询数据
(2) 需要使用MCP协议与ByteHouse交互
(3) 用户提到"ByteHouse"、"MCP"、"查询数据库"、"看表"
(4) 需要生成数据资产目录和血缘分析
📁 文件说明
- SKILL.md - 本文件，技能主文档
- mcp_client.py - MCP客户端模块，用于程序化调用MCP Server
- test_mcp_server.py - MCP Server测试脚本
- example_mcp_usage.py - MCP使用示例
- query_top10_tables_mcp.py - 使用MCP查询Top 10大表
- test_list_tables.py - 测试list_tables tool
- data_asset_analyzer.py - 数据资产和血缘分析工具（新增）
- start_mcp_service.sh - 启动常驻MCP Server服务
- stop_mcp_service.sh - 停止MCP Server服务
- status_mcp_service.sh - 查看MCP Server状态
- restart_mcp_service.sh - 重启MCP Server服务
前置条件
- Python 3.8+
- uv (已安装在
/root/.local/bin/uv)
- ByteHouse连接信息（需自行配置）
配置信息
ByteHouse连接配置
{
"host": "<ByteHouse-host>",
"port": "<ByteHouse-port>",
"user": "<ByteHouse-user>",
"password": "<ByteHouse-password>",
"secure": true,
"verify": true
}
环境变量
在使用前请设置以下环境变量：
export BYTEHOUSE_HOST="<ByteHouse-host>"
export BYTEHOUSE_PORT="<ByteHouse-port>"
export BYTEHOUSE_USER="<ByteHouse-user>"
export BYTEHOUSE_PASSWORD="<ByteHouse-password>"
export BYTEHOUSE_SECURE="true"
export BYTEHOUSE_VERIFY="true"
export BYTEHOUSE_CONNECT_TIMEOUT="30"
export BYTEHOUSE_SEND_RECEIVE_TIMEOUT="30"
🎯 ByteHouse MCP Server Tools
| 序号 | Tool名称 | 功能描述 |
|---|
| 1 | list_databases | 列出所有数据库 |
| 2 | list_tables | 列出指定数据库中的所有表 |
| 3 | run_select_query | 运行SELECT查询 |
| 4 | run_dml_ddl_query | 运行DML/DDL查询 |
| 5 | get_bytehouse_table_engine_doc | 获取ByteHouse表引擎文档 |
🚀 快速开始
方法1: 测试MCP Server（推荐先测试）
cd /root/.openclaw/workspace/skills/bytehouse-mcp
# 先设置环境变量，然后运行
uv run test_mcp_server.py
这会：
- 自动安装ByteHouse MCP Server
- 启动MCP Server
- 列出所有可用的tools
- 尝试调用第一个tool
方法2: 列出数据库中的表
cd /root/.openclaw/workspace/skills/bytehouse-mcp
# 先设置环境变量，然后运行
uv run test_list_tables.py
方法3: 使用MCP查询Top 10大表
cd /root/.openclaw/workspace/skills/bytehouse-mcp
# 先设置环境变量，然后运行
uv run query_top10_tables_mcp.py
方法4: 生成数据资产和血缘分析（新增）
cd /root/.openclaw/workspace/skills/bytehouse-mcp
# 先设置环境变量，然后运行
uv run data_asset_analyzer.py
这会：
- 获取数据库的完整schema
- 生成数据资产目录
- 生成血缘分析报告
- 保存JSON文件到
output/ 目录
输出内容包括：
- 数据库schema（所有表和字段）
- 数据资产目录（表统计、标签、引擎分布）
- 血缘分析（表关系、列相似性）
方法5: 启动常驻MCP Server服务
cd /root/.openclaw/workspace/skills/bytehouse-mcp
# 先在脚本中配置环境变量，然后运行
./start_mcp_service.sh
这会：
- 在后台启动MCP Server
- 保存PID到
mcp_server.pid
- 写入日志到
logs/mcp_server_*.log
方法6: 查看MCP Server状态
./status_mcp_service.sh
方法7: 停止MCP Server
./stop_mcp_service.sh
方法8: 重启MCP Server
./restart_mcp_service.sh
💻 数据资产和血缘分析（新增）
功能说明
data_asset_analyzer.py 提供以下功能：
-
完整Schema获取
- 获取指定数据库的所有表
- 获取每张表的所有字段
- 提取表引擎、注释等元数据
-
数据资产目录生成
- 表统计（总表数、总列数）
- 引擎分布统计
- 自动标签生成
- 表资产详情
-
血缘分析
- 表关系识别（Distributed → Local）
- 列相似性分析
- 关系可视化
使用示例
#!/usr/bin/env python3
# /// script
# dependencies = [
# "mcp>=1.0.0",
# ]
# ///
import asyncio
from data_asset_analyzer import DataAssetAnalyzer
async def main():
analyzer = DataAssetAnalyzer()
await analyzer.connect()
# 分析数据库
result = await analyzer.analyze_database("default")
# result 包含:
# - schema: 完整的数据库schema
# - catalog: 数据资产目录
# - lineage: 血缘分析
# - files: 生成的文件路径
asyncio.run(main())
输出文件
分析完成后会在 output/ 目录生成以下文件：
- schema_{database}_{timestamp}.json - 完整的数据库schema
- catalog_{database}_{timestamp}.json - 数据资产目录
- lineage_{database}_{timestamp}.json - 血缘分析报告
💻 程序化使用MCP Client
使用mcp_client.py模块
#!/usr/bin/env python3
# /// script
# dependencies = [
# "mcp>=1.0.0",
# ]
# ///
import asyncio
from mcp_client import ByteHouseMCPClient
async def main():
async with ByteHouseMCPClient() as client:
await client.connect()
# 1. 列出所有tools
tools = await client.list_tools()
print("可用的tools:", [t['name'] for t in tools])
# 2. 调用tool
# result = await client.call_tool("tool_name", {"param": "value"})
# print(result)
asyncio.run(main())
直接使用MCP SDK
#!/usr/bin/env python3
# /// script
# dependencies = [
# "mcp>=1.0.0",
# ]
# ///
import asyncio
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
# 设置环境变量（请自行配置）
env = os.environ.copy()
env.update({
'BYTEHOUSE_HOST': '<ByteHouse-host>',
'BYTEHOUSE_PORT': '<ByteHouse-port>',
'BYTEHOUSE_USER': '<ByteHouse-user>',
'BYTEHOUSE_PASSWORD': '<ByteHouse-password>',
'BYTEHOUSE_SECURE': 'true',
'BYTEHOUSE_VERIFY': 'true',
})
# MCP Server参数
server_params = StdioServerParameters(
command='/root/.local/bin/uvx',
args=[
'--from',
'git+https://github.com/volcengine/mcp-server@main#subdirectory=server/mcp_server_bytehouse',
'mcp_bytehouse',
'-t',
'stdio'
],
env=env
)
# 启动MCP Server并连接
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 列出tools
result = await session.list_tools()
print("Tools:", [t.name for t in result.tools])
# 调用tool
# call_result = await session.call_tool("tool_name", {"param": "value"})
asyncio.run(main())
🔧 服务管理
启动服务
# 先配置环境变量，然后运行
./start_mcp_service.sh
查看状态
./status_mcp_service.sh
查看日志
# 查看最新日志
tail -f logs/mcp_server_*.log
# 查看特定日志文件
tail -f logs/mcp_server_20260312_184500.log
停止服务
./stop_mcp_service.sh
重启服务
./restart_mcp_service.sh
💻 使用MCP Tools示例
示例1: 列出所有数据库
import asyncio
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
# 设置环境变量（请自行配置）
env = os.environ.copy()
env.update({
'BYTEHOUSE_HOST': '<ByteHouse-host>',
'BYTEHOUSE_PORT': '<ByteHouse-port>',
'BYTEHOUSE_USER': '<ByteHouse-user>',
'BYTEHOUSE_PASSWORD': '<ByteHouse-password>',
'BYTEHOUSE_SECURE': 'true',
'BYTEHOUSE_VERIFY': 'true',
})
server_params = StdioServerParameters(
command='/root/.local/bin/uvx',
args=[
'--from',
'git+https://github.com/volcengine/mcp-server@main#subdirectory=server/mcp_server_bytehouse',
'mcp_bytehouse',
'-t',
'stdio'
],
env=env
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 调用list_databases
result = await session.call_tool("list_databases", {})
for content in result.content:
if content.type == 'text':
print(content.text)
asyncio.run(main())
示例2: 列出数据库中的表
# 调用list_tables
result = await session.call_tool("list_tables", {"database": "default"})
示例3: 运行SELECT查询
# 调用run_select_query
sql = "SELECT * FROM default.conversation_feedback LIMIT 10"
result = await session.call_tool("run_select_query", {"query": sql})
示例4: 查询Top 10大表
# 查询Top 10大表
sql = """
SELECT
database,
table,
sum(bytes) as total_bytes,
sum(rows) as total_rows
FROM system.parts
WHERE active = 1
GROUP BY database, table
ORDER BY total_bytes DESC
LIMIT 10
"""
result = await session.call_tool("run_select_query", {"query": sql})
最后更新: 2026-03-12
