kc iot gateway

“读取出厂温度”——物联网网关

    

所以，你有一个满是说MQTT的传感器的工厂，只理解Modbus的PLC，一些坚持CoAP的设备，还有一个供应商说“只需向我们发送一个webhook”。恭喜你，你现在需要一个网关。这就是网关——一个基于插件的IoT网关，它在一个REST API后面争论所有这些，所以堆栈的其余部分不必关心。

它还配备了一个由YAML驱动的规则引擎（因为没有人想仅仅为了改变温度阈值而重新部署）、一个实时web仪表板、一个MCP服务器，这样AI代理就可以加入其中，以及Docker Compose一键部署，适合那些重视自己理智的人。

诞生于运行一个生产物联网平台的疤痕组织，该平台包含6种协议和10多个品牌的28个设备插件。这是经过提炼的版本——相同的架构，更少的噩梦。

______________________________________________________________________

盒子里是什么

• 多协议支持 --MQTT、Modbus-TCP、CoAP、Webhook——全部通过一个API。你的前端开发人员不需要知道Modbus是什么。（幸运的是他们。）
• 插件架构 --通过删除一个.py文件来添加新协议。零核心变化。说真的。
• YAML设备配置文件 --在一个配置文件中定义设备、字段和数据类型，而不是编写另一个适配器类
• 规则引擎 --YAML定义了具有冷却时间、严重性级别和跨设备自动化的警报规则。因为凌晨3点的警报风暴会塑造角色，但只有一次。
• 多通道警报 --电报、Webhook和设备到设备控制
• 实时仪表盘 --基于WebSocket的web UI，具有实时设备数据和警报历史记录。没有React，没有构建步骤，只是工作。
• Webhook模拟器 --内置web UI，无需使用curl或Postman即可测试webhook设备
• MCP服务器 --让AI代理通过自然语言读/写设备。无论我们是否准备好，未来就在这里。
• AI代理技能 --OpenClaw和其他LLM代理的CLI包装器
• Docker就绪 -- docker compose up -d 启动一切，包括模拟器。去喝咖啡。

______________________________________________________________________

安全通知： 这是一个为可信局域网环境设计的POC/开发项目。REST API、WebSocket、MQTT代理和webhook端点不实现身份验证。在没有额外安全措施的情况下，不要将服务端口暴露给公共互联网。

快速开始

本地开发

git clone https://github.com/KerberosClaw/kc_iot_gateway.git
cd kc_iot_gateway
uv sync

# Start simulators
uv run python simulators/modbus_simulator.py &

# Start gateway
uv run python -m src
# Dashboard: http://localhost:8000

Docker编写（简单的方法）

docker compose up -d
open http://localhost:8000

这引发了整个马戏团：

• Mosquitto MQTT代理（端口1883）
• MQTT传感器模拟器
• Modbus PLC模拟器（端口5020）
• 网关+仪表板（端口8000）

______________________________________________________________________

建筑

graph TB
    Agent["AI Agent
(Claude / OpenClaw)"] |MCP| MCP["MCP Server"]
    MCP  Core
    Client["Dashboard / curl"] |"REST API + WS"| Core

    subgraph Core["Gateway Core"]
        direction LR
        Registry["Device Registry"] --- EventBus["Event Bus"]
        EventBus --- RuleEngine["Rule Engine"]
    end

    subgraph Plugins
        MQTT["MQTT"] --- Modbus["Modbus"]
        Modbus --- CoAP["CoAP"]
        CoAP --- Webhook["Webhook"]
    end

    subgraph Actions
        TG["Telegram"] --- WH["Webhook"]
        WH --- DW["Device Write"]
    end

    Core  Plugins
    RuleEngine --> Actions
    Plugins  Devices["IoT Devices"]
    Actions --> Notify["Notifications"]

______________________________________________________________________

设备配置文件（YAML）

在中告诉网关您的设备 devices.yaml。无需代码——只需描述您拥有什么以及如何与之交互：

plugins:
  mqtt_sensor:
    protocol: mqtt
    broker: localhost:1883
    devices:
      - id: factory_temp_01
        name: "Factory temperature sensor"
        topic: factory/sensor/temp_01
        fields:
          temperature: { path: "$.temp", unit: "°C", type: float }
          humidity: { path: "$.hum", unit: "%RH", type: float }

  modbus_plc:
    protocol: modbus
    host: localhost
    port: 5020
    slave_id: 1
    devices:
      - id: plc_01
        name: "Production PLC"
        registers:
          motor_speed: { address: 4, type: uint16, unit: "RPM", access: rw }
          temperature: { address: 0, type: float32, unit: "°C", access: ro }

  webhook_devices:
    protocol: webhook
    listen_path: /webhook
    devices:
      - id: env_sensor_01
        name: "Environment sensor (Vendor A)"
        identity:
          field: "$.device_id"
          value: "ENV-001"
        fields:
          temperature: { path: "$.data.temp", unit: "°C", type: float }

______________________________________________________________________

警报规则（YAML）

定义你的“着火时请叫醒我”规则 rules.yaml:

rules:
  - name: high_temperature
    device: factory_temp_01
    condition:
      field: temperature
      operator: ">"
      threshold: 40
    severity: critical
    cooldown: 300
    actions:
      - type: telegram
        message: "[ALERT]{device_name} temperature {value}°C"

  - name: pump_auto_control
    device: plc_01
    condition:
      field: temperature
      operator: ">"
      threshold: 35
    actions:
      - type: device_write
        target_device: plc_01
        params: { pump_on: true }

规则支持：

• 冷却 --因为连续收到47条相同的Telegram消息不是“监控”，而是垃圾邮件
• 跨设备自动化 --传感器显示温度很高，泵打开。回路中没有人，也没有人在桌子上睡着。
• 运行时修改 --在不重新启动的情况下通过REST API更新规则。无需重新部署的冷汗，即可更改生产阈值。

______________________________________________________________________

REST API

______________________________________________________________________

MCP 服务器

AI代理可以通过MCP控制您的所有设备。他们没有请求许可，老实说，我们也没有：

______________________________________________________________________

项目结构

kc_iot_gateway/
├── src/
│   ├── gateway.py            # Core: startup, plugin loader, event bus
│   ├── plugin_base.py        # DevicePlugin ABC
│   ├── registry.py           # Device registry (in-memory state)
│   ├── api.py                # REST API (FastAPI)
│   ├── mcp_server.py         # MCP Server (FastMCP)
│   ├── rules.py              # Rule engine
│   ├── cooldown.py           # Cooldown manager
│   ├── db.py                 # SQLite (rules + alert history)
│   ├── plugins/
│   │   ├── mqtt_plugin.py
│   │   ├── modbus_plugin.py
│   │   ├── coap_plugin.py
│   │   └── webhook_plugin.py
│   └── actions/
│       ├── telegram.py
│       ├── webhook.py
│       ├── device_write.py
│       └── console.py
├── static/
│   └── index.html            # Dashboard + Webhook Simulator
├── simulators/
│   ├── mqtt_simulator.py
│   └── modbus_simulator.py
├── ai_agent_skill/           # AI agent CLI wrapper
├── tests/                    # Automated tests
├── docs/
│   └── DESIGN.md             # Design document
├── devices.yaml
├── rules.yaml
├── docker-compose.yml
└── Dockerfile

______________________________________________________________________

环境变量

______________________________________________________________________

TODO（又名“我会做到的”）

• \[\]和/或复合规则条件
• \[\]插件热重新加载，无需重新启动