Hermes Agent + MES/IoT 智能制造实战：从部署到落地

一、前言

在智能制造领域，MES（制造执行系统） 和 IoT（物联网平台） 已经积累了海量数据，但数据孤岛问题依然严重：

  • 生产经理想查"3号车间今天的不良率"，需要在 MES 里翻报表、再去 IoT 平台查工艺参数，耗时 30 分钟
  • 设备告警响起，工程师要手动对比 MES 工单记录和 IoT 传感器数据，排查效率低下
  • 每天 18:00 的生产日报，需要专人从两个系统导出数据、做 Excel、发邮件

Hermes Agent 的出现改变了这一切。它是一个能自主思考、调用工具、持久记忆、定时执行的 AI 智能体。本文将手把手教你如何用 Docker Compose 部署 Hermes Agent，并与 MES/IoT 系统集成，实现对话式数据查询和自动化巡检。

二、Hermes Agent 是什么？

Hermes Agent 是由 Nous Research 开源的 AI 智能体框架，核心理念是 “An agent that grows with you”。

2.1 与传统 ChatGPT 的区别

特性 ChatGPT/Claude Hermes Agent
运行方式 网页/App 对话 部署在本地/服务器，7×24 运行
工具调用 受限于平台 通过 MCP 自由连接 MES/IoT/数据库
记忆能力 会话级 持久化记忆，跨会话保留
定时任务 ❌ 不支持 ✅ Cron 自动执行
技能沉淀 ❌ 不支持 ✅ 自动从经验生成 Skill
消息平台 官方客户端 飞书/钉钉/企微/Slack/Telegram

2.2 核心能力

  • Function Calling：LLM 自主决策调用哪些工具（MES 查工单、IoT 查传感器）
  • 持久记忆：记住用户偏好、设备基线、历史异常模式
  • Cron 定时任务：自动巡检、生成日报、异常告警
  • MCP 协议：标准化连接外部系统，即插即用

三、应用场景（MES + IoT）

场景 1：对话式数据查询（主动调用）

用户：“3号车间今天的不良率怎么样？”

Agent 执行：

  1. 调用 MES API → 查询今日质检数据
  2. 调用 IoT API → 获取对应时段工艺参数
  3. LLM 分析 → “今日不良率 2.3%，高于目标 1.5%。问题集中在 14:00-16:00，注塑机 #07 模温波动 ±8℃，建议检查加热系统。”

场景 2：设备健康自动巡检（定时排查）

每 2 小时自动执行：

  1. 读取 IoT 实时数据（温度/压力/振动）
  2. 对比基线阈值
  3. 发现异常 → 自动推送告警到飞书

场景 3：生产日报自动生成

每天 18:00：

  1. 汇总 MES 产量、达成率、不良率
  2. 汇总 IoT 设备 OEE、能耗
  3. 生成结构化报告，推送到管理群

场景 4：质量异常根因追溯

每小时扫描 MES 不合格品：

  1. 获取生产履历
  2. 关联 IoT 工艺参数曲线
  3. 自动定位偏差源，生成 8D 初稿

四、整体架构

┌─────────────────────────────────────────────────────────────┐
│                      Docker 宿主机                            │
│  ┌─────────────────┐    ┌──────────────────────────────┐   │
│  │  Hermes Agent    │    │         MCP 工具层            │   │
│  │  ┌───────────┐  │    │  ┌─────────┐  ┌───────────┐ │   │
│  │  │ Gateway   │  │◄───┤  │ MES API │  │ IoT API   │ │   │
│  │  │ (消息接入) │  │    │ │ Server  │  │ Server    │ │   │
│  │  ├───────────┤  │    │ └─────────┘  └───────────┘ │   │
│  │  │ Cron      │  │    │  ┌─────────┐  ┌───────────┐ │   │
│  │  │ (定时任务) │  │    │  │ Shell   │  │ 其他工具   │ │   │
│  │  ├───────────┤  │    │ │ Tool    │  │           │ │   │
│  │  │ Dashboard │  │    │ └─────────┘  └───────────┘ │   │
│  │  │ (Web管理)  │  │    └──────────────────────────────┘   │
│  │  └───────────┘  │                                       │
│  └─────────────────┘                                       │
│  ┌─────────────────┐                                       │
│  │  API Gateway     │  ← 前端主入口（FastAPI:8080）        │
│  │  (FastAPI)       │                                       │
│  └─────────────────┘                                       │
└─────────────────────────────────────────────────────────────┘
         │                    │                    │
    ┌────▼────┐         ┌────▼────┐          ┌────▼────┐
    │  飞书/   │         │   MES   │          │   IoT   │
    │ 钉钉/企微│         │10.86.3.32│          │10.86.3.33│
    └─────────┘         └─────────┘          └─────────┘

服务说明：

服务 端口 职责
hermes 26429 / 6430 消息平台接入、Cron 定时任务、Dashboard
api-gateway 8080 前端主入口，接收 HTTP 请求 → 调用 Tools → LLM 推理 → 返回结果

五、环境准备

5.1 硬件要求

  • CPU：2 核+
  • 内存：4GB+
  • 磁盘：20GB+
  • 网络：能访问 MES/IoT 内网 + 公网（调用 LLM API）

5.2 软件要求

  • Docker 24.0+
  • Docker Compose 2.20+
  • LLM API Key（OpenAI / OpenRouter / 阿里云百炼等）

5.3 安装 Docker

# Ubuntu/Debian
curl -fsSL https://get.docker.com | sh
sudo usermod -aG docker $USER
# 退出重新登录

六、完整部署

6.1 目录结构

创建项目目录：

mkdir -p ~/hermes-mes-iot/{api-gateway,tools}
cd ~/hermes-mes-iot

最终结构：

hermes-mes-iot/
├── docker-compose.yml          # Docker 编排
├── .env                        # 环境变量（敏感信息）
├── mcp-config.json             # MCP 工具配置
├── cron-tasks.yaml             # 定时任务定义
├── api-gateway/
│   ├── Dockerfile              # API 网关镜像
│   ├── requirements.txt        # Python 依赖
│   └── main.py                 # FastAPI 主程序 ⭐核心
└── tools/
    ├── mes_mcp_server.py       # MES 适配器
    └── iot_mcp_server.py     # IoT 适配器

6.2 docker-compose.yml

version: "3.8"

services:
  hermes:
    image: nousresearch/hermes-agent:latest
    container_name: hermes-agent
    restart: unless-stopped
    command: >
      sh -c "hermes gateway run &
             hermes cron run"
    ports:
      - "26429:26429"
      - "26430:6430"
    volumes:
      - ./hermes-data:/opt/data
      - ./mcp-config.json:/opt/data/mcp-config.json:ro
      - ./cron-tasks.yaml:/opt/data/cron-tasks.yaml:ro
      - ./tools:/opt/tools:ro
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - OPENAI_BASE_URL=${OPENAI_BASE_URL:-https://api.openai.com/v1}
      - HERMES_MODEL=${HERMES_MODEL:-gpt-4o}
      - FEISHU_APP_ID=${FEISHU_APP_ID:-}
      - FEISHU_APP_SECRET=${FEISHU_APP_SECRET:-}
      - FEISHU_DOMAIN=${FEISHU_DOMAIN:-feishu}
      - FEISHU_CONNECTION_MODE=${FEISHU_CONNECTION_MODE:-websocket}
      - DINGTALK_APP_KEY=${DINGTALK_APP_KEY:-}
      - DINGTALK_APP_SECRET=${DINGTALK_APP_SECRET:-}
      - WECHAT_CORP_ID=${WECHAT_CORP_ID:-}
      - WECHAT_AGENT_ID=${WECHAT_AGENT_ID:-}
      - WECHAT_SECRET=${WECHAT_SECRET:-}
      - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN:-}
      - HERMES_MCP_CONFIG=/opt/data/mcp-config.json
      - HERMES_CRON_CONFIG=/opt/data/cron-tasks.yaml
      - HERMES_DASHBOARD=1
      - HERMES_DASHBOARD_HOST=0.0.0.0
      - HERMES_DASHBOARD_PORT=6430
      - HERMES_DASHBOARD_INSECURE=${HERMES_DASHBOARD_INSECURE:-1}
      - HERMES_DASHBOARD_BASIC_AUTH_USERNAME=${DASHBOARD_USER:-admin}
      - HERMES_DASHBOARD_BASIC_AUTH_PASSWORD=${DASHBOARD_PASS:-admin123}
      - TZ=Asia/Shanghai
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 4G
    networks:
      - hermes-net

  api-gateway:
    build:
      context: ./api-gateway
      dockerfile: Dockerfile
    container_name: hermes-api-gateway
    restart: unless-stopped
    ports:
      - "8080:8080"
    volumes:
      - ./tools:/opt/tools:ro
      - ./api-gateway:/app
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - OPENAI_BASE_URL=${OPENAI_BASE_URL:-https://api.openai.com/v1}
      - LLM_MODEL=${HERMES_MODEL:-gpt-4o}
      - MES_API_URL=${MES_API_URL}
      - MES_API_KEY=${MES_API_KEY:-}
      - IOT_API_URL=${IOT_API_URL}
      - IOT_API_KEY=${IOT_API_KEY:-}
      - TZ=Asia/Shanghai
    networks:
      - hermes-net
    depends_on:
      - hermes
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 30s
      timeout: 5s
      retries: 3

networks:
  hermes-net:
    driver: bridge

6.3 .env 环境变量

# --- LLM 配置 ---
OPENAI_API_KEY=sk-your-api-key-here
OPENAI_BASE_URL=https://api.openai.com/v1
HERMES_MODEL=gpt-4o

# --- MES 系统 ---
MES_API_URL=http://10.86.3.32:91/mes
MES_API_KEY=your-mes-api-key

# --- IoT 平台 ---
IOT_API_URL=http://10.86.3.33:80/iot
IOT_API_KEY=your-iot-api-key

# --- 消息平台（按需）---
FEISHU_APP_ID=
FEISHU_APP_SECRET=
FEISHU_DOMAIN=feishu
FEISHU_CONNECTION_MODE=websocket
DINGTALK_APP_KEY=
DINGTALK_APP_SECRET=
WECHAT_CORP_ID=
WECHAT_AGENT_ID=
WECHAT_SECRET=
TELEGRAM_BOT_TOKEN=

# --- Dashboard 安全 ---
HERMES_DASHBOARD_INSECURE=1
DASHBOARD_USER=admin
DASHBOARD_PASS=ChangeMe123!

⚠️ 生产环境：务必修改 DASHBOARD_PASS，并设置 HERMES_DASHBOARD_INSECURE=0。

6.4 MCP 配置（mcp-config.json）

{
  "mcpServers": {
    "mes-api": {
      "command": "python3",
      "args": ["/opt/tools/mes_mcp_server.py"],
      "env": {
        "MES_API_URL": "${MES_API_URL}",
        "MES_API_KEY": "${MES_API_KEY}"
      },
      "description": "MES 制造执行系统 API"
    },
    "iot-api": {
      "command": "python3",
      "args": ["/opt/tools/iot_mcp_server.py"],
      "env": {
        "IOT_API_URL": "${IOT_API_URL}",
        "IOT_API_KEY": "${IOT_API_KEY}"
      },
      "description": "IoT 物联网平台 API"
    },
    "shell": {
      "command": "hermes",
      "args": ["mcp", "shell"],
      "description": "本地 Shell 命令执行"
    }
  }
}

6.5 API Gateway（前端接口）⭐核心

api-gateway/Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
EXPOSE 8080
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]
api-gateway/requirements.txt
fastapi>=0.110.0
uvicorn[standard]>=0.29.0
openai>=1.30.0
pydantic>=2.7.0
python-multipart>=0.0.9
api-gateway/main.py（完整版）
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Hermes API Gateway - 前端 HTTP 接口服务
架构：前端 → API Gateway → 调用 Tools (MES/IoT) → 调用 LLM → 返回结果
"""

import os
import sys
import json
import time
import asyncio
from typing import Optional, List, Dict, Any
from datetime import datetime

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import openai

# ==================== 配置 ====================
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
OPENAI_BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1")
LLM_MODEL = os.environ.get("LLM_MODEL", "gpt-4o")

MES_API_URL = os.environ.get("MES_API_URL", "http://10.86.3.32:91/mes")
MES_API_KEY = os.environ.get("MES_API_KEY", "")
IOT_API_URL = os.environ.get("IOT_API_URL", "http://10.86.3.33:80/iot")
IOT_API_KEY = os.environ.get("IOT_API_KEY", "")

client = openai.AsyncOpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)

app = FastAPI(
    title="Hermes MES/IoT API Gateway",
    description="前端调用 AI 的统一接口，支持 MES/IoT 数据查询与智能分析",
    version="1.0.0"
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ==================== 数据模型 ====================
class ChatRequest(BaseModel):
    message: str = Field(..., description="用户输入的消息/问题")
    session_id: Optional[str] = Field(default=None, description="会话ID，用于保持上下文")
    context: Optional[Dict[str, Any]] = Field(default=None, description="额外上下文信息")

class QueryRequest(BaseModel):
    endpoint: str = Field(..., description="API 端点路径")
    method: str = Field(default="GET", description="HTTP 方法")
    params: Optional[Dict[str, Any]] = Field(default=None, description="查询参数")
    body: Optional[Dict[str, Any]] = Field(default=None, description="POST/PUT 请求体")

class InspectionRequest(BaseModel):
    task_type: str = Field(..., description="巡检类型：equipment_health / quality_traceback / energy_audit / oee_analysis")
    params: Optional[Dict[str, Any]] = Field(default=None, description="巡检参数")

class ApiResponse(BaseModel):
    success: bool
    data: Any
    message: str
    timestamp: str
    elapsed_ms: int

# ==================== Tool 调用层 ====================
import urllib.request
import urllib.parse
import urllib.error

def call_mes_api(endpoint: str, method: str = "GET", params: dict = None, body: dict = None):
    url = f"{MES_API_URL}/{endpoint}"
    if params:
        url += "?" + urllib.parse.urlencode(params)
    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {MES_API_KEY}"}
    req = urllib.request.Request(url, headers=headers, method=method)
    if body:
        req.data = json.dumps(body).encode("utf-8")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        return {"error": f"MES API HTTP {e.code}: {e.reason}", "details": e.read().decode("utf-8", errors="ignore")}
    except Exception as e:
        return {"error": str(e)}

def call_iot_api(endpoint: str, method: str = "GET", params: dict = None, body: dict = None):
    url = f"{IOT_API_URL}/{endpoint}"
    if params:
        url += "?" + urllib.parse.urlencode(params)
    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {IOT_API_KEY}"}
    req = urllib.request.Request(url, headers=headers, method=method)
    if body:
        req.data = json.dumps(body).encode("utf-8")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        return {"error": f"IoT API HTTP {e.code}: {e.reason}", "details": e.read().decode("utf-8", errors="ignore")}
    except Exception as e:
        return {"error": str(e)}

# ==================== LLM Function Calling 定义 ====================
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "mes_query",
            "description": "查询 MES 制造执行系统数据，支持工单、设备、质量、排程等",
            "parameters": {
                "type": "object",
                "properties": {
                    "endpoint": {
                        "type": "string",
                        "description": "MES API 端点，如 workorders, equipment/status, quality/inspection, production/summary"
                    },
                    "method": {"type": "string", "enum": ["GET", "POST", "PUT"], "description": "HTTP 方法"},
                    "params": {"type": "object", "description": "查询参数"}
                },
                "required": ["endpoint"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "iot_query",
            "description": "查询 IoT 物联网平台数据，支持设备实时数据、历史数据、告警、能耗、OEE等",
            "parameters": {
                "type": "object",
                "properties": {
                    "endpoint": {
                        "type": "string",
                        "description": "IoT API 端点，如 devices/realtime, devices/history, alarms, energy/consumption, analytics/oee"
                    },
                    "method": {"type": "string", "enum": ["GET", "POST"], "description": "HTTP 方法"},
                    "params": {"type": "object", "description": "查询参数"}
                },
                "required": ["endpoint"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取当前时间，用于需要日期/时间上下文的查询",
            "parameters": {"type": "object", "properties": {}, "required": []}
        }
    }
]

async def execute_tool(name: str, arguments: dict) -> dict:
    if name == "mes_query":
        return call_mes_api(
            endpoint=arguments.get("endpoint"),
            method=arguments.get("method", "GET"),
            params=arguments.get("params")
        )
    elif name == "iot_query":
        return call_iot_api(
            endpoint=arguments.get("endpoint"),
            method=arguments.get("method", "GET"),
            params=arguments.get("params")
        )
    elif name == "get_current_time":
        return {"datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "date": datetime.now().strftime("%Y-%m-%d")}
    else:
        return {"error": f"Unknown tool: {name}"}

async def agent_chat(user_message: str, conversation_history: list = None) -> dict:
    if conversation_history is None:
        conversation_history = []

    messages = [
        {
            "role": "system",
            "content": """你是一位智能制造领域的 AI 助手，精通 MES 和 IoT 数据分析。
你的职责：
1. 理解用户的生产管理问题
2. 调用 MES/IoT 工具获取实时/历史数据
3. 基于数据进行分析和推理
4. 给出专业、简洁、可执行的建议

重要规则：
- 调用工具时，endpoint 必须是实际存在的 API 路径
- MES 常用端点：workorders（工单）、equipment/status（设备状态）、quality/inspection（质检）、production/summary（生产汇总）、schedule（排程）
- IoT 常用端点：devices/realtime（实时数据）、devices/history（历史数据）、alarms（告警）、energy/consumption（能耗）、analytics/oee（OEE）
- 如果用户问"今天"，先调用 get_current_time 获取当前日期
- 回答使用中文，数据要准确，建议要具体"""
        }
    ] + conversation_history + [
        {"role": "user", "content": user_message}
    ]

    response = await client.chat.completions.create(
        model=LLM_MODEL,
        messages=messages,
        tools=TOOLS,
        tool_choice="auto",
        temperature=0.3
    )

    assistant_message = response.choices[0].message

    if assistant_message.tool_calls:
        messages.append({
            "role": "assistant",
            "content": assistant_message.content or "",
            "tool_calls": [tc.model_dump() for tc in assistant_message.tool_calls]
        })

        for tool_call in assistant_message.tool_calls:
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)
            print(f"[Tool Call] {tool_name}({json.dumps(tool_args, ensure_ascii=False)})")
            tool_result = await execute_tool(tool_name, tool_args)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(tool_result, ensure_ascii=False)
            })

        final_response = await client.chat.completions.create(
            model=LLM_MODEL,
            messages=messages,
            temperature=0.3
        )

        reply = final_response.choices[0].message.content
        tool_calls_log = [{"tool": tc.function.name, "arguments": json.loads(tc.function.arguments)} for tc in assistant_message.tool_calls]

        return {"reply": reply, "tool_calls": tool_calls_log, "model": LLM_MODEL}
    else:
        return {"reply": assistant_message.content, "tool_calls": [], "model": LLM_MODEL}

# ==================== API 路由 ====================
@app.get("/health")
async def health_check():
    return {"status": "ok", "service": "hermes-api-gateway", "timestamp": datetime.now().isoformat()}

@app.post("/api/chat", response_model=ApiResponse)
async def chat(request: ChatRequest):
    start = time.time()
    try:
        result = await agent_chat(request.message)
        elapsed = int((time.time() - start) * 1000)
        return ApiResponse(success=True, data=result, message="success", timestamp=datetime.now().isoformat(), elapsed_ms=elapsed)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/query/mes", response_model=ApiResponse)
async def query_mes(request: QueryRequest):
    start = time.time()
    try:
        data = call_mes_api(request.endpoint, request.method, request.params, request.body)
        elapsed = int((time.time() - start) * 1000)
        return ApiResponse(success="error" not in data, data=data, message="success" if "error" not in data else data.get("error"), timestamp=datetime.now().isoformat(), elapsed_ms=elapsed)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/query/iot", response_model=ApiResponse)
async def query_iot(request: QueryRequest):
    start = time.time()
    try:
        data = call_iot_api(request.endpoint, request.method, request.params, request.body)
        elapsed = int((time.time() - start) * 1000)
        return ApiResponse(success="error" not in data, data=data, message="success" if "error" not in data else data.get("error"), timestamp=datetime.now().isoformat(), elapsed_ms=elapsed)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/inspection", response_model=ApiResponse)
async def inspection(request: InspectionRequest):
    start = time.time()

    task_prompts = {
        "equipment_health": """请执行设备健康巡检：
1. 查询 IoT 平台获取指定车间所有关键设备的实时数据
2. 查询 MES 获取设备基线参数和维护记录
3. 识别异常参数，生成巡检报告并给出处理建议""",

        "quality_traceback": """请执行质量异常追溯：
1. 查询 MES 过去24小时的不合格品记录
2. 获取生产履历和 IoT 工艺参数，定位根因
3. 生成 8D 初稿格式的分析报告""",

        "energy_audit": """请执行能耗审计：
1. 查询 IoT 平台过去24小时全厂能耗数据
2. 查询 MES 对应产量，计算单件能耗
3. 识别异常产线/设备，给出节能建议""",

        "oee_analysis": """请执行 OEE 分析：
1. 查询 MES 过去一周的工单完成情况
2. 查询 IoT 设备运行数据
3. 计算各设备 OEE，分析趋势，识别 TOP3 改进点"""
    }

    if request.task_type not in task_prompts:
        raise HTTPException(status_code=400, detail=f"不支持的巡检类型: {request.task_type}")

    prompt = task_prompts[request.task_type]
    if request.params:
        prompt += f"\n\n巡检参数: {json.dumps(request.params, ensure_ascii=False)}"

    try:
        result = await agent_chat(prompt)
        elapsed = int((time.time() - start) * 1000)
        return ApiResponse(success=True, data=result, message="success", timestamp=datetime.now().isoformat(), elapsed_ms=elapsed)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/")
async def root():
    return {
        "service": "Hermes MES/IoT API Gateway",
        "version": "1.0.0",
        "endpoints": {
            "chat": "POST /api/chat - 通用智能对话",
            "query_mes": "POST /api/query/mes - 直接查询 MES",
            "query_iot": "POST /api/query/iot - 直接查询 IoT",
            "inspection": "POST /api/inspection - 执行巡检任务",
            "health": "GET /health - 健康检查"
        },
        "mes_address": MES_API_URL,
        "iot_address": IOT_API_URL
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

6.6 MES 适配器（tools/mes_mcp_server.py）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""MES MCP Server - 适配 MES 系统 (10.86.3.32:91/mes)"""

import os
import sys
import json
import urllib.request
import urllib.parse
import urllib.error

MES_API_URL = os.environ.get("MES_API_URL", "http://10.86.3.32:91/mes")
MES_API_KEY = os.environ.get("MES_API_KEY", "")

def call_mes_api(endpoint: str, method: str = "GET", data: dict = None, params: dict = None):
    url = f"{MES_API_URL}/{endpoint}"
    if params:
        url += "?" + urllib.parse.urlencode(params)
    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {MES_API_KEY}"}
    req = urllib.request.Request(url, headers=headers, method=method)
    if data:
        req.data = json.dumps(data, ensure_ascii=False).encode("utf-8")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        return {"error": f"HTTP {e.code}", "message": e.reason}
    except Exception as e:
        return {"error": str(e)}

def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            req = json.loads(line)
            method = req.get("method", "")
            params = req.get("params", {})
            result = None

            if method == "get_work_orders":
                result = call_mes_api("workorders", params=params)
            elif method == "get_work_order_detail":
                wo_id = params.get("work_order_id")
                result = call_mes_api(f"workorders/{wo_id}")
            elif method == "get_equipment_status":
                result = call_mes_api("equipment/status", params=params)
            elif method == "get_equipment_list":
                result = call_mes_api("equipment", params=params)
            elif method == "get_quality_data":
                result = call_mes_api("quality/inspection", params=params)
            elif method == "get_quality_defects":
                result = call_mes_api("quality/defects", params=params)
            elif method == "get_production_summary":
                result = call_mes_api("production/summary", params=params)
            elif method == "get_production_output":
                result = call_mes_api("production/output", params=params)
            elif method == "get_schedule":
                result = call_mes_api("schedule", params=params)
            elif method == "get_maintenance_records":
                eq_id = params.get("equipment_id")
                result = call_mes_api(f"equipment/{eq_id}/maintenance")
            elif method == "create_work_order":
                result = call_mes_api("workorders", method="POST", data=params)
            elif method == "update_schedule":
                result = call_mes_api("schedule", method="PUT", data=params)
            else:
                result = {"error": f"Unknown method: {method}", "available_methods": [
                    "get_work_orders", "get_work_order_detail", "get_equipment_status",
                    "get_equipment_list", "get_quality_data", "get_quality_defects",
                    "get_production_summary", "get_production_output", "get_schedule",
                    "get_maintenance_records", "create_work_order", "update_schedule"
                ]}

            print(json.dumps({"result": result}, ensure_ascii=False), flush=True)
        except Exception as e:
            print(json.dumps({"error": str(e)}, ensure_ascii=False), flush=True)

if __name__ == "__main__":
    main()

⚠️ 注意：上述 endpoint（如 workorders、equipment/status）是通用示例，请根据你们实际 MES 系统的接口文档修改。

6.7 IoT 适配器（tools/iot_mcp_server.py）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""IoT MCP Server - 适配 IoT 平台 (10.86.3.33:80/iot)"""

import os
import sys
import json
import urllib.request
import urllib.parse
import urllib.error

IOT_API_URL = os.environ.get("IOT_API_URL", "http://10.86.3.33:80/iot")
IOT_API_KEY = os.environ.get("IOT_API_KEY", "")

def call_iot_api(endpoint: str, method: str = "GET", data: dict = None, params: dict = None):
    url = f"{IOT_API_URL}/{endpoint}"
    if params:
        url += "?" + urllib.parse.urlencode(params)
    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {IOT_API_KEY}"}
    req = urllib.request.Request(url, headers=headers, method=method)
    if data:
        req.data = json.dumps(data, ensure_ascii=False).encode("utf-8")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        return {"error": f"HTTP {e.code}", "message": e.reason}
    except Exception as e:
        return {"error": str(e)}

def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            req = json.loads(line)
            method = req.get("method", "")
            params = req.get("params", {})
            result = None

            if method == "get_device_realtime":
                result = call_iot_api("devices/realtime", params=params)
            elif method == "get_device_history":
                result = call_iot_api("devices/history", params=params)
            elif method == "get_device_list":
                result = call_iot_api("devices", params=params)
            elif method == "get_device_status":
                result = call_iot_api("devices/status", params=params)
            elif method == "get_device_detail":
                dev_id = params.get("device_id")
                result = call_iot_api(f"devices/{dev_id}")
            elif method == "get_alarms":
                result = call_iot_api("alarms", params=params)
            elif method == "get_alarms_active":
                result = call_iot_api("alarms/active", params=params)
            elif method == "get_energy_data":
                result = call_iot_api("energy/consumption", params=params)
            elif method == "get_energy_summary":
                result = call_iot_api("energy/summary", params=params)
            elif method == "get_oee_data":
                result = call_iot_api("analytics/oee", params=params)
            elif method == "get_telemetry":
                result = call_iot_api("telemetry", params=params)
            elif method == "send_command":
                result = call_iot_api("commands", method="POST", data=params)
            else:
                result = {"error": f"Unknown method: {method}", "available_methods": [
                    "get_device_realtime", "get_device_history", "get_device_list",
                    "get_device_status", "get_device_detail", "get_alarms",
                    "get_alarms_active", "get_energy_data", "get_energy_summary",
                    "get_oee_data", "get_telemetry", "send_command"
                ]}

            print(json.dumps({"result": result}, ensure_ascii=False), flush=True)
        except Exception as e:
            print(json.dumps({"error": str(e)}, ensure_ascii=False), flush=True)

if __name__ == "__main__":
    main()

⚠️ 注意：上述 endpoint（如 devices/realtime、alarms）是通用示例，请根据你们实际 IoT 平台的接口文档修改。

6.8 定时任务配置（cron-tasks.yaml）

cron_jobs:
  - name: daily_production_report
    schedule: "0 18 * * *"
    timezone: "Asia/Shanghai"
    description: "每天18:00自动汇总MES生产数据，生成日报"
    prompt: >
      请执行以下任务：
      1. 调用 mes-api 查询今日所有产线产量、达成率、不良率
      2. 调用 iot-api 获取今日设备OEE、故障时长、能耗
      3. 对比昨日数据，标注异常波动（±10%以上）
      4. 生成结构化日报并推送给生产管理团队
    channel: feishu
    target_group: "生产管理群"
    enabled: true

  - name: equipment_health_check
    schedule: "0 */2 * * *"
    timezone: "Asia/Shanghai"
    description: "每2小时巡检关键设备健康状态"
    prompt: >
      请执行设备健康巡检：
      1. 通过 iot-api 获取关键设备实时数据
      2. 对比基线阈值，识别异常参数
      3. 如发现异常，调用 mes-api 查询维护记录并推送告警
    channel: feishu
    target_group: "设备维护群"
    enabled: true

  - name: quality_traceback
    schedule: "0 * * * *"
    timezone: "Asia/Shanghai"
    description: "每小时检查MES质检异常，自动追溯根因"
    prompt: >
      请执行质量异常追溯：
      1. 调用 mes-api 查询过去1小时不合格品记录
      2. 如存在，获取生产履历 + IoT工艺参数，生成8D初稿并推送
    channel: feishu
    target_group: "质量管理群"
    enabled: true

  - name: energy_consumption_audit
    schedule: "0 1 * * *"
    timezone: "Asia/Shanghai"
    description: "每日凌晨分析能耗数据"
    prompt: >
      请执行能耗审计：
      1. 调用 iot-api 获取昨日全厂能耗
      2. 调用 mes-api 获取产量，计算单件能耗
      3. 识别异常并生成节能建议
    channel: feishu
    target_group: "能源管理群"
    enabled: true

  - name: weekly_oee_analysis
    schedule: "0 8 * * 1"
    timezone: "Asia/Shanghai"
    description: "每周一早上生成OEE分析报告"
    prompt: >
      请生成上周OEE分析：
      1. 调用 mes-api 获取上周工单完成情况
      2. 调用 iot-api 获取设备运行数据
      3. 计算OEE，分析趋势，识别TOP3改进点
    channel: feishu
    target_group: "生产管理群"
    enabled: true

七、启动部署

cd ~/hermes-mes-iot

# 1. 构建并启动
docker compose up -d --build

# 2. 查看状态
docker compose ps

# 3. 查看日志
docker compose logs -f api-gateway
docker compose logs -f hermes

# 4. 健康检查
curl http://localhost:8080/health

# 5. 测试对话
curl -X POST http://localhost:8080/api/chat \
  -H "Content-Type: application/json" \
  -d '{"message":"测试连接，MES和IoT地址是否正确配置？"}'

八、前端调用实战

8.1 通用智能对话（推荐）

const response = await fetch('http://your-server:8080/api/chat', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    message: "3号车间今天的不良率怎么样？",
    session_id: "user_001"
  })
});

const result = await response.json();
console.log(result.data.reply);       // AI 分析回复
console.log(result.data.tool_calls);  // 实际调用了哪些工具

Agent 内部执行流程：

用户输入: "3号车间今天的不良率怎么样？"
    ↓
LLM 决策 → 调用 get_current_time()
    ↓
LLM 决策 → 调用 mes_query(endpoint="quality/inspection", params={date:"2026-07-16", workshop:"3号车间"})
    ↓
LLM 决策 → 调用 iot_query(endpoint="devices/realtime", params={device_ids:["injection_07"]})
    ↓
LLM 分析数据 → 生成回复
    ↓
返回: "3号车间今日不良率 2.3%，高于目标 1.5%。问题集中在 14:00-16:00，
      注塑机 #07 模温波动 ±8℃，建议检查加热系统。"

8.2 直接查询原始数据

// 查 MES 工单
fetch('http://your-server:8080/api/query/mes', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    endpoint: "workorders",
    method: "GET",
    params: { date: "2026-07-16", status: "in_progress" }
  })
});

// 查 IoT 实时数据
fetch('http://your-server:8080/api/query/iot', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    endpoint: "devices/realtime",
    params: { device_ids: ["injection_01", "injection_02"] }
  })
});

8.3 执行巡检任务

fetch('http://your-server:8080/api/inspection', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    task_type: "equipment_health",  // equipment_health / quality_traceback / energy_audit / oee_analysis
    params: { workshop: "3号车间" }
  })
});

九、接口文档

启动后访问自动生成的 Swagger UI：

http://your-server:8080/docs
接口 方法 说明
/api/chat POST 通用智能对话（前端主入口）
/api/query/mes POST 直接查询 MES（绕过 LLM）
/api/query/iot POST 直接查询 IoT（绕过 LLM）
/api/inspection POST 执行预设巡检任务
/health GET 健康检查

参考资源

如果本文对你有帮助，欢迎 点赞 👍 收藏 ⭐ 关注 🔖，有问题请在评论区留言！

Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

当 AI 修图师装上“万能插头“：MCP 如何改变修图工作流

avatar AI编程社区
cover

2026年8月开发者实战指南：ChatGPT Plus / Pro + Codex 如何协同完成需求分析、编码、测试、调试与代码审查

avatar AI编程社区
cover

投资一台B300，部署Claude Opus 5多久才能回本？

avatar AI编程社区