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）
• 基于处理字节数的成本估算
• 支持参数化查询