byted-bytehouse-mcp

ByteHouse MCP Server Skill

🔵 ByteHouse brand logo

"ByteHouse" — Volcano Engine cloud-native data warehouse: extremely fast, stable, secure, and easy to use

This Skill is based on the official ByteHouse MCP Server and provides complete ByteHouse data access capabilities

Description

A skill for launching the ByteHouse MCP Server locally and calling its tools.

Use this Skill when the following situations occur: (1) Need to connect to the ByteHouse database to query data (2) Need to use the MCP protocol to interact with ByteHouse (3) The user mentions "ByteHouse", "MCP", "query database", "view tables" (4) Need to generate data asset catalogs and lineage analysis

🎯 ByteHouse MCP Server Tools

No.	Tool name	Function description
1	list_databases	List all databases
2	list_tables	List all tables in the specified database
3	run_select_query	Run a SELECT query
4	run_dml_ddl_query	Run a DML/DDL query
5	get_bytehouse_table_engine_doc	Get the ByteHouse table engine documentation

📁 File description

SKILL.md - This file, the main skill document
mcp_client.py - MCP client module, used to programmatically call the MCP Server
test_mcp_server.py - MCP Server test script
example_mcp_usage.py - MCP usage example
query_top10_tables_mcp.py - Use MCP to query the Top 10 largest tables
test_list_tables.py - Test the list_tables tool
data_asset_analyzer.py - Data asset and lineage analysis tool (new)
start_mcp_service.sh - Start the resident MCP Server service
stop_mcp_service.sh - Stop the MCP Server service
status_mcp_service.sh - View MCP Server status
restart_mcp_service.sh - Restart the MCP Server service

Configuration instructions

The configuration is saved in ~/.bytehouse_config.json . If the file exists and is not empty, use the configuration in the file directly. If it does not exist, ask the user to provide the ByteHouse connection information (also send this document to the customer; the document explains how to obtain the host address and password:https://www.volcengine.com/docs/6517/1121919?lang=zh ). After the user provides the information, save it to the json file to avoid repeatedly requesting connection information from the user. When the user switches ByteHouse clusters, modify this file as well.

{
   "BYTEHOUSE_HOST": "<ByteHouse-host>",
   "BYTEHOUSE_PORT": "8123",
   "BYTEHOUSE_USER": "bytehouse",
   "BYTEHOUSE_PASSWORD": "<ByteHouse-password>",
   "BYTEHOUSE_SECURE": true,
   "BYTEHOUSE_VERIFY": true, 
   "BH_ARK_API_KEY": "<火山引擎方舟API密钥>",
   "BH_ARK_BASE_URL": "https://ark.cn-beijing.volces.com/api/v3",
   "BH_EMBEDDING_MODEL": "doubao-embedding-vision-251215"
}

Among them, BYTEHOUSE_HOST (host address) and BYTEHOUSE_PASSWORD (password)must be provided bythe user. BH_ARK_API_KEY is an optional configuration, used only for embedding, and can be ignored by the user on first use. The remaining configurations are fixed.

Prerequisites

Python 3.8+
uv (already installed at /root/.local/bin/uv)
ByteHouse connection information (must be configured by yourself)

🚀 Quick start

Method 1: Test the MCP Server (testing first is recommended)

# 从配置文件读取配置，导出到环境变量
source scripts/export_config.sh
# 测试mcp server
uv run scripts/test_mcp_server.py

This will:

Automatically install the ByteHouse MCP Server
Start the MCP Server
List all available tools
Try to call the first tool

Method 2: List tables in the database

cd /root/.openclaw/workspace/skills/bytehouse-mcp
# 先设置环境变量，然后运行
uv run test_list_tables.py

Method 3: Use MCP to query the Top 10 largest tables

cd /root/.openclaw/workspace/skills/bytehouse-mcp
# 先设置环境变量，然后运行
uv run query_top10_tables_mcp.py

Method 4: Generate data asset and lineage analysis (new)

cd /root/.openclaw/workspace/skills/bytehouse-mcp
# 先设置环境变量，然后运行
uv run data_asset_analyzer.py

This will:

Obtain the complete schema of the database
Generate a data asset catalog
Generate a lineage analysis report
Save the JSON file to output/ Table of Contents

Output includes:

Database schema (all tables and fields)
Data asset catalog (table statistics, tags, engine distribution)
Lineage analysis (table relationships, column similarity)

Method 5: Start the resident MCP Server service

cd /root/.openclaw/workspace/skills/bytehouse-mcp
# 先在脚本中配置环境变量，然后运行
./start_mcp_service.sh

This will:

Start the MCP Server in the background
Save the PID to mcp_server.pid
Write logs to logs/mcp_server_*.log

Method 6: Check MCP Server status

./status_mcp_service.sh

Method 7: Stop MCP Server

./stop_mcp_service.sh

Method 8: Restart MCP Server

./restart_mcp_service.sh

💻 Data assets and lineage analysis (new)

Feature description

data_asset_analyzer.py Provides the following features:

Full schema retrieval
- Retrieve all tables of a specified database
- Retrieve all fields of each table
- Extract metadata such as table engine and comments
Data asset catalog generation
- Table statistics (total tables, total columns)
- Engine distribution statistics
- Automatic tag generation
- Table asset details
Lineage analysis
- Table relationship identification (Distributed → Local)
- Column similarity analysis
- Relationship visualization

Usage example

#!/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 files

After analysis completes, the following files will be generated in the output/ directory:

schema_{database}_{timestamp}.json - Complete database schema
catalog_{database}_{timestamp}.json - Data asset catalog
lineage_{database}_{timestamp}.json - Lineage analysis report

💻 Programmatic use of MCP Client

Use the mcp_client.py module

#!/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())

Use the MCP SDK directly

#!/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())

🔧 Service management

Start service

# 先配置环境变量，然后运行
./start_mcp_service.sh

Check status

./status_mcp_service.sh

View logs

# 查看最新日志
tail -f logs/mcp_server_*.log

# 查看特定日志文件
tail -f logs/mcp_server_20260312_184500.log

Stop service

./stop_mcp_service.sh

Restart service

./restart_mcp_service.sh

💻 Examples of using MCP Tools

Example 1: List all databases

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())

Example 2: List tables in a database

# 调用list_tables
result = await session.call_tool("list_tables", {"database": "default"})

Example 3: Run a SELECT query

# 调用run_select_query
sql = "SELECT * FROM default.conversation_feedback LIMIT 10"
result = await session.call_tool("run_select_query", {"query": sql})

Example 4: Query the Top 10 largest tables

# 查询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})

About this skill

🔵 ByteHouse brand logo

Description

🎯 ByteHouse MCP Server Tools

📁 File description

Configuration instructions

Prerequisites

🚀 Quick start

Method 1: Test the MCP Server (testing first is recommended)

Method 2: List tables in the database

Method 3: Use MCP to query the Top 10 largest tables

Method 4: Generate data asset and lineage analysis (new)

Method 5: Start the resident MCP Server service

Method 6: Check MCP Server status

Method 7: Stop MCP Server

Method 8: Restart MCP Server

💻 Data assets and lineage analysis (new)

Feature description

Usage example

Output files

💻 Programmatic use of MCP Client

Use the mcp_client.py module

Use the MCP SDK directly

🔧 Service management

Start service

Check status

View logs

Stop service

Restart service

💻 Examples of using MCP Tools

Example 1: List all databases

Example 2: List tables in a database

Example 3: Run a SELECT query

Example 4: Query the Top 10 largest tables