Pulse Backend MCP Server

Pulse Backend MCP Server

一个 Model Context Protocol（MCP）服务，为我们公司的开发者提供 BigQuery 访问能力和专用数据工具，以提升工作效率。

概述

此 MCP 服务实现了 Model Context Protocol，使大模型驱动的应用能够以受控方式访问公司数据并执行专用数据功能。该服务提供多种工具，用于与 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？

Model Context Protocol（MCP）是一种开放协议，用于标准化应用向大模型提供上下文的方式。正如 USB-C 提供了连接设备与外围设备的标准化方式，MCP 也提供了连接 AI 模型与不同数据源和工具的标准化方式。

MCP 架构

MCP 采用客户端-服务端架构：

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

通信流程

1. 主机应用（例如 Claude Desktop）与我们的 MCP 服务建立连接
2. 客户端通过 tools/list 端点发现可用工具
3. 用户发出提示后，大模型可以使用我们的工具执行 BigQuery 查询或检索客户端信息
4. 我们的服务执行所请求的操作，并将结果返回给客户端
5. 客户端在主机应用中向用户呈现结果

主要功能

• BigQuery 集成：针对公司的 BigQuery 数据集执行 SQL 查询
• 客户端数据访问：从我们的数据仓库检索客户端详情和数据集
• 可扩展架构：添加新工具以支持更多使用场景

前置条件

• Python 3.13 或更高版本
• 可访问 BigQuery 的 Google Cloud 账号
• 具有适当权限的服务账号凭据
• ClickUp 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

使用方法

在 Inspector 中启动 MCP 服务：

mcp dev src/server.py

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

使用 MCP Inspector 测试

测试服务实现：

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

可用工具

1. execute_bigquery

执行 BigQuery SQL 查询，并以结构化数据的形式接收结果。

参数：

• query（字符串，必填）：要执行的 SQL 查询
• project_id（字符串，可选）：Google Cloud 项目 ID（默认值："insightsprod"）
• service_account_path（字符串，可选）：服务账号 JSON 凭据的路径

工具注解：

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

2. get_client_details

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

参数：

• client_id（string，可选）：要筛选的特定客户端 ID
• client_name（string，可选）：要搜索的客户端名称（支持部分匹配）
• project_id（string，可选）：Google Cloud 项目 ID（默认值："insightsprod"）
• service_account_path（string，可选）：服务账号 JSON 凭据的路径

工具注释：

• 只读：是（不会修改数据）
• 开放世界：否（在内部数据仓库中运行）

3. get_client_datasets

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

参数：

• client_id（string，可选）：要筛选的特定客户端 ID
• client_name（string，可选）：要搜索的客户端名称（支持部分匹配）
• project_id（string，可选）：Google Cloud 项目 ID（默认值："insightsprod"）
• service_account_path（string，可选）：服务账号 JSON 凭据的路径

工具注释：

• 只读：是（不会修改数据）
• 开放世界：否（在内部数据仓库中运行）

4. get_dataset_tables

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

参数：

• dataset_id（string，必填）：要列出表的 BigQuery 数据集 ID
• project_id（string，可选）：Google Cloud 项目 ID（默认值：来自环境）
• service_account_path（string，可选）：服务账号 JSON 凭据的路径

工具注释：

• 只读：是（不会修改数据）
• 开放世界：否（在内部数据仓库中运行）

5. get_clickup_task

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

参数：

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

工具注释：

• 只读：是（不会修改数据）
• 开放世界：是（与外部 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. 提交拉取请求，并清楚描述所做的更改

许可证

[你的许可证]