mcp-federated-data

Föderierter MCP-Server, der relationale Metadaten (MySQL) mit Zeitreihenwerten (InfluxDB) hinter einer einzigen, LLM-freundlichen Entitätsschicht verbindet.

Sprachen: Englisch · Chinesisch

Das Problem

Wenn ein LLM eine Frage beantworten muss wie:

"Zeige mir die Temperaturtrends der letzten 7 Tage für aktive Sensoren in Zone A"

muss es zwei Speicher durchsuchen:

• Relationale Metadaten (welche Sensoren existieren, wo, welcher Typ, welche Einheiten) — MySQL

• Zeitreihenwerte (die Messwerte im Zeitverlauf) — InfluxDB

Das Einbinden von zwei unabhängigen MCP-Servern zwingt das LLM dazu:

1. MySQL abzufragen → passende Entitäten zu finden → deren IDs zu extrahieren

2. Diese IDs in eine InfluxDB-Abfrage einzufügen

3. Die beiden Ergebnismengen im eigenen Kontext zusammenzuführen

LLMs scheitern zuverlässig bei Schritt 3 — besonders wenn der Join-Schlüssel zusammengesetzt ist (z. B. wenn ein InfluxDB-Tag die Verkettung zweier relationaler Felder ist).

Dieser Server fasst alle drei Schritte in einem Tool-Aufruf zusammen.

Was das LLM sieht

Fünf Tools, kein SQL oder Flux im Prompt:

Jedes Tool wird mit umfangreichen JSON-Schema-Beschreibungen geliefert, sodass das LLM die richtigen Argumente ohne Prompt-Tricks auswählt.

Wie es sich von naiven Zwei-Server-Setups unterscheidet

Schnellstart

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

Verbindung von jedem beliebigen MCP-Client (Claude Desktop, Cursor, mcp-inspector).

Ausprobieren

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

Konfiguration

Das gesamte Serververhalten wird in einer einzigen YAML-Datei definiert. Drei Abschnitte.

1. Datenquellen

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. Entitäten

Jede Entität bindet an eine relationale Tabelle oder View, mit optionalen Beziehungen und einer optionalen Zeitreihenbindung.

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. Standardwerte (Leitplanken)

defaults:
  max_entities_per_query: 50
  max_points_per_entity: 500
  query_timeout_ms: 15000

Zusammengesetzte Join-Schlüssel (v0.2 Highlight)

Wenn der InfluxDB-Tag-Wert eine Zusammensetzung aus mehreren relationalen Feldern ist — üblich in IoT-/Industriesystemen, wo ein Tag wie 400001240.438000066 den Wert {deviceId}.{signalId} kodiert — deklarieren Sie es so:

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

Der Server:

1. Zieht Metadaten aus MySQL (die lokalen Felder des Composers werden automatisch in SELECT aufgenommen).

2. Setzt den Tag-Wert jeder Zeile mithilfe der Vorlage zusammen.

3. Schiebt die zusammengesetzte Liste in den InfluxDB-Tag-Filter.

4. Verbindet die Ergebnisse über dieselbe Vorlage wieder zurück.

Einzel-Feld-Bindungen (die v0.1-Form) funktionieren weiterhin unverändert — der Server behandelt sie als ein ein-elementiges Kompositum, sodass alle Pfade einheitlich bleiben.

Funktionsweise — der 50-Zeilen-Kern

Jedes föderierte Tool folgt denselben drei Schritten:

1. Metadaten abrufen — relationale Abfrage gegen den Geschäftsspeicher, mit sicheren Identifikator-Prüfungen und parametrisiertem WHERE. Erlaubte Filterfelder sind auf die in der Entitätskonfiguration deklarierten Felder beschränkt.

2. Zeitreihen abrufen — Join-Schlüsselwerte aus Schritt 1 ziehen, in einen Tag-Filter gegen den Zeitreihenspeicher schieben, mit optionaler Aggregation und Punkt-Limit pro Aufruf.

3. Zusammenführen — Zeitreihenpunkte nach dem Remote-Tag-Wert gruppieren, dann jede Metadatenzeile mit ihrer sortierten Serie verknüpfen.

Kein SQL-Parser. Kein speicherübergreifender Abfrageplaner. Absichtlich so konzipiert.

Architektur

┌──────────────────────────────────┐
│  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  │
        └──────┘ └──────────┘

Wo dies passt

mcp-federated-data ist schema-gesteuert, nicht domänenspezifisch. Es lässt sich überall dort einsetzen, wo Geschäftsmetadaten in einem relationalen Speicher und beobachtete Werte in einem Zeitreihenspeicher leben:

• IoT-Gerätetelemetrie

• Überwachung industrieller Prozesse

• Asset-Performance-Management

• Gebäudeautomation

• Energie-/Stromüberwachung

• Umweltsensornetzwerke

• Überwachung von Netzwerkgeräten

Wenn Ihr Stack MySQL + InfluxDB ist und Sie möchten, dass LLMs darüber nachdenken können — ist dieser Server für Sie geeignet.

Vergleich mit verwandten Projekten

Mitwirken

Issues und PRs sind willkommen. Bevor Sie eines öffnen:

• Für neue Datenquellen-Adapter — öffnen Sie zuerst ein Issue, damit wir uns auf die Schnittstelle abstimmen können.

• Für neue Tools — halten Sie sich an das bestehende JSON-Schema + Audit-Log-Muster.

• Öffentliche APIs müssen die Abwärtskompatibilität innerhalb von Nebenversionen wahren.

Danksagungen

• Model Context Protocol — der Standard, den dieser Server spricht

• @modelcontextprotocol/sdk — TypeScript-Referenz-SDK

• Cube.dev / dbt — frühere Arbeiten zur Idee der "semantischen Schicht"

• InfluxDB und MySQL Communities

Lizenz

MIT