industrial-mcp
by zhiningsun
README.md
# industrial-mcp
> **工业设备的 USB-C 接口** — 让 AI 模型通过标准化协议读取和控制工业设备。
[](https://www.python.org/)
[](LICENSE)
[](https://modelcontextprotocol.io/)
[](https://docs.astral.sh/uv/)
---
## 解决的问题
在工业自动化领域，AI 模型（如 Claude）想要与工厂设备交互面临三大障碍：
1. **协议碎片化** — Modbus、OPC UA、MQTT 各有各的驱动和 API
2. **缺乏标准化 AI 接口** — 每个设备都需要定制集成代码（M×N 问题）
3. **调试困难** — stdio 模式下 stdout 被 JSON-RPC 独占，传统 `print()` 调试完全失效
**industrial-mcp** 通过 [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) 解决这些问题：
把工厂里的每台设备、每个传感器、每个执行器都变成 AI 可以直接理解和操作的标准接口。
---
## 功能特性
- 🔌 **5 个 MCP Tools** — `list_devices`、`read_device`、`set_speed`、`start_device`、`stop_device`
- 📡 **2 个 MCP Resources** — `device://{id}/status`、`plant://overview`
- 📋 **2 个 MCP Prompts** — 设备巡检、紧急停机
- 🏭 **3 种工业协议** — Modbus TCP、OPC UA、MQTT
- 🎮 **双模式运行** — 仿真模式（无需硬件）+ 真实设备模式
- 🔗 **物理关联模拟** — 电机转速 ↑ → 泵流量 ↑ → 传感器读数 ↑，自动传播
- 🐛 **完整调试工具链** — MCP 报文拦截器、结构化日志、健康检查端点
- 🤖 **Claude Desktop 集成** — 用自然语言控制设备："启动传送带"、"把温度调到 80°C"
- 🐳 **Docker 支持** — 一键部署到生产环境
---
## 系统架构
```mermaid
graph TD
CLAUDE[Claude Desktop / AI Model] -->|stdio JSON-RPC| SERVER[MCP Server]
subgraph "industrial-mcp"
SERVER --> DM[DeviceManager]
DM -->|mode: simulation| SIM[SimulatorAdapter]
DM -->|mode: real| MOD[ModbusAdapter]
DM -->|mode: real| MQTT[MQTTAdapter]
SIM --> DEV[BaseDevice subclasses]
DEV --> MOTOR[Motor Physics]
DEV --> PUMP[Pump Physics]
DEV --> SENSOR[Sensor Models]
MOD -->|pymodbus| HW1[Real Hardware]
MQTT -->|paho-mqtt| HW2[Real MQTT Broker]
end
MOTOR -.->|correlated_devices| PUMP
PUMP -.->|correlated_devices| SENSOR
```
---
## 快速开始
### 前置要求
- Python 3.10+
- [uv](https://docs.astral.sh/uv/) 包管理器
### 安装
```bash
git clone https://github.com/your-org/industrial-mcp.git
cd industrial-mcp
uv sync
```
### 运行（仿真模式 — 无需硬件）
```bash
# 直接启动 MCP 服务器
uv run python -m industrial_mcp.server
# 或使用交互式演示脚本
uv run python examples/demo_simulation.py
```
### 运行（MCP Inspector 测试）
```bash
npx @modelcontextprotocol/inspector uv run python -m industrial_mcp.server
```
浏览器打开后 → Connect → 点击 `list_devices` → 看到 8 台设备。
### Claude Desktop 集成
编辑配置文件：
**macOS:** `~/Library/Application Support/Claude/claude_desktop_config.json`
```json
{
"mcpServers": {
"industrial-mcp": {
"command": "uv",
"args": [
"run",
"--directory", "/absolute/path/to/industrial-mcp",
"python", "-m", "industrial_mcp.server"
]
}
}
}
```
完全退出 Claude Desktop（Cmd+Q）再重启，然后在对话中输入：
> 列出所有工业设备
详细配置指南 → [docs/claude-desktop-setup.md](docs/claude-desktop-setup.md)
---
## 自然语言控制示例
配置完成后，直接在 Claude Desktop 中用自然语言交互：
| 你对 Claude 说 | 实际效果 |
|---------------|---------|
| "列出所有设备" | 调用 `list_devices()`，展示 8 台设备及其状态 |
| "启动传送带电机" | 调用 `start_device("motor_01_modbus")`，电机开始运转 |
| "把传送带调到 1200 RPM" | 调用 `set_speed("motor_01_modbus", 1200)` |
| "检查冷却泵的温度和振动" | 调用 `read_device("pump_01_opcua")`，返回实时数据 |
| "做一次全面巡检" | 加载 device_checkup prompt，逐台检查并生成报告 |
| "紧急停机！" | 加载 emergency_stop prompt，按优先级关闭所有设备 |
---
## 项目结构
```
industrial-mcp/
├── pyproject.toml # 项目配置、依赖、入口点
├── LICENSE # MIT 许可证
├── README.md # 项目说明（本文件）
│
├── config/
│ └── devices.yaml # 设备定义（仿真 + 真实硬件配置）
│
├── src/industrial_mcp/
│ ├── server.py # MCP 服务器入口（Tools / Resources / Prompts）
│ ├── device_manager.py # 设备管理器（工厂模式 + 生命周期）
│ ├── health.py # HTTP 健康检查端点
│ ├── mcp_logger.py # JSON-RPC 报文拦截器（调试用）
│ │
│ ├── devices/ # 模拟设备层（物理仿真）
│ │ ├── base.py # 设备基类 + 全局注册表
│ │ ├── modbus_device.py # Modbus 模拟设备
│ │ ├── opcua_device.py # OPC UA 模拟设备
│ │ └── mqtt_device.py # MQTT 模拟设备
│ │
│ ├── protocols/ # 协议适配器层（统一接口）
│ │ ├── base.py # ProtocolAdapter 抽象基类 + 重试逻辑
│ │ ├── simulator.py # 仿真适配器（包装 BaseDevice）
│ │ ├── modbus.py # Modbus TCP 适配器（pymodbus）
│ │ └── mqtt.py # MQTT 适配器（paho-mqtt）
│ │
│ ├── models/ # Pydantic 数据模型
│ │ ├── device.py # 设备配置、寄存器映射
│ │ └── telemetry.py # 遥测数据（电机/泵/传感器）
│ │
│ └── utils/ # 工具
│ ├── logger.py # 结构化日志（stderr + 文件轮转）
│ └── logging.py # 日志配置
│
├── docs/ # 文档
│ ├── architecture.md # 架构设计（含 Mermaid 图）
│ ├── api.md # MCP API 参考文档
│ ├── claude-desktop-setup.md # Claude Desktop 集成指南
│ ├── debugging-guide.md # 调试指南
│ └── deployment.md # 部署指南（Docker / systemd）
│
└── examples/ # 示例
├── demo_simulation.py # 交互式设备仿真演示
├── mcp_client_demo.py # MCP 客户端编程示例
├── claude_desktop_config.json # Claude Desktop 配置模板
└── claude_desktop_config_debug.json # 调试配置模板
```
---
## 技术栈
| 类别 | 技术 | 用途 |
|------|------|------|
| **MCP 协议** | `mcp` SDK v2 (Python) | MCP 服务器实现 |
| **数据模型** | Pydantic v2 | 配置和遥测数据结构 |
| **工业协议** | pymodbus, opcua-asyncio, paho-mqtt | Modbus/OPC UA/MQTT 通信 |
| **配置** | YAML (PyYAML) | 声明式设备定义 |
| **日志** | structlog | 结构化日志（stderr + 文件） |
| **包管理** | uv (Astral) | 依赖管理和虚拟环境 |
| **代码质量** | ruff, mypy | Lint + 类型检查 |
| **测试** | pytest, pytest-asyncio | 异步测试框架 |
| **部署** | Docker, systemd | 容器化和系统服务 |
---
## 贡献指南
我们欢迎 Issue、PR 和任何形式的贡献！
### 开发环境搭建
```bash
git clone https://github.com/your-org/industrial-mcp.git
cd industrial-mcp
uv sync --dev
```
### 代码规范
```bash
uv run ruff check src/ # Lint
uv run ruff format src/ # 格式化
uv run mypy src/ # 类型检查
```
### 提交 PR 流程
1. Fork 本仓库
2. 创建功能分支 (`git checkout -b feat/amazing-feature`)
3. 提交更改 (`git commit -m 'Add amazing feature'`)
4. 推送到分支 (`git push origin feat/amazing-feature`)
5. 创建 Pull Request
### 添加新协议适配器
1. 在 `src/industrial_mcp/protocols/` 中创建新文件
2. 继承 `ProtocolAdapter` 抽象基类
3. 实现 `connect`, `disconnect`, `read_register`, `write_register`, `read_all`
4. 在 `DeviceManager._create_adapter()` 中注册
5. 在 `config/devices.yaml` 中添加配置示例
---
## 许可证
[MIT License](LICENSE) — 随意使用、修改和分发。
---
## 致谢
- [Model Context Protocol](https://modelcontextprotocol.io/) — Anthropic 提出的开放协议
- [pymodbus](https://github.com/pymodbus-dev/pymodbus) — Python Modbus 协议栈
- [FastMCP](https://gofastmcp.com/) — MCP 开发的灵感来源
- [uv](https://docs.astral.sh/uv/) — 极速 Python 包管理器
---
> 🤖 **industrial-mcp** — 让 AI 进入工厂车间，从这一行命令开始：
>
> ```bash
> npx @modelcontextprotocol/inspector uv run python -m industrial_mcp.server
> ```
This server cannot be deployed
Maintenance
ActivitySlowing
ResponsivenessNo issues