BigQuery数据访问MCP服务器

Pulse后端MCP服务器

一个模型上下文协议（MCP）服务器，为我们公司的开发人员提供BigQuery访问和专门的数据工具，以提高生产力。

概述

此MCP服务器实现 模型上下文协议 以使LLM驱动的应用程序能够以受控的方式访问公司数据并执行专门的数据功能。服务器公开了几个用于与BigQuery和公司特定数据结构交互的工具。

项目结构

pulse-backend-mcp/
├── README.md                 # Project documentation
├── pyproject.toml            # Python project configuration
├── uv.lock                   # Dependency lock file
├── .env                      # Environment variables (create this file)
└── src/                      # Source code directory
    └── server.py             # Main MCP server implementation

什么是MCP？

模型上下文协议（MCP）是一种开放协议，它规范了应用程序如何向LLM提供上下文。与USB-C提供将设备连接到外围设备的标准化方式类似，MCP提供了将AI模型连接到不同数据源和工具的标准化方法。

MCP架构

MCP遵循客户端-服务器架构：

• MCP主机：希望通过MCP访问数据的程序，如Claude Desktop、IDE或AI工具
• MCP客户端：与服务器保持1:1连接的协议客户端
• MCP服务器：通过标准化协议公开特定功能的轻量级程序（如本程序）
• 数据源：MCP服务器可以安全访问的数据库、文件或服务

通信流

1. 主机应用程序（例如Claude Desktop）初始化与我们的MCP服务器的连接
2. 客户端通过以下方式发现可用工具 tools/list 端点
3. 当用户提示时，LLM可以使用我们的工具执行BigQuery查询或检索客户端信息
4. 我们的服务器执行请求的操作并将结果返回给客户端
5. 客户端在主机应用程序中向用户呈现结果

主要特点

• BigQuery集成：对公司BigQuery数据集执行SQL查询
• 客户端数据访问：从我们的数据仓库中检索客户详细信息和数据集
• 可扩展架构：添加新工具以支持其他用例

先决条件

• Python 3.13或更高版本
• 具有BigQuery访问权限的Google Cloud帐户
• 具有适当权限的服务帐户凭据
• 单击API键（用于任务集成）

安装

1. 克隆存储库：

   复制

   git clone https://github.com/yourusername/saras-mcp.git
   cd saras-mcp
   

2. 创建虚拟环境：

   复制

   python -m venv venv
   source venv/bin/activate  # On Windows, use: venv\Scripts\activate
   

3. 安装依赖项：

   复制

   pip install -r requirements.txt
   

配置

1. 通过以下任一方式设置Google Cloud凭据：

   • 设置 GOOGLE_APPLICATION_CREDENTIALS 指向服务帐户密钥文件的环境变量：
     复制

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

   • 调用工具时将服务帐户路径直接传递给工具

2. 创建一个 .env 根目录中的文件，包含以下变量：

   复制

   GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
   GOOGLE_PROJECT_ID=your-project-id
   CLICKUP_API_KEY=your-clickup-api-key
   

3. （可选）如果需要，调整工具定义中的默认项目ID

用法

在检查器中启动MCP服务器：

mcp dev src/server.py

服务器将在默认的MCP端口（通常为8080）上启动。现在，您可以将兼容MCP的客户端连接到此服务器。

MCP检验员测试

要测试服务器实现，请执行以下操作：

1. 安装 MCP检查员
2. 连接到正在运行的服务器
3. 探索可用工具并测试其功能

可用工具

1.execute_bigquery

执行BigQuery SQL查询，并将结果作为结构化数据接收。

参数：

• query （string，必填）：要执行的SQL查询
• project_id （字符串，可选）：谷歌云项目ID（默认值：“insightsprod”）
• service_account_path （字符串，可选）：服务帐户JSON凭据的路径

工具注释：

• 只读：是（不修改数据）
• 开放世界：是（与外部BigQuery服务交互）

2.获取客户端详细信息

从我们的数据仓库中检索客户信息。

参数：

• client_id （字符串，可选）：要筛选的特定客户端ID
• client_name （字符串，可选）：要搜索的客户端名称（支持部分匹配）
• project_id （字符串，可选）：谷歌云项目ID（默认值：“insightsprod”）
• service_account_path （字符串，可选）：服务帐户JSON凭据的路径

工具注释：

• 只读：是（不修改数据）
• 开放世界：否（在内部数据仓库上操作）

3.获取客户端数据集

检索特定客户端的可用数据集。

参数：

• client_id （字符串，可选）：要筛选的特定客户端ID
• client_name （字符串，可选）：要搜索的客户端名称（支持部分匹配）
• project_id （字符串，可选）：谷歌云项目ID（默认值：“insightsprod”）
• service_account_path （字符串，可选）：服务帐户JSON凭据的路径

工具注释：

• 只读：是（不修改数据）
• 开放世界：否（在内部数据仓库上操作）

4.get_dataset_tables

列出特定BigQuery数据集中的所有表及其元数据。

参数：

• dataset_id （string，必填）：用于列出表的BigQuery数据集的ID
• project_id （字符串，可选）：谷歌云项目ID（默认：来自环境）
• service_account_path （字符串，可选）：服务帐户JSON凭据的路径

工具注释：

• 只读：是（不修改数据）
• 开放世界：否（在内部数据仓库上操作）

5.获取链接任务

检索特定ClickUp任务的详细信息。

参数：

• task_id （string，必填）：ClickUp任务的唯一标识符
• api_key （字符串，可选）：用于身份验证的ClickUp API密钥（默认：来自环境）
• include_subtasks （布尔值，可选）：是否包含子任务信息
• include_comments （布尔值，可选）：是否包含任务注释

工具注释：

• 只读：是（不修改数据）
• 开放世界：是（与外部ClickUp API交互）

扩展服务器

添加新工具

要向MCP服务器添加新工具，请执行以下操作：

1. 添加新功能 server.py 装饰有 @mcp.tool()
2. 定义函数的参数和返回类型
3. 添加全面的文档字符串来记录工具的目的和使用方法
4. 实施错误处理以获得稳健的用户体验

例子：

@mcp.tool()
def my_new_tool(param1: str, param2: int = 0) -> dict:
    """Description of what the tool does.

    Args:
        param1: Description of param1
        param2: (Optional) Description of param2

    Returns:
        Dictionary containing the results or error information
    """
    try:
        # Implementation
        return {"success": True, "results": [...]}
    except Exception as e:
        return {
            "success": False, 
            "error": "Error Type", 
            "message": str(e),
            "code": 500
        }

正确的错误处理

对于可能遇到错误的工具：

1. 使用适当的错误结构
2. 如果可能，返回特定的错误代码
3. 提供有意义的错误消息

例子：

try:
    # Tool operation
    result = perform_operation()
    return {"success": True, "results": result}
except NotFound as e:
    return {
        "success": False,
        "error": "Not Found",
        "message": str(e),
        "code": 404,
    }
except Exception as e:
    return {
        "success": False,
        "error": "Execution Error",
        "message": str(e),
        "code": 500,
    }

工具注释

在定义工具时，考虑添加注释以帮助客户理解工具的行为：

• readOnlyHint：指示工具是否修改其环境
• destructiveHint：指示工具是否可以执行破坏性操作
• idempotentHint：指示具有相同参数的重复调用是否没有额外效果
• openWorldHint：指示工具是否与外部实体交互

安全考虑

在开发MCP服务器时，请遵循以下安全最佳实践：

1. 输入验证

   • 根据模式验证所有参数
   • 对SQL查询进行消毒以防止注入攻击
   • 检查参数大小和范围

2. 访问控制

   • 必要时实施适当的身份验证
   • 使用适当的授权访问敏感数据
   • 考虑对资源密集型操作进行速率限制

3. 错误处理

   • 不要向客户暴露内部错误
   • 记录与安全相关的错误
   • 出错后适当清理资源

MCP协议资源

• MCP简介
• 核心架构
• 资源
• 工具
• MCP服务器开发人员指南

贡献

1. 为您的功能或错误修复创建新分支
2. 为您的更改添加适当的测试
3. 提交一个拉取请求，明确描述更改

许可证

[您的许可证在这里]