BigQuery MCP服务器

一个用于Google BigQuery的模型上下文协议(MCP)服务器,使AI助手能够查询和与BigQuery数据集交互。

  • by effcamp
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 effcamp/bigquery-mcp

使用说明

BigQuery MCP服务器

用于Google BigQuery的模型上下文协议(MCP)服务器,使AI助手能够查询BigQuery数据集并与之交互。

特性

  • 查询目录:预定义的、经过测试的SQL查询作为MCP资源公开(没有上下文窗口膨胀!)
  • 自定义SQL查询:在需要时执行参数化的BigQuery SQL查询
  • 数据集发现:列出项目中的数据集和表
  • 架构检查:获取具有字段类型和描述的表模式
  • 简单查询:在不编写SQL的情况下,使用WHERE子句和LIMIT查询表
  • 类型安全:使用Pydantic构建,用于强大的数据验证
  • 可配置的:基于YAML的配置,支持环境变量
  • 可扩展的:在不影响LLM性能的情况下添加数百个业务查询

安装

先决条件

  • Python 3.13+ (建议使用完整功能)
    • 对于Python 3.10+:大多数功能都有效,但类型提示使用 | 语法(要求3.10+)
  • 启用BigQuery的谷歌云平台帐户
  • GCP凭据(服务帐户或默认凭据)

方法1:使用pip(推荐)

从GCP工件注册表安装:

# Configure pip to use your GCP Artifact Registry
pip install bigquery-mcp \
  --index-url https://us-central1-python.pkg.dev/<your-project-id>/<your-repo>/simple/

或者配置一次身份验证:

# Authenticate with GCP Artifact Registry
gcloud auth application-default login
pip install keyring
pip config set global.index-url https://us-central1-python.pkg.dev/<your-project-id>/<your-repo>/simple/

# Install the package
pip install bigquery-mcp

# Run the server
bigquery-mcp

方法2:使用pipx(隔离安装)

对于不影响全局Python环境的隔离安装:

# Install pipx if not already installed
pip install pipx
pipx ensurepath

# Install bigquery-mcp from Artifact Registry
pipx install bigquery-mcp \
  --pip-args="--index-url https://us-central1-python.pkg.dev/<your-project-id>/<your-repo>/simple/"

# Run the server
bigquery-mcp

方法3:使用紫外线(显影)

对于本地来源的开发:

# Clone the repository
git clone <your-repo-url>
cd python-mcp

# Install dependencies
uv sync

# Run the server
uv run bigquery-mcp

我应该使用哪种方法?

方法 最佳 优点 缺点
点 生产,全系统安装 标准工具,简单 可能影响全球Python环境
pipx 用户安装,多个版本 每个应用程序的独立环境 需要额外的工具
紫外线 开发,贡献 快速,完整的开发环境 需要源代码

配置

选项1:配置文件(推荐)

创建 config/bigquery.yaml:

project_id: "your-gcp-project-id"
dataset_id: "your-default-dataset"  # optional
credentials_path: "/path/to/service-account.json"  # optional
location: "US"  # optional, default: US
max_results: 1000  # optional, default: 1000

选项2:环境变量

export GCP_PROJECT_ID="your-gcp-project-id"
export GCP_DATASET_ID="your-default-dataset"  # optional
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"  # optional
export GCP_LOCATION="US"  # optional
export GCP_MAX_RESULTS="1000"  # optional

选项3:自定义配置路径

export BIGQUERY_MCP_CONFIG="/custom/path/to/config.yaml"
uv run bigquery-mcp

认证

服务器支持多种身份验证方法(按优先级顺序):

  1. 服务帐户文件:在配置中指定或 GOOGLE_APPLICATION_CREDENTIALS
  2. 应用程序默认凭据:使用gcloud凭据

要设置gcloud凭据,请执行以下操作:

gcloud auth application-default login

查询目录

服务器包括 查询目录系统 它允许您定义LLM可以发现和执行的预先编写的SQL查询。这避免了LLM上下文窗口因数百个工具定义而膨胀。

运作原理

  1. 定义查询:在中创建YAML文件 config/queries/ 与您的业务查询
  2. 暴露为资源:查询会自动作为MCP资源公开
  3. LLM发现:LLM可以通过资源搜索和读取查询定义
  4. 按ID执行:使用 execute_catalog_query 运行查询的工具

益处

  • 无上下文框:在不影响LLM性能的情况下添加数百个查询
  • 自然语言:LLM按描述/标签将用户请求与目录查询相匹配
  • 一致的SQL:预先编写、经过测试的查询可确保可靠性
  • 易于维护:在不更改代码的情况下更新YAML文件中的查询

查询定义示例

queries:
  - id: account_balance_reconciliation
    name: Account Balance Reconciliation
    description: Compare account balances between internal and external systems
    category: reconciliation
    tags: [reconciliation, accounting, balance]
    parameters:
      - name: start_date
        type: DATE
        description: Start date (YYYY-MM-DD)
        required: true
      - name: end_date
        type: DATE
        description: End date (YYYY-MM-DD)
        required: true
    sql: |
      SELECT account_id, internal_balance, external_balance,
             (internal_balance - external_balance) as discrepancy
      FROM ...
      WHERE date BETWEEN @start_date AND @end_date

示例用例

所包含的示例查询演示了:

  • 账户余额对账:比较内部余额和外部余额
  • 钱包余额对账:验证钱包交易是否与余额匹配
  • 交易对账:将内部交易与支付处理器匹配
  • 每日交易摘要:每日收入和交易量指标
  • 客户终身价值:LTV计算和细分
  • 按类别划分的收入:月度细分和趋势

看 config/queries/README.md 有关创建自己的查询的完整文档。

可用工具

1.execute_catalog_query

从目录中执行预定义的查询。将其用于常见的业务操作。

参数:

  • query_id (字符串,必填):要执行的目录查询的ID
  • parameters (object,可选):将参数作为键值对进行查询
  • max_results (整数,可选):要返回的最大结果数

例子:

{
  "query_id": "account_balance_reconciliation",
  "parameters": {
    "start_date": "2024-01-01",
    "end_date": "2024-01-31"
  }
}

发现:

  • 使用MCP资源(list_resources / read_resource)发现可用查询
  • LLM自动将用户请求与目录查询相匹配

2.execute_query

使用可选参数执行自定义BigQuery SQL查询。

参数:

  • query (字符串,必填):要执行的SQL查询(使用 @param_name 参数)
  • parameters (object,可选):将参数作为键值对进行查询
  • max_results (整数,可选):要返回的最大结果数

例子:

{
  "query": "SELECT * FROM `project.dataset.table` WHERE name = @name AND age > @min_age",
  "parameters": {
    "name": "John",
    "min_age": 25
  },
  "max_results": 100
}

目录查询与自定义SQL

服务器为LLM提供了何时使用目录查询与自定义SQL的智能指导:

  • 更喜欢目录查询 用于常见的业务操作和报告
  • 使用自定义SQL 用于临时探索或独特的一次性查询
  • 资源公开目录 不会污染工具列表

3.列表_数据集

列出BigQuery项目中的所有数据集。

参数: 无

退货:

{
  "datasets": ["dataset1", "dataset2", "dataset3"]
}

4.列表表

列出特定数据集中的所有表。

参数:

  • dataset_id (字符串,必填):用于列出表的数据集ID

退货:

{
  "tables": ["table1", "table2", "table3"]
}

5.get_table_schema

获取特定表的架构。

参数:

  • dataset_id (字符串,必填):数据集ID
  • table_id (字符串,必填):表ID

退货:

{
  "schema": [
    {
      "name": "id",
      "type": "INTEGER",
      "mode": "REQUIRED",
      "description": "Unique identifier"
    },
    {
      "name": "name",
      "type": "STRING",
      "mode": "NULLABLE",
      "description": "User name"
    }
  ]
}

6.查询表

使用可选的WHERE子句和LIMIT查询表。

参数:

  • dataset_id (字符串,必填):数据集ID
  • table_id (字符串,必填):表ID
  • limit (整数,可选):要返回的最大行数
  • where (字符串,可选):WHERE子句(不带WHERE关键字)

例子:

{
  "dataset_id": "my_dataset",
  "table_id": "users",
  "limit": 10,
  "where": "age > 18 AND country = 'US'"
}

MCP资源

服务器将目录查询作为MCP资源公开:

  • URI格式: bigquery://queries/{query_id}
  • 发现:使用 list_resources 查看所有可用的目录查询
  • 详情:使用 read_resource 获取完整的查询详细信息,包括SQL、参数和描述
  • 无上下文框:资源不占用工具槽或上下文窗口空间

与MCP客户端一起使用

克劳德桌面

添加到您的Claude Desktop配置(~/Library/Application Support/Claude/claude_desktop_config.json 在macOS或 %APPDATA%\Claude\claude_desktop_config.json 在Windows上):

使用pip或pipx(推荐)

{
  "mcpServers": {
    "bigquery": {
      "command": "bigquery-mcp",
      "env": {
        "GCP_PROJECT_ID": "your-project-id",
        "GOOGLE_APPLICATION_CREDENTIALS": "/path/to/credentials.json"
      }
    }
  }
}

使用紫外线(显影)

{
  "mcpServers": {
    "bigquery": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/python-mcp",
        "run",
        "bigquery-mcp"
      ],
      "env": {
        "GCP_PROJECT_ID": "your-project-id",
        "GOOGLE_APPLICATION_CREDENTIALS": "/path/to/credentials.json"
      }
    }
  }
}

其他MCP客户端

服务器使用MCP协议通过stdio进行通信。任何兼容MCP的客户端都可以通过运行以下命令进行连接:

bigquery-mcp

发展

运行类型检查

uv run basedpyright src/

跑步林特尔

uv run ruff check src/

格式化代码

uv run ruff format src/

运行测试

uv run pytest

项目结构

python-mcp/
├── src/
│   └── bigquery_mcp/
│       ├── __init__.py
│       ├── server.py          # Main MCP server
│       ├── bigquery_client.py # BigQuery client wrapper
│       ├── tools.py           # Tool definitions
│       ├── models.py          # Pydantic models
│       └── config.py          # Configuration loader
├── config/
│   └── bigquery.yaml.example  # Example configuration
├── examples/
│   └── example_queries.sql    # Sample SQL queries
├── tests/                     # Test files
├── pyproject.toml            # Project configuration
└── README.md

建筑

服务器遵循MCP工具箱模式:

  1. 配置层 (config.py):从YAML或环境变量加载设置
  2. 客户端层 (bigquery_client.py):使用异步方法包装BigQuery API
  3. 工具层 (tools.py):定义MCP工具并处理参数验证
  4. 服务器层 (server.py):使用stdio传输的MCP服务器实现
  5. 模型层 (models.py):用于类型安全的Pydantic模型

与MCP工具箱的比较

此实现的灵感来自 googleapis/genai工具箱 但是:

  • 基于Python 而不是Go
  • BigQuery特定 而不是多数据库
  • 更简单的POC 专注于核心功能
  • 直接MCP集成 没有单独的控制平面

许可证

MIT许可证-请参阅 许可证 文件以获取详细信息。

贡献

欢迎投稿!请随时提交拉取请求。