byted-bytehouse-mcp

ByteHouse MCP サーバーのスキル

🔵 ByteHouse ブランドロゴ

「ByteHouse」—— 火山引擎のクラウドネイティブデータウェアハウス。高速、安定、安全、使いやすい製品です。

このスキルはByteHouse公式MCP Serverに基づき、ByteHouseのデータにアクセスするための機能を一通り提供します。

---

説明

ローカルでByteHouse MCP Serverを起動し、そのツールを呼び出すためのスキルです。

以下の場合にこのスキルを使用します: (1) ByteHouseデータベースに接続してデータをクエリする必要がある場合 (2) MCPプロトコルを使用してByteHouseとやり取りする必要がある場合 (3) ユーザーが「ByteHouse」、「MCP」、「データベースをクエリ」、「テーブルを見る」と言及した場合 (4) データアセットカタログとリネージ分析を生成する必要がある場合

🎯 ByteHouse MCP サーバーのツール

番号	ツール名	機能の説明
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のテーブルエンジンに関するドキュメントを取得します

📁 ファイルの説明

• 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を使用して上位10件の大きなテーブルをクエリします
• test_list_tables.py - list_tablesツールをテストします
• 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サービスを再起動します

設定の説明

設定は ~/.bytehouse_config.json に保存されます。このファイルが存在し、空でない場合は、ファイル内の設定をそのまま使用します。ファイルが存在しない場合は、ユーザーにByteHouseの接続情報を提供してもらいます（ホストアドレスとパスワードの取得方法を説明したこのドキュメントもお客様に共有してください：https://www.volcengine.com/docs/6517/1121919?lang=zh）。ユーザーから情報を提供してもらった後、JSONファイルに保存し、接続情報を繰り返し求めないようにします。ユーザーがByteHouseクラスターを切り替える場合は、このファイルもあわせて変更します。

{
   "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"
}

BYTEHOUSE_HOST（ホストアドレス）とBYTEHOUSE_PASSWORD（パスワード）は、必ずユーザーが指定する必要があります。BH_ARK_API_KEYは任意の設定で、embeddingにのみ使用します。初回利用時は省略できます。その他の設定は固定です。

前提条件

• Python 3.8+
• uv（/root/.local/bin/uv にインストール済み）
• ByteHouseの接続情報（ご自身で設定する必要があります）

🚀 クイックスタート

方法1: MCP Serverをテストする（最初にテストすることを推奨）

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

以下を実行します：

1. ByteHouse MCP Serverを自動でインストールします
2. MCP Serverを起動します
3. 利用可能なすべてのツールを一覧表示します
4. 最初のツールの呼び出しを試みます

方法2: データベース内のテーブルを一覧表示する

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

方法3: MCPを使用して上位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

以下を実行します：

1. データベースの完全なスキーマを取得します
2. データアセットカタログを生成します
3. リネージ分析レポートを生成します
4. output/ ディレクトリにJSONファイルを保存します

出力内容：

• データベースのスキーマ（すべてのテーブルとフィールド）
• データアセットカタログ（テーブルの統計、タグ、エンジンの分布）
• リネージ分析（テーブル間の関係、列の類似性）

方法5: 常駐MCP Serverサービスを起動する

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

以下を実行します：

1. バックグラウンドでMCP Serverを起動します
2. PIDを mcp_server.pid に保存します
3. ログを 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 は、以下の機能を提供します：

1. 完全なスキーマの取得

   • 指定したデータベースのすべてのテーブルを取得します
   • 各テーブルのすべてのフィールドを取得します
   • テーブルエンジン、コメントなどのメタデータを抽出します

2. データアセットカタログの生成

   • テーブルの統計（テーブルの総数、列の総数）
   • エンジンの分布統計
   • タグの自動生成
   • テーブルアセットの詳細

3. リネージ分析

   • テーブル間の関係を識別します（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/ ディレクトリに以下のファイルが生成されます：

1. schema_{database}_{timestamp}.json - データベースの完全なスキーマ
2. catalog_{database}_{timestamp}.json - データアセットカタログ
3. lineage_{database}_{timestamp}.json - リネージ分析レポート

💻 MCP クライアントをプログラムから使用する

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ツールの使用例

例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: 上位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})

---