mcp-federated-data

Федеративный MCP-сервер, который объединяет реляционные метаданные (MySQL) со значениями временных рядов (InfluxDB) в единый слой сущностей, удобный для LLM.

Языки: English · 中文

Проблема

Когда LLM нужно ответить на вопрос вроде:

"покажи мне тренды температуры за последние 7 дней для активных датчиков в зоне А",

ей приходится обращаться к двум хранилищам:

• Реляционные метаданные (какие датчики существуют, где, какого типа, в каких единицах) — MySQL

• Значения временных рядов (показания во времени) — InfluxDB

Подключение двух независимых MCP-серверов заставляет LLM:

1. Запросить MySQL → найти соответствующие сущности → извлечь их идентификаторы

2. Вставить эти идентификаторы в запрос к InfluxDB

3. Объединить два набора результатов в своем собственном контексте

LLM стабильно спотыкаются на шаге 3 — особенно когда ключ объединения является составным (например, тег InfluxDB представляет собой конкатенацию двух реляционных полей).

Этот сервер сворачивает все три шага в один вызов инструмента.

Что видит LLM

Пять инструментов, никакого SQL или Flux в промпте:

Каждый инструмент поставляется с подробными описаниями JSON-Schema, поэтому LLM выбирает правильные аргументы без ухищрений в промпте.

Чем это отличается от наивных настроек с двумя серверами

Быстрый старт

git clone https://github.com/baller-coder/mcp-federated-data.git
cd mcp-federated-data
pnpm install

# Sample environment (MySQL + InfluxDB in Docker, with seeded data)
docker compose -f examples/industrial-monitoring/docker-compose.yml up -d
pnpm seed

# Start the MCP server over stdio
pnpm dev -- --config examples/industrial-monitoring/config.yaml

Подключайтесь из любого MCP-клиента (Claude Desktop, Cursor, mcp-inspector).

Попробуйте

// list active sites
{
  "name": "list_entities",
  "arguments": {
    "entity": "site",
    "filters": [{ "field": "status", "op": "eq", "value": "active" }]
  }
}

// list sensors attached to site 1
{
  "name": "list_related",
  "arguments": {
    "source_entity": "site",
    "source_id": 1,
    "target_entity": "sensor"
  }
}

// federated query — metadata + timeseries in ONE call
{
  "name": "get_entity_timeseries",
  "arguments": {
    "entity": "sensor",
    "filters": [
      { "field": "site_id", "op": "eq", "value": 1 },
      { "field": "kind",    "op": "eq", "value": "temperature" }
    ],
    "time_range":  { "start": "-7d" },
    "aggregation": { "window": "1h", "fn": "mean" }
  }
}

// compare 3 specific sensors over the same window
{
  "name": "compare_timeseries",
  "arguments": {
    "entity": "sensor",
    "ids":    [101, 201, 301],
    "time_range":  { "start": "-24h" },
    "aggregation": { "window": "10m", "fn": "mean" }
  }
}

Конфигурация

Все поведение сервера определяется в одном YAML-файле. Три раздела.

1. Источники данных

datasources:
  - name: business
    type: mysql
    host: localhost
    port: 3306
    database: my_db
    username: readonly_user
    password: secret

  - name: timeseries
    type: influxdb
    url: http://localhost:8086
    token: my-token
    org: my_org
    bucket: my_bucket

2. Сущности

Каждая сущность привязывается к реляционной таблице или представлению, с опциональными связями и опциональной привязкой к временным рядам.

entities:
  - name: site
    description: Physical monitoring location.
    source:
      datasource: business
      table: sites
      primary_key: id
      fields:
        - { name: id,     type: number }
        - { name: name,   type: string, description: Display name }
        - { name: region, type: string }
        - { name: status, type: string, description: "active / inactive / maintenance" }

  - name: sensor
    description: A sensor attached to a site.
    source:
      datasource: business
      table: sensors
      primary_key: id
      fields:
        - { name: id,      type: number }
        - { name: site_id, type: number }
        - { name: name,    type: string }
        - { name: kind,    type: string, description: "temperature / humidity / voltage / ..." }
        - { name: unit,    type: string }
    relations:
      - target: site
        type: many-to-one
        local_key: site_id
        foreign_key: id
    timeseries:
      datasource: timeseries
      measurement: sensor_data
      value_field: value
      join_key:
        local: id
        remote_tag: sensor_id

3. Значения по умолчанию (ограничители)

defaults:
  max_entities_per_query: 50
  max_points_per_entity: 500
  query_timeout_ms: 15000

Составные ключи объединения (особенность v0.2)

Когда значение тега InfluxDB является составным из нескольких реляционных полей — что часто встречается в IoT / промышленных системах, где тег вроде 400001240.438000066 кодирует {deviceId}.{signalId} — объявите его так:

timeseries:
  measurement: sensor_data
  value_field: value
  join_key:
    local: [device_id, signal_id]
    remote_tag: signal_id
    composer: "{device_id}.{signal_id}"

Сервер:

1. Извлекает метаданные из MySQL (локальные поля композитора автоматически включаются в SELECT).

2. Составляет значение тега для каждой строки, используя шаблон.

3. Передает составленный список в фильтр тегов InfluxDB.

4. Объединяет результаты обратно по тому же шаблону.

Привязки к одному полю (формат v0.1) продолжают работать без изменений — сервер обрабатывает их как одноэлементный составной ключ, поэтому все пути остаются единообразными.

Как это работает — ядро из 50 строк

Каждый федеративный инструмент следует одним и тем же трем шагам:

1. Извлечение метаданных — реляционный запрос к бизнес-хранилищу с проверкой безопасных идентификаторов и параметризованным WHERE. Разрешенные поля фильтрации ограничены теми, что объявлены в конфигурации сущности.

2. Извлечение временных рядов — получение значений ключа объединения из шага 1, передача их в фильтр тегов для хранилища временных рядов с опциональной агрегацией и лимитом точек на вызов.

3. Объединение — группировка точек временных рядов по значению удаленного тега, затем объединение каждой строки метаданных с соответствующим отсортированным рядом.

Никакого SQL-парсера. Никакого планировщика запросов между хранилищами. Это сделано намеренно.

Архитектура

┌──────────────────────────────────┐
│  MCP client (Claude / Cursor)    │
└────────────┬─────────────────────┘
             │ stdio  (JSON-RPC)
             ▼
┌──────────────────────────────────┐
│       mcp-federated-data         │
│  ┌────────────────────────────┐  │
│  │  Tools  (5 tools)          │  │
│  ├────────────────────────────┤  │
│  │  Entity registry           │  │
│  │  Join-key normalizer       │  │
│  │  Composer engine           │  │
│  │  Guards (limits/timeout)   │  │
│  │  Audit logger              │  │
│  ├────────────────────────────┤  │
│  │  Datasource adapters       │  │
│  └─────────┬────────┬─────────┘  │
└────────────┼────────┼────────────┘
             ▼        ▼
        ┌──────┐ ┌──────────┐
        │MySQL │ │InfluxDB  │
        └──────┘ └──────────┘

Где это применимо

mcp-federated-data — это решение, управляемое схемой, а не предметно-ориентированное. Оно применимо везде, где бизнес-метаданные живут в реляционном хранилище, а наблюдаемые значения — в хранилище временных рядов:

• Телеметрия IoT-устройств

• Мониторинг промышленных процессов

• Управление производительностью активов

• Автоматизация зданий

• Мониторинг энергетики / питания

• Сети экологических датчиков

• Мониторинг сетевых устройств

Если ваш стек — MySQL + InfluxDB и вы хотите, чтобы LLM могли рассуждать на его основе — этот сервер для вас.

Сравнение с похожими проектами

Участие в разработке

Приветствуются Issues и PR. Перед открытием:

• Для новых адаптеров источников данных — сначала откройте issue, чтобы мы могли согласовать интерфейс.

• Для новых инструментов — соблюдайте существующий шаблон JSON-Schema + лог аудита.

• Публичные API должны сохранять обратную совместимость в рамках минорных версий.

Благодарности

• Model Context Protocol — стандарт, который использует этот сервер

• @modelcontextprotocol/sdk — эталонный SDK на TypeScript

• Cube.dev / dbt — предыдущие наработки по идее "семантического слоя"

• Сообщества InfluxDB и MySQL

Лицензия

MIT