DataForge Semantic MCP Server

Read-only semantic gateway between AI agents and DataForge Product API. Fetches projects, versions, measures, dimensions and full RMD, normalizes and caches the data, and exposes it via MCP protocol or as a Python library.

Features

• Library-first — use directly from Python, no MCP server required
• MCP adapter — 7 tools for Claude Desktop, Cursor and other MCP clients
• Caching — file-based cache with TTL and last-known-good fallback
• Normalization — inconsistent API fields mapped to clean canonical models
• Retry & error handling — exponential backoff on 5xx, proper error codes for auth issues

Quick Start

Installation

pip install -e ".[dev]"

Configuration

Copy .env.example to .env and set your values:

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

As a Python Library

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())

As an MCP Server (stdio)

python -m dataforge_mcp

Add to Claude Desktop config (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 mode)

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

MCP Tools

Tool	Description
df_health	Check server, API and cache status
df_list_projects	List available DataForge projects
df_list_versions	List versions for a project
df_get_measures	Get measures (metrics) for a project version
df_get_dimensions	Get dimensions for a project version
df_get_rmd	Get full RMD (measures + dimensions)
df_refresh_cache	Force refresh cached data

Architecture

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 is the single entry point. MCP tools only delegate to it.

Development

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

# Run tests
pytest

# Lint
ruff check src/ tests/

# Format
ruff format src/ tests/

Configuration Reference

Variable	Default	Description
DATAFORGE_BASE_URL	https://api.prod-df.businessqlik.com	DataForge API base URL
DATAFORGE_API_KEY	—	API key (required)
DEFAULT_LANGUAGE	ru	Default language for measures/dimensions
CACHE_DIR	./cache	Cache directory path
CACHE_TTL_SECONDS	3600	Cache TTL in seconds
MCP_TRANSPORT	stdio	Transport: stdio or sse
LOG_LEVEL	INFO	Log level

Design Documents

Detailed specs are in the docs/ directory.