MCP Bigquery

🚀 mcp-bigquery

mcp-bigquery 包為 BigQuery SQL 驗證、預運行分析和查詢結構分析提供了一個全面的 MCP 服務器。該服務器提供了五種工具，可在不執行查詢的情況下對 BigQuery SQL 查詢進行驗證、分析和理解。

🚀 快速開始

前提條件

• Python 3.10 及以上版本
• 啟用了 BigQuery API 的 Google Cloud SDK
• 配置好的應用默認憑據

安裝

從 PyPI 安裝（推薦）

# 從 PyPI 安裝
pip install mcp-bigquery

# 或者使用 uv
uv pip install mcp-bigquery

從源代碼安裝

# 克隆倉庫
git clone https://github.com/caron14/mcp-bigquery.git
cd mcp-bigquery

# 使用 uv 安裝（推薦）
uv pip install -e .

# 或者使用 pip 安裝
pip install -e .

身份驗證

設置應用默認憑據：

gcloud auth application-default login

或者使用服務賬號密鑰：

export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json

配置

環境變量

Claude Code 集成

添加到您的 Claude Code 配置中：

{
  "mcpServers": {
    "mcp-bigquery": {
      "command": "mcp-bigquery",
      "env": {
        "BQ_PROJECT": "your-gcp-project",
        "BQ_LOCATION": "asia-northeast1",
        "SAFE_PRICE_PER_TIB": "5.0"
      }
    }
  }
}

如果是從源代碼安裝的：

{
  "mcpServers": {
    "mcp-bigquery": {
      "command": "python",
      "args": ["-m", "mcp_bigquery"],
      "env": {
        "BQ_PROJECT": "your-gcp-project",
        "BQ_LOCATION": "asia-northeast1",
        "SAFE_PRICE_PER_TIB": "5.0"
      }
    }
  }
}

✨ 主要特性

• SQL 驗證：在不運行查詢的情況下檢查 BigQuery SQL 語法
• 預運行分析：獲取成本估算、引用的表和架構預覽
• 查詢結構分析：分析 SQL 複雜度、JOIN、CTE 和查詢模式
• 依賴提取：從查詢中提取表和列的依賴關係
• 增強的語法驗證：提供詳細的錯誤報告和建議
• 參數支持：驗證參數化查詢
• 成本估算：根據處理的字節數計算美元估算值

💻 使用示例

基礎用法

驗證簡單查詢

# 工具: bq_validate_sql
{
  "sql": "SELECT 1"
}
# 返回: {"isValid": true}

帶參數驗證

# 工具: bq_validate_sql
{
  "sql": "SELECT * FROM users WHERE name = @name AND age > @age",
  "params": {
    "name": "Alice",
    "age": 25
  }
}

獲取成本估算

# 工具: bq_dry_run_sql
{
  "sql": "SELECT * FROM `bigquery-public-data.samples.shakespeare`",
  "pricePerTiB": 5.0
}
# 返回處理的字節數、美元估算值和架構

分析複雜查詢

# 工具: bq_dry_run_sql
{
  "sql": """
    WITH user_stats AS (
      SELECT user_id, COUNT(*) as order_count
      FROM orders
      GROUP BY user_id
    )
    SELECT * FROM user_stats WHERE order_count > 10
  """
}

分析查詢結構

# 工具: bq_analyze_query_structure
{
  "sql": """
    WITH ranked_products AS (
      SELECT 
        p.name,
        p.price,
        ROW_NUMBER() OVER (PARTITION BY p.category ORDER BY p.price DESC) as rank
      FROM products p
      JOIN categories c ON p.category_id = c.id
    )
    SELECT * FROM ranked_products WHERE rank <= 3
  """
}
# 返回: 包含 CTE、窗口函數和 JOIN 的複雜查詢分析

提取查詢依賴關係

# 工具: bq_extract_dependencies
{
  "sql": "SELECT u.name, u.email, o.total FROM users u LEFT JOIN orders o ON u.id = o.user_id"
}
# 返回: 表 (users, orders) 和列 (name, email, total, id, user_id)

增強語法驗證

# 工具: bq_validate_query_syntax
{
  "sql": "SELECT * FROM users WHERE name = 'John' LIMIT 10"
}
# 返回: 包含性能警告和建議的驗證結果

驗證 BigQuery 特定語法

# 工具: bq_validate_query_syntax
{
  "sql": "SELECT ARRAY[1, 2, 3] as numbers, STRUCT('John' as name, 25 as age) as person"
}
# 返回: 識別 BigQuery ARRAY 和 STRUCT 語法的驗證結果

高級用法

本項目的高級用法主要體現在使用新添加的工具進行更全面的 SQL 分析，以下是相關示例：

# 工具: bq_analyze_query_structure
{
  "sql": "SELECT u.name, COUNT(*) FROM users u LEFT JOIN orders o ON u.id = o.user_id GROUP BY u.name",
  "params": {}  # 可選
}
# 該查詢包含 JOIN 和聚合操作，可通過此工具分析其複雜度、JOIN 類型等

# 工具: bq_extract_dependencies
{
  "sql": "SELECT u.name, u.email FROM users u WHERE u.created_at > '2023-01-01'",
  "params": {}  # 可選
}
# 此查詢可通過該工具提取表和列的依賴關係，並生成依賴圖

# 工具: bq_validate_query_syntax
{
  "sql": "SELECT * FROM users WHERE name = 'John' LIMIT 10",
  "params": {}  # 可選
}
# 該工具可對查詢進行增強的語法驗證，給出詳細的錯誤報告和建議

📚 詳細文檔

工具介紹

bq_validate_sql

在不執行查詢的情況下驗證 BigQuery SQL 語法。 輸入：

{
  "sql": "SELECT * FROM dataset.table WHERE id = @id",
  "params": {"id": "123"}  // 可選
}

成功響應：

{
  "isValid": true
}

錯誤響應：

{
  "isValid": false,
  "error": {
    "code": "INVALID_SQL",
    "message": "Syntax error at [3:15]",
    "location": {
      "line": 3,
      "column": 15
    },
    "details": [...]  // 可選
  }
}

bq_dry_run_sql

執行預運行以獲取成本估算和元數據，而不執行查詢。 輸入：

{
  "sql": "SELECT * FROM dataset.table",
  "params": {"id": "123"},  // 可選
  "pricePerTiB": 6.0  // 可選，覆蓋默認值
}

成功響應：

{
  "totalBytesProcessed": 1073741824,
  "usdEstimate": 0.005,
  "referencedTables": [
    {
      "project": "my-project",
      "dataset": "my_dataset",
      "table": "my_table"
    }
  ],
  "schemaPreview": [
    {
      "name": "id",
      "type": "STRING",
      "mode": "NULLABLE"
    },
    {
      "name": "created_at",
      "type": "TIMESTAMP",
      "mode": "REQUIRED"
    }
  ]
}

錯誤響應：

{
  "error": {
    "code": "INVALID_SQL",
    "message": "Table not found: dataset.table",
    "details": [...]  // 可選
  }
}

bq_analyze_query_structure

分析 BigQuery SQL 查詢結構和複雜度。 輸入：

{
  "sql": "SELECT u.name, COUNT(*) FROM users u LEFT JOIN orders o ON u.id = o.user_id GROUP BY u.name",
  "params": {}  // 可選
}

成功響應：

{
  "query_type": "SELECT",
  "has_joins": true,
  "has_subqueries": false,
  "has_cte": false,
  "has_aggregations": true,
  "has_window_functions": false,
  "has_union": false,
  "table_count": 2,
  "complexity_score": 15,
  "join_types": ["LEFT"],
  "functions_used": ["COUNT"]
}

bq_extract_dependencies

從 BigQuery SQL 中提取表和列的依賴關係。 輸入：

{
  "sql": "SELECT u.name, u.email FROM users u WHERE u.created_at > '2023-01-01'",
  "params": {}  // 可選
}

成功響應：

{
  "tables": [
    {
      "project": null,
      "dataset": "users",
      "table": "u",
      "full_name": "users.u"
    }
  ],
  "columns": ["created_at", "email", "name"],
  "dependency_graph": {
    "users.u": ["created_at", "email", "name"]
  },
  "table_count": 1,
  "column_count": 3
}

bq_，validate_query_syntax

增強的語法驗證，提供詳細的錯誤報告。 輸入：

{
  "sql": "SELECT * FROM users WHERE name = 'John' LIMIT 10",
  "params": {}  // 可選
}

成功響應：

{
  "is_valid": true,
  "issues": [
    {
      "type": "performance",
      "message": "SELECT * may impact performance - consider specifying columns",
      "severity": "warning"
    },
    {
      "type": "consistency",
      "message": "LIMIT without ORDER BY may return inconsistent results",
      "severity": "warning"
    }
  ],
  "suggestions": [
    "Specify exact columns needed instead of using SELECT *",
    "Add ORDER BY clause before LIMIT for consistent results"
  ],
  "bigquery_specific": {
    "uses_legacy_sql": false,
    "has_array_syntax": false,
    "has_struct_syntax": false
  }
}

🔧 技術細節

測試

使用 pytest 運行測試：

# 運行所有測試（需要 BigQuery 憑據）
pytest tests/

# 僅運行不需要憑據的測試
pytest tests/test_min.py::TestWithoutCredentials

開發

# 安裝開發依賴
uv pip install -e ".[dev]"

# 在本地運行服務器
python -m mcp_bigquery

# 或者使用控制檯腳本
mcp-bigquery

限制

• 無查詢執行：此服務器僅執行預運行和驗證
• 成本估算：美元估算值是基於處理的字節數的近似值
• 參數類型：初始實現將所有參數視為 STRING 類型
• 緩存禁用：查詢始終以 use_query_cache=False 運行，以獲得準確的估算值

📄 許可證

本項目採用 MIT 許可證。

📈 更新日誌

0.3.0 (2025-08-17)

• 新增工具：添加了三個新的 SQL 分析工具，用於全面的查詢分析
• bq_analyze_query_structure：分析 SQL 複雜度、JOIN、CTE、窗口函數，並計算複雜度得分
• bq_extract_dependencies：提取表和列的依賴關係，並進行依賴圖映射
• bq_validate_query_syntax：增強的語法驗證，提供詳細的錯誤報告和建議
• SQL 分析引擎：新的 SQLAnalyzer 類，具有全面的 BigQuery SQL 解析能力
• BigQuery 特定功能：檢測 ARRAY/STRUCT 語法、舊版 SQL 模式和 BigQuery 特定驗證
• 向後兼容性：所有現有工具（bq_validate_sql、bq_dry_run_sql）保持不變
• 增強文檔：更新了所有五個工具的綜合示例

0.2.1 (2025-08-16)

• 修復了 GitHub Pages 文檔佈局問題
• 增強了 MkDocs Material 主題兼容性
• 改進了文檔依賴項和構建過程
• 將 site/ 目錄添加到 .gitignore
• 簡化了文檔佈局以提高兼容性

0.2.0 (2025-08-16)

• 通過預提交鉤子提高了代碼質量
• 使用 Black、Ruff、isort 和 mypy 增強了開發設置
• 改進了 CI/CD 管道
• 增強了文檔

0.1.0 (2025-08-16)

• 初始版本
• 從 mcp-bigquery-dryrun 重命名為 mcp-bigquery
• SQL 驗證工具（bq_validate_sql）
• 預運行分析工具（bq_dry_run_sql）
• 基於處理字節數的成本估算
• 支持參數化查詢