内容
# mcp-federated-data
> **联合MCP服务器**，将关系元数据（MySQL）与时间序列值（InfluxDB）结合在一个单一的、适合LLM的实体层后面。
**语言：** [English](./README.md) · [中文](./README.zh-CN.md)
---
## 问题
当LLM需要回答诸如以下问题时：
> *"显示区域A中活动传感器的最近7天温度趋势"*
它必须遍历 **两个存储**：
- **关系元数据**（存在哪些传感器，在哪里，什么类型，什么单位）—— MySQL
- **时间序列值**（随时间变化的读数）—— InfluxDB
插入两个 **独立** 的MCP服务器迫使LLM：
1. 查询MySQL → 查找匹配的实体 → 提取它们的ID
2. 将这些ID推入InfluxDB查询
3. **在自己的上下文中缝合两个结果集**
LLM在步骤3中可靠地 stumbled — 特别是在连接键是复合键时（例如，InfluxDB标签是两个关系字段的串联）。
该服务器将所有三个步骤合并为 **一个工具调用**。
---
## LLM所见
五个工具，无需在提示中输入SQL或Flux：
| 工具 | 目的 |
|---|---|
| `list_entities` | 按实体字段过滤业务实体 |
| `get_entity` | 按主键查找单个实体 |
| `list_related` | 在实体之间行走配置的关系 |
| `get_entity_timeseries` | **联合** - 元数据 + 时间序列在一个调用中 |
| `compare_timeseries` | 在同一时间窗口内比较2-20个特定实体 |
每个工具都附带丰富的JSON-Schema描述，以便LLM在不使用提示技巧的情况下选择正确的参数。
---
## 它与朴素的双服务器设置有何不同
| 场景 | 两个独立的MCP服务器 | `mcp-federated-data` |
|---|---|---|
| "区域A中传感器的趋势" | LLM：查询MySQL → 提取ID → 推入Flux → 在其头部连接 | 一个`get_entity_timeseries`调用 |
| `{deviceId}。{signalId}`等复合标签 | LLM在提示中组成字符串 - 容易出错 | `composer: "{deviceId}。{signalId}"`在YAML中，服务器执行 |
| 元数据↔时间序列对齐 | LLM执行连接，频繁错误配对 | 服务器按配置键连接 |
| 数据量膨胀 | 无保护 | 强制最大实体 + 自动降采样 + 每个实体点上限 |
| LLM的业务语义 | 原始`CREATE TABLE`仅 | YAML字段带有`description` |
| 审计轨迹 | 无 | 结构化每个调用审计日志 |
---
## 快速入门
```bash
git clone https://github.com/baller-coder/mcp-federated-data.git
cd mcp-federated-data
pnpm install
# 示例环境（MySQL + InfluxDB在Docker中，带有种子数据）
docker compose -f examples/industrial-monitoring/docker-compose.yml up -d
pnpm seed
# 在stdio上启动MCP服务器
pnpm dev -- --config examples/industrial-monitoring/config.yaml
```
从任何MCP客户端（Claude Desktop，Cursor， [mcp-inspector](https://github.com/modelcontextprotocol/inspector) ）连接。
### 试用
```jsonc
// 列出活动站点
{
"name": "list_entities",
"arguments": {
"entity": "site",
"filters": [{ "field": "status", "op": "eq", "value": "active" }]
}
}
// 列出连接到站点1的传感器
{
"name": "list_related",
"arguments": {
"source_entity": "site",
"source_id": 1,
"target_entity": "sensor"
}
}
// 联合查询 - 元数据 + 时间序列在一个调用中
{
"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" }
}
}
// 在同一时间窗口内比较3个特定传感器
{
"name": "compare_timeseries",
"arguments": {
"entity": "sensor",
"ids": [101, 201, 301],
"time_range": { "start": "-24h" },
"aggregation": { "window": "10m", "fn": "mean" }
}
}
```
---
## 配置
整个服务器行为在一个YAML文件中定义。分为三个部分。
### 1. 数据源
```yaml
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. 实体
每个实体绑定到一个关系表或视图，具有可选的关系和可选的时间序列绑定。
```yaml
entities:
- name: site
description: 物理监控位置。
source:
datasource: business
table: sites
primary_key: id
fields:
- { name: id, type: number }
- { name: name, type: string, description: 显示名称 }
- { name: region, type: string }
- { name: status, type: string, description: "active / inactive / maintenance" }
- name: sensor
description: 连接到站点的传感器。
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. 默认值（防护栏）
```yaml
defaults:
max_entities_per_query: 50
max_points_per_entity: 500
query_timeout_ms: 15000
```
---
## 复合连接键（v0.2亮点）
当InfluxDB标签值是 **多个关系字段的复合** 时 - 在IoT /工业系统中常见，其中标签如`400001240.438000066`编码 `{deviceId}。{signalId}` - 声明如下：
```yaml
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客户端（Claude / Cursor） │
└────────────┬─────────────────────┘
│ stdio （JSON-RPC）
▼
┌──────────────────────────────────┐
│ mcp-federated-data │
│ ┌────────────────────────────┐ │
│ │ 工具（5个工具） │ │
│ ├────────────────────────────┤ │
│ │ 实体注册表 │ │
│ │ 连接键规范化 │ │
│ │ 作曲引擎 │ │
│ │ 防护（限制/超时） │ │
│ │ 审计日志 │ │
│ ├────────────────────────────┤ │
│ │ 数据源适配器 │ │
│ └─────────┬────────┬─────────┘ │
└────────────┼────────┼────────────┘
▼ ▼
┌──────┐ ┌──────────┐
│MySQL │ │InfluxDB │
└──────┘ └──────────┘
```
---
## 它适合哪里
`mcp-federated-data`是 **模式驱动** 的，而不是特定于领域的。它适用于任何业务元数据存在于关系存储中，而观察值存在于时间序列存储中：
- 物联网设备遥测
- 工业过程监控
- 资产性能管理
- 建筑自动化
- 能量/电源监控
- 环境传感器网络
- 网络设备监控
如果您的堆栈是 **MySQL + InfluxDB** ，并且您希望LLM对其进行推理 - 此服务器就是为您准备的。
---
## 与相关项目比较
| | mcp-federated-data | mcp-server-mysql | mcp-server-influxdb | Wren AI / Vanna |
|---|:---:|:---:|:---:|:---:|
| 跨存储连接 | ✅ | ❌ | ❌ | 部分（自己的引擎） |
| 复合标签作曲家 | ✅ | 不适用 | ❌ | ❌ |
| LLM安全SQL防护 | ✅ | 变化 | 不适用 | ✅ |
| 模式作为语义层 | ✅ | ❌ | ❌ | ✅ |
| 配置 | YAML | 环境变量/参数 | 环境变量/参数 | 专用DSL |
| 范围 | 窄但深 | 薄包装 | 薄包装 | 全BI平台 |
---
## 贡献
欢迎提出问题和PR。在打开之前：
- 对于新的数据源适配器 - 首先打开一个问题，以便我们统一接口。
- 对于新工具 - 匹配现有的JSON-Schema +审计日志模式。
- 公共API必须在小版本内保持向后兼容。
---
## 致谢
- [Model Context Protocol](https://modelcontextprotocol.io) - 该服务器所遵循的标准
- [`@modelcontextprotocol/sdk`](https://github.com/modelcontextprotocol/typescript-sdk) - TypeScript参考SDK
- [Cube.dev](https://cube.dev) / [dbt](https://www.getdbt.com/) - 先前关于“语义层”理念的作品
- [InfluxDB](https://www.influxdata.com/) 和 [MySQL](https://www.mysql.com/) 社区
---
## License
[MIT](./LICENSE)
连接信息
猜你喜欢
暂无推荐服务器