mcp-federated-data

Servidor MCP federado que une metadatos relacionales (MySQL) con valores de series temporales (InfluxDB) detrás de una única capa de entidades compatible con LLM.

Idiomas: Inglés · 中文

El problema

Cuando un LLM necesita responder a una pregunta como:

"muéstrame las tendencias de temperatura de los últimos 7 días para los sensores activos en la zona A"

tiene que recorrer dos almacenes:

• Metadatos relacionales (qué sensores existen, dónde, qué tipo, qué unidades) — MySQL

• Valores de series temporales (las lecturas a lo largo del tiempo) — InfluxDB

Conectar dos servidores MCP independientes obliga al LLM a:

1. Consultar MySQL → encontrar entidades coincidentes → extraer sus ids

2. Insertar esos ids en una consulta de InfluxDB

3. Unir los dos conjuntos de resultados en su propio contexto

Los LLM fallan de forma fiable en el paso 3, especialmente cuando la clave de unión es compuesta (por ejemplo, una etiqueta de InfluxDB es la concatenación de dos campos relacionales).

Este servidor colapsa los tres pasos en una sola llamada de herramienta.

Lo que ve el LLM

Cinco herramientas, sin SQL ni Flux en el prompt:

Cada herramienta incluye descripciones detalladas en JSON-Schema para que el LLM elija los argumentos correctos sin trucos en el prompt.

En qué se diferencia de las configuraciones ingenuas de dos servidores

Inicio rápido

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

Conéctese desde cualquier cliente MCP (Claude Desktop, Cursor, mcp-inspector).

Pruébelo

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

Configuración

Todo el comportamiento del servidor se define en un único archivo YAML. Tres secciones.

1. Fuentes de datos

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. Entidades

Cada entidad se vincula a una tabla o vista relacional, con relaciones opcionales y un enlace opcional a series temporales.

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. Valores predeterminados (barreras de seguridad)

defaults:
  max_entities_per_query: 50
  max_points_per_entity: 500
  query_timeout_ms: 15000

Claves de unión compuestas (destacado de v0.2)

Cuando el valor de la etiqueta de InfluxDB es un compuesto de múltiples campos relacionales — común en sistemas IoT / industriales donde una etiqueta como 400001240.438000066 codifica {deviceId}.{signalId} — declárelo así:

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

El servidor:

1. Extrae metadatos de MySQL (los campos locales del compositor se incluyen automáticamente en SELECT).

2. Compone el valor de la etiqueta de cada fila usando la plantilla.

3. Inserta la lista compuesta en el filtro de etiquetas de InfluxDB.

4. Une los resultados de vuelta mediante la misma plantilla.

Los enlaces de un solo campo (la forma de la v0.1) siguen funcionando sin cambios: el servidor los trata como un compuesto de un solo elemento, por lo que todas las rutas permanecen uniformes.

Cómo funciona — el núcleo de 50 líneas

Cada herramienta federada sigue los mismos tres pasos:

1. Obtener metadatos — consulta relacional contra el almacén de negocio, con comprobaciones de identificadores seguros y WHERE parametrizado. Los campos de filtro permitidos están restringidos a los declarados en la configuración de la entidad.

2. Obtener series temporales — extraer valores de clave de unión del paso 1, insertarlos en un filtro de etiquetas contra el almacén de series temporales, con agregación opcional y límite de puntos por llamada.

3. Fusionar — agrupar puntos de series temporales por el valor de la etiqueta remota, luego unir cada fila de metadatos con su serie ordenada.

Sin analizador SQL. Sin planificador de consultas entre almacenes. Por diseño.

Arquitectura

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

Dónde encaja esto

mcp-federated-data está impulsado por esquemas, no es específico de un dominio. Se aplica en cualquier lugar donde los metadatos de negocio vivan en un almacén relacional y los valores observados vivan en un almacén de series temporales:

• Telemetría de dispositivos IoT

• Monitoreo de procesos industriales

• Gestión del rendimiento de activos

• Automatización de edificios

• Monitoreo de energía / potencia

• Redes de sensores ambientales

• Monitoreo de dispositivos de red

Si su stack es MySQL + InfluxDB y quiere que los LLM razonen sobre él — este servidor es para usted.

Comparación con proyectos relacionados

Contribución

Las incidencias y PR son bienvenidas. Antes de abrir una:

• Para nuevos adaptadores de fuentes de datos — abra una incidencia primero para que podamos alinearnos en la interfaz.

• Para nuevas herramientas — siga el patrón existente de JSON-Schema + registro de auditoría.

• Las API públicas deben mantener la compatibilidad con versiones anteriores dentro de las versiones menores.

Agradecimientos

• Model Context Protocol — el estándar que habla este servidor

• @modelcontextprotocol/sdk — SDK de referencia en TypeScript

• Cube.dev / dbt — trabajo previo sobre la idea de "capa semántica"

• Comunidades de InfluxDB y MySQL

Licencia

MIT