DataForge Semantic MCP Server

DataForge语义MCP服务器

AI代理和服务器之间的只读语义网关 DataForge API产品。获取项目、版本、度量、维度和完整的RMD（参考模型数据），对数据进行规范化和缓存，并通过MCP协议或作为Python库公开。

此服务器提供什么

DataForge商店 语义层元数据 --分析项目中使用的度量和维度的业务定义。此MCP服务器为AI代理提供了对元数据的结构化访问：

• 项目 --分析模型的顶级容器
• 版本 --项目语义层的快照（一个版本可以标记为“全局”/生产）
• 措施 --带有公式、数据类型、源映射的业务指标（如“总收入”、“毛利率”）
• 维度 --用于通过分组和源映射对数据进行切片的属性（例如“客户细分”、“地区”）
• RMD --将所有度量和维度组合在一个响应中的完整参考模型数据

所有回复均为 JSON.服务器将原始DataForge API字段规范化为一个干净、稳定的架构，因此无论API发生什么变化，AI代理都能获得一致的数据。

特性

• 图书馆优先 --直接从Python使用，不需要MCP服务器
• MCP适配器 --适用于Claude Desktop、Cursor和其他MCP客户端的7个工具
• 缓存 --基于文件的缓存，具有TTL和最后已知的良好回退
• 归一化 -映射到干净规范模型的不一致API字段
• 重试和错误处理 --5xx上的指数回退，身份验证问题的正确错误代码

快速开始

安装

pip install -e ".[dev]"

配置

复制 .env.example 到 .env 并设定你的价值观：

DATAFORGE_BASE_URL=https://api.prod-df.businessqlik.com
DATAFORGE_API_KEY=your_api_key_here
DEFAULT_LANGUAGE=ru

作为Python库

import asyncio
from dataforge_mcp import create_semantic_service

async def main():
    service = create_semantic_service()

    projects = await service.list_projects()
    print(projects)

    versions = await service.list_versions(project_id=392)
    print(versions)

    rmd = await service.get_rmd(project_id=392, version_id=948)
    print(f"Measures: {rmd['stats']['measure_count']}")
    print(f"Dimensions: {rmd['stats']['dimension_count']}")

asyncio.run(main())

作为MCP服务器（stdio）

python -m dataforge_mcp

添加到Claude桌面配置(claude_desktop_config.json):

{
  "mcpServers": {
    "dataforge": {
      "command": "python",
      "args": ["-m", "dataforge_mcp"],
      "env": {
        "DATAFORGE_BASE_URL": "https://api.prod-df.businessqlik.com",
        "DATAFORGE_API_KEY": "your_api_key_here"
      }
    }
  }
}

Docker（SSE模式）

cp .env.example .env
# edit .env with your API key
docker compose up

MCP工具——详细参考

所有工具都通过MCP返回JSON TextContent每个响应都有一个稳定的模式，如下所述。

df_health

检查服务器、API和缓存状态。

输入： 无

输出：

{
  "server_status": "ok",
  "product_api_status": "ok",
  "base_url": "https://api.prod-df.businessqlik.com",
  "cache_status": "ok"
}

______________________________________________________________________

df_list_projects

列出可使用配置的API密钥访问的可用DataForge项目。

输入：

输出：

{
  "projects": [
    {
      "id": 392,
      "name": "Fashion Retail",
      "description": "Retail analytics project"
    }
  ],
  "pagination": {
    "total": 10,
    "page": 1,
    "page_size": 100,
    "total_pages": 1
  }
}

项目领域：

______________________________________________________________________

df_list_versions

列出特定项目的版本。

输入：

输出：

{
  "project_id": 392,
  "versions": [
    {
      "id": 948,
      "name": "Global Version",
      "is_global": true
    }
  ],
  "pagination": {
    "total": 5,
    "page": 1,
    "page_size": 100,
    "total_pages": 1
  }
}

版本字段：

______________________________________________________________________

df_get_measures

获取项目版本的所有度量（业务指标）。

输入：

输出：

{
  "project_id": 392,
  "version_id": 948,
  "measures": [
    {
      "row_number": "1",
      "group": "Sales",
      "block": "Revenue",
      "name": "Total Revenue",
      "description": "Total revenue from all sales",
      "data_type": "Numeric",
      "measure_type": "Sum",
      "formula": "[Revenue]",
      "restrictions": null,
      "connected_source": {
        "db": "Sales DB",
        "schema": null,
        "table": "sales_table",
        "column": null
      },
      "original_source_type": "Database",
      "original_source": "Sales DB",
      "original_object": "sales_table",
      "report_for_verification": null,
      "comment": "Primary revenue metric",
      "display_data_type": null,
      "status": "Active",
      "relevance": "High",
      "required": true,
      "visibility": "Public",
      "responsible_for_data": "Data Team",
      "variation": null,
      "raw": {}
    }
  ]
}

测量字段：

______________________________________________________________________

df_get_dimensions

获取项目版本的所有维度（用于切片/过滤数据的属性）。

输入： 与...相同 df_get_measures.

输出：

{
  "project_id": 392,
  "version_id": 948,
  "dimensions": [
    {
      "row_number": "1",
      "group": "Sales",
      "block": "Customer",
      "name": "Customer ID",
      "description": "Unique customer identifier",
      "dimension_group": "Customer Attributes",
      "data_type": "String",
      "connected_source": {
        "db": 161,
        "schema": null,
        "table": "customers",
        "column": "CompanyName"
      },
      "dimension_type": null,
      "original_source_type": "Database",
      "original_source": "Sales DB",
      "original_object": "customers_table",
      "comment": "Primary key for customers",
      "formula": null,
      "value_options": null,
      "display_data_type": null,
      "source_data_type": null,
      "status": "Active",
      "relevance": "High",
      "required": true,
      "visibility": "Public",
      "responsible_for_data": "Data Team",
      "raw": {}
    }
  ]
}

维度字段：

______________________________________________________________________

df_get_rmd

在一次调用中获取完整的RMD——项目版本的所有度量和维度。

输入：

输出：

{
  "project": {
    "id": 392,
    "name": "Fashion Retail",
    "description": null
  },
  "version": {
    "id": 948,
    "name": "Global Version",
    "is_global": true
  },
  "measures": [ /* array of Measure objects (see df_get_measures) */ ],
  "dimensions": [ /* array of Dimension objects (see df_get_dimensions) */ ],
  "stats": {
    "measure_count": 120,
    "dimension_count": 75
  }
}

当 include_raw 是 false （默认） raw 从每个度量和维度中删除字段以减小响应大小。

______________________________________________________________________

df_refresh_cache

强制刷新项目版本的缓存数据。

输入：

输出：

{
  "status": "refreshed",
  "cache_key": "rmd:392:948:ru",
  "fetched_at": "2026-03-29T12:00:00+00:00"
}

______________________________________________________________________

连接的源对象

度量和维度可以参考其物理数据库源：

{
  "db": "Sales DB",
  "schema": null,
  "table": "sales_table",
  "column": "revenue"
}

错误格式

所有错误都遵循一致的格式：

{
  "error": {
    "code": "DATAFORGE_API_KEY_INVALID",
    "message": "Invalid API key",
    "details": {
      "http_status": 401
    }
  }
}

错误代码：

典型的AI代理工作流程

使用此服务器的AI代理的典型工作流程：

1. df_health              → verify connectivity
2. df_list_projects       → discover available projects
3. df_list_versions       → find the global (production) version
4. df_get_rmd             → fetch all measures and dimensions at once
   (or df_get_measures / df_get_dimensions separately)
5. Use the semantic metadata to understand the data model,
   generate queries, or answer business questions

建筑

AI Agent / MCP Client
    |
    v
MCP Adapter (mcp/)           — thin wrappers, no business logic
    |
    v
SemanticService (application/) — cache-first orchestration (CORE)
    |
    +--> DataForgeClient (dataforge/) — HTTP calls with retry
    +--> Normalizer (semantic/)       — raw API -> canonical models
    +--> FileCacheStore (cache/)      — TTL + last-known-good fallback

SemanticService 是唯一的切入点。MCP工具只委托给它。

发展

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Lint
ruff check src/ tests/

# Format
ruff format src/ tests/

配置参考