Data Product Mcp

Data Product MCP

一个用于在 Data Mesh Manager 中发现数据产品并申请访问权限，以及在数据平台上执行查询以访问业务数据的 Model Context Protocol（MCP）服务。

概念

理念：让 AI Agent 能够查找并访问任意数据产品，以获取业务语义上下文，同时落实数据治理政策。

或者，如果你更喜欢：

让 AI 能够回答任何业务问题。

数据产品 是由数据契约定义、经过管理的高质量业务数据集，可在组织内部与其他团队共享。 数据契约描述数据的结构、语义、质量和使用条款。数据产品为 AI 提供理解数据所需的语义上下文，不仅让 AI 知道有哪些数据，还能理解数据的含义以及如何正确使用。 我们使用 Data Mesh Manager 作为数据产品市场，搜索可用的数据产品，并通过分析其元数据来判断这些数据产品是否与任务相关。

确定数据产品后，数据治理对于确保数据产品访问受到控制、查询符合数据契约的使用条款，以及遵守组织的全局政策至关重要。如有必要，AI Agent 可以申请访问数据产品的输出端口，这可能需要数据产品所有者手动批准。

最后，大模型可以根据数据契约中的数据模型描述和语义生成 SQL 查询。执行 SQL 查询时，会启用安全防护措施，确保不会滥用敏感数据，并缓解提示词注入等攻击途径。查询结果会返回给 AI Agent，由其用于回答最初的业务问题。

步骤：

1. 发现： 在数据产品市场中查找与任务相关的数据产品
2. 治理： 检查并申请访问数据产品
3. 查询： 使用特定于平台的 MCP 服务执行 SQL 语句。

Data Mesh Manager 是集中的数据产品市场和治理层，为组织中的所有数据产品提供元数据、访问控制和数据契约。

数据平台（Snowflake、Databricks 等）托管实际数据并执行查询。MCP 服务连接到这些平台，对你有权访问的数据产品执行 SQL 查询。

工具

1. dataproduct_search

   • 根据搜索词搜索数据产品。使用多种搜索方式（列表、语义搜索）以获得全面结果。仅返回启用的数据产品。
   • 可选输入：
     • search_term（string）：用于筛选数据产品的搜索词。会在 ID、标题和描述中搜索。支持多个搜索词，以空格分隔。
   • 返回：包含数据产品 ID、名称和描述、所有者信息以及结果来源的结构化列表。

2. dataproduct_get

   • 根据 ID 获取数据产品。数据产品包含其所有输出端口和服务信息。响应会包含每个输出端口的访问状态，并内嵌任何数据契约。
   • 必需输入：
     • data_product_id（string）：数据产品 ID。
   • 返回：数据产品详情及增强后的输出端口，包括访问状态和内嵌的数据契约。

3. dataproduct_request_access

   • 请求访问数据产品的特定输出端口。此操作会创建访问请求。根据数据产品配置、用途和数据治理规则，访问权限会自动授予，或由数据产品所有者审核。
   • 必需输入：
     • data_product_id（string）：数据产品 ID。
     • output_port_id（string）：输出端口 ID。
     • purpose（string）：用户使用数据的具体目的及需要访问权限的原因。如果访问请求需要数据所有者批准，数据所有者会根据该用途，从业务、技术和治理角度判断是否符合访问条件。
   • 返回：访问请求详情，包括 access_id、状态和审批信息。

4. dataproduct_query

   • 对数据产品的输出端口执行 SQL 查询。此工具会连接到底层数据平台并执行提供的 SQL 查询。执行查询前，必须拥有该输出端口的有效访问权限。
   • 必需输入：
     • data_product_id（string）：数据产品 ID。
     • output_port_id（string）：输出端口 ID。
     • query（string）：要执行的 SQL 查询。
   • 返回：以结构化数据形式返回查询结果（最多 100 行）

安装

必须安装 uv。

然后将此条目添加到 MCP 客户端配置中：

{
  "mcpServers": {
    "dataproduct": {
      "command": "uvx",
      "args": [
        "dataproduct_mcp"
      ],
      "env": {
        "DATAMESH_MANAGER_API_KEY": "dmm_live_user_...",
        "DATAMESH_MANAGER_HOST": "https://api.datamesh-manager.com",
        "QUERY_ACCESS_EVALUATION_ENABLED": "true",
        "SNOWFLAKE_USER": "",
        "SNOWFLAKE_PASSWORD": "",
        "SNOWFLAKE_ROLE": "",
        "SNOWFLAKE_WAREHOUSE": "COMPUTE_WH",
        "DATABRICKS_HOST": "adb-xxx.azuredatabricks.net",
        "DATABRICKS_HTTP_PATH": "/sql/1.0/warehouses/xxx",
        "DATABRICKS_CLIENT_ID": "",
        "DATABRICKS_CLIENT_SECRET": "",
        "BIGQUERY_CREDENTIALS_PATH": "/path/to/service-account-key.json"
      }
    }
  }
}

以下是 Claude Desktop（~/Library/Application Support/Claude/claude_desktop_config.json）的配置格式，其他 MCP 客户端也有类似的配置选项。

配置

Data Mesh Manager 配置

环境变量	描述	必需	默认值
DATAMESH_MANAGER_API_KEY	用于身份验证的 API 密钥	是	N/A
DATAMESH_MANAGER_HOST	可自托管实例的基础 URL	否	https://api.datamesh-manager.com
QUERY_ACCESS_EVALUATION_ENABLED	启用／禁用基于 AI 的查询访问评估	否	true

要通过 Data Mesh Manager 进行身份验证，需将 DATAMESH_MANAGER_API_KEY 变量设置为你的 API 密钥。

如何在 Data Mesh Manager 中创建 API 密钥。

对于可自托管的 Data Mesh Manager 实例，请将 DATAMESH_MANAGER_HOST 环境变量设置为你的实例 URL。

如果 Data Mesh Manager 实例未启用 AI 功能，请将 QUERY_ACCESS_EVALUATION_ENABLED 设置为 false，以禁用基于 AI 的查询访问评估。

（没错，我们计划在未来实现基于 OAuth2 的身份验证，以简化此流程。）

Snowflake

如果你使用 Snowflake 作为数据平台，请为你的用户创建一个程序化访问令牌。如果 AI Agent 并非代表真实用户操作，请在 Snowflake 中创建一个新用户，为 AI Agent 创建一个新的服务用户，并授予该用户访问数据产品所需的权限。

你可能还需要配置网络策略，以启用程序化访问令牌。

用户需要：

• 对你要使用的仓库拥有 USAGE 权限。
• 分配一个角色（例如 DATAPRODUCT_MCP），该角色对你要访问的数据产品所在的数据库和架构拥有 USAGE 权限。

你可以使用 Snowflake Connector，在 Data Mesh Manager 中批准访问请求后自动授予对 Snowflake 中数据的访问权限。

环境变量	描述
DATACONTRACT_SNOWFLAKE_USERNAME	你的用户名
DATACONTRACT_SNOWFLAKE_PASSWORD	你的程序化访问令牌
DATACONTRACT_SNOWFLAKE_WAREHOUSE	你要使用的仓库，例如 COMPUTE_WH。
DATACONTRACT_SNOWFLAKE_ROLE	分配的用户角色，例如 DATAPRODUCT_MCP

Databricks

如果你使用 Databricks 作为数据平台，需要创建一个服务主体，并为其分配访问数据产品所需的权限。为该服务主体创建 OAuth2 客户端 ID 和密钥。

你可以使用 Databricks Connector，在 Data Mesh Manager 中批准访问请求后自动授予对 Databricks 中数据的访问权限。

你需要配置一个 Databricks SQL 仓库。建议使用无服务器仓库，以加快查询执行速度。

环境变量	描述
DATABRICKS_CLIENT_ID	服务主体的 OAuth2 客户端 ID
DATABRICKS_CLIENT_SECRET	服务主体的 OAuth2 客户端密钥
DATABRICKS_HOST	Databricks 工作区 URL，不包含开头的 https://。例如 adb-xxx.azuredatabricks.net。前往 Compute -> SQL warehouses -> Your Warehouse -> Connection details -> Server hostname
DATABRICKS_HTTP_PATH	SQL 端点的 HTTP 路径，例如 /sql/1.0/warehouses/xxx。前往 Compute -> SQL warehouses -> Your Warehouse -> Connection details -> HTTP path

BigQuery

如果您将 BigQuery 用作数据平台，则需要创建一个服务账号，并为其分配访问数据产品所需的权限。将服务账号密钥下载为 JSON 文件。

当 Data Mesh Manager 中的访问请求获得批准后，您可以使用 BigQuery Connector 自动管理 BigQuery 中的权限。

服务账号需要以下 IAM 角色：

• BigQuery Data Viewer - 用于查询数据集
• BigQuery Job User - 用于以作业形式执行查询

环境变量	描述
BIGQUERY_CREDENTIALS_PATH	服务账号密钥 JSON 文件的路径

注意：Google Cloud 项目 ID 和数据集信息在数据产品的输出端口服务器配置中指定，而非作为环境变量。

获取服务账号凭据的方法：

1. 前往 Google Cloud Console
2. 导航至 IAM & Admin > Service Accounts
3. 创建新的服务账号，或使用现有账号
4. 添加 BigQuery Data Viewer 和 BigQuery Job User 角色
5. 生成并下载 JSON 密钥文件
6. 将 BIGQUERY_CREDENTIALS_PATH 设置为 JSON 文件的路径

支持的服务器类型

dataproduct_query 工具支持对数据产品执行查询。MCP 客户端会根据数据合约及其数据模型结构和语义来构造 SQL 查询。

目前开箱即用地支持以下服务器类型：

服务器类型	状态	说明
Snowflake	✅	需要设置 SNOWFLAKE_USER、SNOWFLAKE_PASSWORD、SNOWFLAKE_WAREHOUSE、SNOWFLAKE_ROLE 环境变量
Databricks	✅	需要设置 DATABRICKS_HOST、DATABRICKS_HTTP_PATH、DATABRICKS_CLIENT_ID、DATABRICKS_CLIENT_SECRET 环境变量
BigQuery	✅	需要设置 BIGQUERY_CREDENTIALS_PATH 环境变量
S3	即将推出	通过 DuckDB 客户端实现
Fabric	即将推出

**注意：**对于其他数据平台类型（例如 Redshift、PostgreSQL），请将额外的平台专用 MCP 服务添加到 MCP 客户端中。

贡献

有关开发环境设置和贡献指南，请参阅 CONTRIBUTING.md。

Credits

Maintained by Simon Harrer, André Deuerling, and Jochen Christ.

.github-corner:hover .octo-arm{animation:octocat-wave 560ms ease-in-out}@keyframes octocat-wave{0%,100%{transform:rotate(0)}20%,60%{transform:rotate(-25deg)}40%,80%{transform:rotate(10deg)}}@media (max-width:500px){.github-corner:hover .octo-arm{animation:none}.github-corner .octo-arm{animation:octocat-wave 560ms ease-in-out}}