语义模型管理 MCP 服务器

DataForge Semantic MCP Server

Give your AI agent the business meaning behind your analytics — and the power to change it.

Quick start · Tools · Writing data · Open WebUI · Documentation

[!CAUTION]

THIS SERVER IS NOT READ-ONLY — IT CAN DELETE YOUR DATA

38 of its 65 tools create, overwrite and permanently delete DataForge content: projects, versions, measures, dimensions, facts, dimension groups, fact tables, relationships, access grants and Git connections.

Tool	What it does
df_delete_project	Deletes a project with every version and all of its content
df_import_version_from_git (method=replace)	Overwrites an entire version
df_write_measure (mode=replace)	Resets every optional field you do not pass

There is no kill switch in this server.

The only thing between an LLM agent and your production semantic model is the effective project role of the API key in DATAFORGE_API_KEY.

➜ Want read-only? Use a read-only key.

Issue a key whose project role is analyst or viewer. Every write endpoint requires developer or above and answers 403 DF_API.WRITE_ACCESS_DENIED — the API enforces this, this server does not. Point agents at a non-production version first.

Every write lands in the DataForge audit log with before/after snapshots.

Why

DataForge holds the business definitions behind your analytics: what "Total Revenue" actually means, which database column feeds it, how customers are grouped. Without that context an AI agent is guessing.

This server hands the agent that context — structured, normalized and cached — and gives it a safe, idempotent way to evolve the model.

What You Get

65 MCP tools — 24 read-only, 41 that change state (38 of them write to DataForge).

Read — 24 tools

Write — 41 tools ⚠️

Full reference: read tools · data model · write tools

Quick Start

1 · Install

pip install -e ".[dev]"

2 · Configure

cp .env.example .env

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

DATAFORGE_BASE_URL is the root of the API, not of the site — requests go to <base>/df-api/v2/…. Installations that serve the API behind a prefix need it spelled out (https://dataforge.example.com/api); point it at the site root and the web app answers with HTML and HTTP 200, which the server reports as DATAFORGE_INVALID_RESPONSE.

[!WARNING] The key decides what an agent can destroy. analyst / viewer → effectively read-only. developer and above → can delete measures, versions and whole projects.

3 · Run

python -m dataforge_mcp          # stdio, for Claude Desktop / Cursor
docker compose up                # Streamable HTTP on http://localhost:8080/mcp

Over HTTP the port is the only thing between the network and the 38 write tools, so set MCP_AUTH_TOKEN for any bind that is not loopback. Full reference: docs/api/transports.md.

Or skip MCP entirely and use it as a library:

import asyncio
from dataforge_mcp import create_semantic_service

async def main():
    service = create_semantic_service()

    projects = await service.list_projects()
    project_id = projects["projects"][0]["id"]

    versions = await service.list_versions(project_id=project_id)
    version_id = next(v["id"] for v in versions["versions"] if v["is_global"])

    rmd = await service.get_rmd(project_id=project_id, version_id=version_id)
    print(f"{rmd['stats']['measure_count']} measures, "
          f"{rmd['stats']['dimension_count']} dimensions, "
          f"{rmd['stats']['fact_count']} facts")

asyncio.run(main())

Claude Desktop

{
  "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"
      }
    }
  }
}

[!CAUTION] Before you paste this: the key above decides whether an agent can delete your projects. Use an analyst / viewer key for read-only work. Write tools carry readOnlyHint=false and destructiveHint=true, so a well-behaved client asks you to confirm — but that confirmation belongs to the client, not to this server.

Then ask Claude: "What measures are available in the Fashion Retail project?" — it will chain df_list_projects → df_list_versions → df_get_measures on its own.

Open WebUI

Open WebUI speaks Streamable HTTP only, which this server now serves natively — no mcpo proxy needed.

cp .env.example .env
# set DATAFORGE_API_KEY, and MCP_AUTH_TOKEN:
#   python -c "import secrets; print(secrets.token_urlsafe(32))"
docker compose up -d

Then in Open WebUI, as an administrator: Settings → Admin → Integrations → + Add Connection, type MCP (Streamable HTTP), URL http://host.docker.internal:8080/mcp, Auth Bearer with your MCP_AUTH_TOKEN. All 65 tools appear.

[!WARNING] Every Open WebUI user who can use the tools gets all 65, including the 38 that delete data — and they all share one DataForge key. Start with an analyst / viewer key and a non-production version.

Full walkthrough, URL table and troubleshooting: docs/api/open-webui.md.

Examples

Read, then write

Every measure, dimension and fact carries a stable id. That id is exactly what the write tools take, so a row you just read goes straight back:

measures = await service.get_measures(project_id=392, version_id=948)
target = next(m for m in measures["measures"] if m["name"] == "Total revenue")

await service.write_measure(
    project_id=392, version_id=948,
    mode="update",                       # PATCH — touches only what you pass
    measure_id=int(target["id"]),
    measure_description="Gross revenue across all channels",
)

mode="replace" is a PUT: it resets every optional field you omit. Use update unless you mean a full overwrite.

SQL for a measure

measures = await service.get_measures(project_id=392, version_id=948, include_sql=True)

for m in measures["measures"]:
    if m.get("sql_code"):
        for script in m["sql_code"]["sql_scripts"]:
            print(f"{m['name']} @ {script['fact_table_name']}:\n  {script['sql']}")

SQL for a data mart

A generation failure comes back as data, not an error — that is what separates "this mart cannot produce SQL right now" from "this mart does not exist":

marts = await service.list_data_marts(project_id=392, version_id=948, search="Sales")
mart_id = int(marts["data_marts"][0]["id"])

result = await service.generate_sql(
    project_id=392, version_id=948, data_mart_id=mart_id, limit=100
)

if result["succeeded"]:
    print(result["target_db_type"], result["sql_script"])   # postgres | clickhouse | sqlserver
else:
    for err in result["validation_errors"]:
        print("cannot generate:", err["message"])

Explore the star schema

tables = await service.list_fact_tables(project_id=392, version_id=948)
for ft in tables["fact_tables"]:
    print(f"{ft['name']}: {ft['measures_count']}M / {ft['dimensions_count']}D")

rels = await service.list_relationships(project_id=392, version_id=948)
for r in rels["relationships"]:
    print(f"{r['source_fact_table']['name']} -> "
          f"{r['target_dimension_group']['name']} ({r['relationship_type']})")

Writing Data

result = await service.bulk_write_measures(
    project_id=392, version_id=948,
    measures=[
        {"measure_name": "Net revenue", "measure_type": "Base"},   # created
        {"id": "1000", "measure_description": "Updated"},          # updated
    ],
)

if result["status"] == "partial":
    for failure in result["failed"]:
        print(failure["index"], failure["error"]["code"])

{
  "error": {
    "code": "DATAFORGE_INVALID_SOURCE_TABLE",
    "api_code": "invalid_source_table",
    "message": "Table not found in the connection schema",
    "fields": [{ "field": "connected_source.table", "code": "invalid_value" }],
    "hint": "Call df_get_connection_schema to list the tables cached for this connection.",
    "retryable": false
  }
}

Troubleshooting

More, per area: configuration.md · transports.md

Documentation

Key Features

• Complete DF API v2 surface — reads, writes, version transfer, Git connection registry
• Three transports — stdio, Streamable HTTP (Open WebUI-ready, with bearer auth and Origin validation), and legacy SSE
• Safety annotations — every tool carries MCP readOnlyHint / destructiveHint, so clients can warn before a destructive call
• Idempotent writes — automatic Idempotency-Key makes retries safe
• Smart caching — TTL + last-known-good for reads; scope invalidation for writes
• Actionable errors — api_code, offending fields[] and a fix hint
• Rate-limit aware — 100 req/60 s per key, with retry_after_seconds
• Credential hygiene — API keys and Git tokens never reach logs or tool output
• Library-first — usable from plain Python, no MCP server required

Development

pip install -e ".[dev]"
pytest                                       # 502 tests
pytest tests/test_mcp_server.py -v           # end-to-end over MCP, stdio path
pytest tests/test_mcp_streamable_http.py -v  # end-to-end over MCP, HTTP path
ruff check src/ tests/
ruff format src/ tests/

Fixtures in tests/fixtures/api_v2.py are transcribed verbatim from the DataForge API documentation. When the API changes, that file is what gets updated first — inventing fixtures is how a v1-shaped test suite once hid five real defects.

License

Proprietary. For use with the DataForge Product API.