BigQuery数据分析服务

BigQuery FastMCP服务器

基于FastMCP的BigQuery服务器实现，通过专门的AI代理提供智能数据发现和分析功能。此软件包包括独立的FastMCP服务器功能和ADK Web界面兼容性。

特性

• 多代理架构：数据发现和分析的专业代理
• FastMCP集成：用于可扩展web应用程序的现代HTTP/SSE传输
• ADK Web兼容性：与Google ADK Web界面无缝协作
• BigQuery操作：完全支持查询、模式发现和数据集管理
• 示例数据创建：用于创建测试数据集和样本数据的内置工具

建筑

代理系统

该软件包实现了一个多代理编排系统：

1. 编排器代理：将用户请求路由到适当的专业代理
2. 数据发现代理：处理模式探索、数据编目和结构分析
3. 数据分析代理：执行统计分析、商业智能和见解生成

运输选项

• FastMCP服务器 (server.py)：独立web应用程序的HTTP/SSE传输
• ADK兼容代理 (agent.py)：用于ADK Web界面集成的基于stdio的传输

安装

1. 克隆存储库并安装依赖项：

pip install fastmcp google-cloud-bigquery python-dotenv google-adk

2. 在中设置环境变量 .env 文件：

BIGQUERY_PROJECT=your-project-id
BIGQUERY_LOCATION=your-location  # e.g., asia-south1, US
BIGQUERY_KEY_FILE=/path/to/service-account-key.json  # Optional

3. 配置BigQuery身份验证：
   • 选项1：使用服务帐户密钥文件（设置 BIGQUERY_KEY_FILE)
   • 选项2：使用应用程序默认凭据（ADC）
   • 选项3：使用gcloud身份验证

用法

ADK Web界面（推荐）

对于与Google ADK Web界面一起使用，代理会自动配置：

from bigquery_fastmcp import agent

# The root_agent is ready to use with ADK Web Interface
# It automatically handles routing between discovery and analytics agents

ADK代理提供：

• 专用代理之间的智能请求路由
• 全面的BigQuery操作
• Web优化的性能和错误处理

独立FastMCP服务器

对于独立的web应用程序或直接HTTP/SSE访问：

# Start the FastMCP server
python bigquery_fastmcp/server.py --project YOUR_PROJECT --location YOUR_LOCATION --port 8001

# Server runs on http://127.0.0.1:8001 by default
# SSE endpoint available at http://127.0.0.1:8001/sse/

服务器选项：

python server.py --help

optional arguments:
  --project PROJECT     BigQuery project ID
  --location LOCATION   BigQuery location (default: US)
  --key-file KEY_FILE   Path to service account key file
  --host HOST           Host to run server on (default: localhost)
  --port PORT           Port to run server on (default: 8001)

代理能力

数据发现代理

专长于：

• 数据目录管理：对现有数据集和表格的系统探索
• 模式分析：深入了解表结构、列类型和约束
• 数据概况：数据分布分析和质量评估
• 关系发现：查找表之间的连接
• 元数据抽取：全面记录数据资产

查询示例：

• “项目中有哪些表？”
• “描述客户表的模式”
• “显示销售数据集的结构”

数据分析代理

专长于：

• 统计分析：综合统计分析和分布
• 商业智能：KPI计算和业务指标
• 趋势分析：模式识别和异常检测
• 比较分析：分段和期间比较
• 数据聚合：有意义的决策摘要

查询示例：

• “分析上一季度的销售趋势”
• “按地区划分的平均订单价值是多少？”
• “比较不同群体之间的用户参与度”

配置

环境变量

BigQuery身份验证

服务器支持多种身份验证方法：

1. 服务帐户密钥文件 （推荐用于生产）：

   复制

   BIGQUERY_KEY_FILE=/path/to/service-account-key.json
   

2. 应用程序默认凭据:

   复制

   gcloud auth application-default login
   

3. 计算引擎/云外壳：自动使用附加的服务帐户

示例数据创建

服务器包括用于创建测试样本数据集的实用程序：

# Create a complete sample environment
create_complete_sample("test_dataset", "asia-south1")

这将创建：

• 一个新的BigQuery数据集
• 样品 departments 和 employees 表格
• 拥有10个部门和50名员工

日志记录

FastMCP服务器同时记录到stdout和文件：

• 日志文件: mcp_bigquery_fastmcp_server.log
• 日志级别：调试（可配置）
• 日志格式：时间戳、记录器名称、级别、消息

错误处理

该软件包包括以下全面的错误处理：

• BigQuery身份验证失败
• 无效查询和格式错误的SQL
• 网络连接问题
• 配置缺失或无效
• 表/数据集访问权限

发展

项目结构

bigquery_fastmcp/
├── __init__.py          # Package initialization
├── agent.py             # ADK Web Interface compatible agent
├── server.py            # FastMCP HTTP/SSE server
├── config.py            # Configuration management
└── README.md            # This file

扩展服务器

要添加新工具，请修改 server.py:

@mcp.tool()
def your_new_tool(param: str) -> str:
    """Description of your new tool"""
    # Implementation here
    return result

故障排除

常见问题

1. 身份验证错误:

   • 验证服务帐户密钥文件路径
   • 检查服务帐户是否具有BigQuery权限
   • 尝试 gcloud auth application-default login

2. 连接问题:

   • 验证项目ID是否正确
   • 检查与BigQuery的网络连接
   • 确保位置/地区有效

3. 权限错误:

   • 验证服务帐户是否具有所需的BigQuery角色：
     • BigQuery Data Editor
     • BigQuery Job User
     • BigQuery Data Viewer

调试

通过设置日志级别启用详细日志记录：

import logging
logging.getLogger('mcp_bigquery_fastmcp_server').setLevel(logging.DEBUG)

许可证

该项目根据MIT许可证获得许可。

贡献

1. 复刻仓库
2. 创建要素分支
3. 进行更改
4. 如果适用，添加测试
5. 提交拉取请求

支持

对于问题和疑问：

• 检查上面的故障排除部分
• 查看BigQuery文档
• 在存储库中提交问题