MES-MCP Server
把 OTD MES/WMS 的车间执行与仓储物流数据,暴露成 AI Agent 可直接调用的 MCP 工具。
与 ERP-MCP、CRM-MCP 同栈同范式,三者装齐即可跑通完整制造链路: 商机 → 合同 → 销售订单 → 生产工单 → 工序/报工 → 质检 → 入库 → 发货 → 回款。
MES-MCP 提供 ERP/CRM 给不了的三样东西:
- 工序级颗粒度 —— ERP 只知道工单"完工 60%",MES 知道卡在哪道工序、哪条线、哪台设备
- SN 级正反向追溯 —— 成品序列号 ↔ 用料批次双向族谱,质量事故圈定召回范围
- 实时车间状态 —— 在制分布、良率、设备报警、齐套率,分钟级而非日结级
当前进度:43 个工具可用(24 原子查询 + 8 聚合 + 5 复合 Skill + 4 元数据 + 2 版本管理)。 完整设计见
MES-MCP-MVP-设计文档.md。
快速开始
uv venv
uv pip install -e ".[dev]"
cp .env.example .env # 填数据库连接
python scripts/probe_otdmes.py # 先看这个客户库长什么样
mes-mcp # 启动 MCP Server(stdio)
接入 Claude Code:
{ "mcpServers": { "mes": { "command": "mes-mcp", "cwd": "E:/AIDev/MES-MCP" } } }
probe_otdmes.py 会输出「这个客户库长什么样」:模块启用度、枚举值集、缺失表。接新客户第一件事就是跑它。
双库对比(验证表结构基线是否仍成立):
python scripts/probe_otdmes.py \
--other-port 20000 --other-database OTDMES_CUSTOMER_B \
--out reports/compare.md
两条必须先知道的设计约束
这两点是双库实测(OTDMES_CUSTOMER_A + OTDMES_CUSTOMER_B)得出的,直接决定了代码怎么写。
一、表结构是统一产品基线,枚举不是
14 张核心表在两个客户库里逐字段完全一致(PRD_WORK_ORDERS 都是 90 字段,WMS_PICKING_LIST_HEADER 都是 43……),连 CUS_ 前缀的"定制表"都是同样的 18 张——它们是标准产品包的一部分。
所以:
| 层 | 做法 | 位置 |
|---|---|---|
| 表名 / 字段名 | 硬编码 | catalog.py |
| 枚举值 / 字典 | 启动时读库 | probe.py |
| 模块启用度 | 启动时探测 | probe.py |
枚举确实按客户扩展:领料单类型客户A有 3 种、客户B有 7 种(多出的全是委外相关);仓库编码客户A是 RM01/FG01/SP1、客户B是 CK002/CK003/SFG。别把任何枚举写死在业务代码里。
二、MODULE_NOT_DEPLOYED 不是错误,是事实
同样的表,两家客户使用情况几乎相反:
| 模块 | 客户A | 客户B |
|---|---|---|
| 生产报工 | 19,750 | 0 |
| 设备与工装 | 675 | 0 |
| 成品/过程检验 | 291 | 0 |
| 领料发料 | 931 | 15,335 |
| 工艺与BOM | 42 | 1,698 |
若工具对无数据模块返回空列表,Agent 会把"没上报工模块"读成"今天没产出",结论错得离谱。因此:
deployment.require("reporting") # 无数据则抛 ModuleNotDeployedError
错误信息里带明确的替代路径("请改用工单产出数 ORDER_OUT_QTY 估算"),而不是让 Agent 自己猜。
措辞注意:探测只能看出当前库里没有数据,看不出原因——可能是没买、没上线,也可能只是测试库没铺数据。所有提示一律写「当前库无 X 数据」,不写「该客户未启用」。
目录结构
mes_mcp/
├── server.py ★ 43 个 MCP 工具注册 + main()
├── runtime.py 配置/数据库/探测结果的运行时单例(惰性探测)
├── rendering.py QueryResult → Markdown;模块门禁与错误兜底装饰器
├── query.py ★ 查询构造器(过滤/绑参/分页/别名),所有工具共用
├── charts.py chart-data 嵌入(ERP 的信封格式 + MES 自己的 type)
├── skills/ ★ 复合 Skill:给结论,不只给数据
│ ├── base.py SkillResult(含 degraded / caveats)+ SkillContext
│ ├── genealogy.py SN 族谱(正反向)
│ ├── diagnosis.py 工单延期归因、良率归因
│ └── alerts.py 缺料预警、每日简报
├── domains/ 按业务域的查询构造
│ ├── aggregates.py ★ 聚合与老板视角(逐块降级)
│ ├── integration.py ERP 回写台账与下发失败(金蝶云星空)
│ ├── production.py 工单、工序任务、报工、不良
│ ├── warehouse.py 库存、库位、领料、到货、入库、发货、盘点、流水、追溯
│ ├── quality.py 检验批、IQC
│ ├── equipment.py 设备台账、报警
│ └── engineering.py 工艺路线、BOM、物料主数据
├── config.py pydantic-settings。一份 .env 绑一个客户库,运行时不切库
├── errors.py MESError + ModuleNotDeployedError + 错误码字典
├── db.py SQL Server 只读访问层(pymssql)
├── dates.py ★ 日期风格归一化 —— 见下
├── catalog.py 表白名单、模块定义、枚举源、固定枚举、日期字段声明
├── probe.py 部署探测:模块启用度 + 枚举发现 + Markdown 渲染
├── ontology/ MES/WMS 语义模型 + 跨系统关联键
├── formatting.py dict → Markdown 表(移植自 erp_mcp)
└── timezone.py +8 时区助手(移植自 erp_mcp)
scripts/probe_otdmes.py 客户库探测 / 双库对比
tests/unit 纯逻辑,不连库
tests/integration 连真库冒烟,未配 .env 自动 skip
工具清单(43)
先调 mes_describe_deployment,它告诉你这个客户启用了哪些模块、有哪些仓库、
单据类型有哪几种。不看就查,很容易把"模块没数据"误读成"业务量为零"。
| 域 | 工具 |
|---|---|
| 生产 | mes_query_production_orders mes_query_process_tasks mes_query_work_reports mes_query_defects |
| 质量 | mes_query_quality_inspections mes_query_iqc_orders |
| 设备 | mes_query_equipment mes_query_equipment_alarms |
| 工程 | mes_query_routes mes_query_bom mes_query_materials |
| 仓储 | wms_query_stock wms_query_bins wms_query_picking_lists wms_query_picking_list_lines wms_query_arrivals wms_query_input_orders wms_query_shipping_orders wms_query_inventory_orders wms_query_stock_movements wms_query_traceability |
| 聚合 | mes_production_dashboard mes_order_progress mes_yield_ranking mes_top_defects mes_wip_distribution mes_equipment_alarm_ranking wms_stock_summary wms_material_readiness |
| ERP 集成 | mes_erp_sync_health mes_query_erp_sync_records mes_query_erp_inbound_errors |
| Skill | mes_sn_genealogy mes_order_delay_diagnosis mes_yield_attribution mes_shortage_alert mes_daily_brief |
| 元数据 | mes_describe_deployment mes_list_tables mes_describe_table mes_get_ontology |
| 版本 | mes_check_update mes_self_update |
几个值得单说的:
wms_query_traceability—— SN 正反向族谱。给lot_no反查这批料流向了哪些成品, 是质量召回圈定范围的基础。ERP 答不了这个问题。必须给定位条件,否则拒绝执行 (无条件全表扫追溯表会拖垮生产库)。mes_query_work_reports——group_by支持process/line/product/day, 汇总模式直接给yield_pct良率。wms_query_picking_list_lines—— 返回需求量、已扫量、缺口, 以及委外占用/损耗三列。委外客户做齐套分析必须计入这几列。
聚合工具的两个特点:
- 逐块降级:
mes_production_dashboard跨 6 个模块取数,某模块无数据时该板块 被列进「未取到的板块」并说明原因,其余照常输出。所以在只上了 WMS 的客户那里 它也能用。mes_order_progress同理。 - 嵌 chart-data:返回 Markdown 末尾带
<!-- chart-data {json} -->, 沿用 ERP-MCP 的信封格式(erp-charts-hook能直接提取),但 type 是 MES 自己的 (mes_ranking/mes_pareto/mes_wip/mes_dashboard)——ERP 那几个渲染器 是金额导向的,拿来画良率百分比会得出荒谬的图。纯查询工具不嵌。
ERP 集成:两个方向别只看一半
某客户与金蝶云星空的接口很重,而且失败率不低——实测 870 条回写里 108 条失败(12.4%),
WMS_FG_INPUT 更是 75%。回写失败直接等于 ERP 账实不符。
MES → ERP(回写/过账) WMS_ERP_REPORT_MANAGE_HEADER 标准产品表
ERP → MES(下发) ErpErrorLog ⚠️ 客户定制表,仅部分客户有
mes_erp_sync_health 会把金蝶的报错归一化后聚类(抹掉单号、物料号、行号再分组),
否则同一类错误会散成一条一组——实测「反写采购订单超出可退数量」被行号拆成 4 个桶,
归一后才看出它是最大的一类(45 次)。
三个坑:
- 失败原因在
REPORTED_ERP_RESULT_DETAIL,_MESSAGE与_CODE实测全是空串 - 失败 / 待过账 / 结果未知是三种状态,别混。失败率的分母只取已判定的(OK+NG), 把「结果未知」算进分母会把失败率稀释掉
- 区间内 0 条回写不等于一切正常。工具会区分「这段时间没过账」(附最近一条的时间, 是告警)和「从未有过回写」(未上集成)
Skill 层的一条硬规矩
Skill 给的是结论,不只是数据。所以每个 Skill 的返回都强制带两段:
- 本次分析跳过的环节 —— 哪些模块无数据、由此带来的结论缺口。 在残缺数据上给一个笃定的归因,比不给归因危险得多。
- 口径与限制 —— 这个结论算的到底是什么。
最要紧的是 mes_sn_genealogy 的精度声明。实测 OTDMES 的发料记录挂在领料单上
(OPERATION_ORDER_NO 是领料单号,客户A 167/167 命中领料单表),链路是:
成品 SN → PRD_UNIT_MASTER.ORDER_NO → WMS_PICKING_LIST_WORK_ORDER
→ 领料单 → 发料流水 → UNIT_MSN → 批次/供应商/到货单 → 采购订单
给出的是工单级(批次级)关联,不是 SN 级。只能确定"这批料发给了这张工单", 不能确定"这颗料装在这台机器上"。用于召回时它给的是外延(可能受影响的最大范围)。 有一条测试专门断言这句话在每次返回里都出现——说小了会漏召回,是要出事的。
(WMS_UNIT_TRACEABILITY_INFO 表名字像用料追溯,实测存的是 SN↔父SN 的包装层级,
MPN/MANUFACTURER_NAME 全空,别拿它当用料族谱用。)
新增工具时记得同步改 tests/integration/test_tools.py 的 CALLS 与工具数断言——
有一条测试专门检查"每个注册的工具都有冒烟调用覆盖"。
开发时最容易踩的坑
按被坑概率排序。完整清单见设计文档 §7。
1. 日期有三种风格混用,拼错不报错只出错数
SPLIT_VARCHAR PRD_WORK_ORDERS.CREATION_DATE + .CREATION_TIME 两个 varchar
DATE_VARCHAR PRD_WORK_ORDERS.PLAN_BEGIN_DATE 单个 varchar
DATETIME PRD_LOT_REPORT_MANAGEMENT.CREATE_TIME 标准 datetime
同一张表里都可能并存。永远用 dates.build_date_filter(),不要手拼 WHERE:
from mes_mcp.catalog import date_field
from mes_mcp.dates import build_date_filter
sql_frag, params = build_date_filter(
date_field("PRD_WORK_ORDERS", "created"), "2026-08-01", "2026-08-31"
)
varchar 日期实测是 YYYY-MM-DD(带横杠),字典序比较即正确;datetime 的 end 用 < 次日零点,否则漏数据。这些差异由该函数吸收。
2. charset 必须是 CP936
各库排序规则是 Chinese_PRC_*,varchar 列存 GBK 字节。用 UTF-8 连接不报错,只是把中文静默读成 ²ð°ü×÷Òµ。集成测试里有专门的回归断言。
3. as_dict=True 下所有 SELECT 表达式必须起别名
SELECT COUNT(*) 会抛 ColumnsWithoutNamesError。写成 SELECT COUNT(*) AS n。
4. SQL Server 18456 同时代表"口令错"和"无权访问该库"
服务器故意不区分(防止探测库名)。db.py 的错误翻译把三种可能都列进建议里——本项目开发期就因为这个把权限问题误判成了口令问题。
5. 报工时间用 CREATE_TIME,不是 OPREAT_DATE
后者有 1900-01-01 哨兵值(实测 76/19750)。DateField(sentinel=True) 会自动排除。
6. WMS_OPERATION_TYPE 字典表不完备
客户B数据里有 33 条 OP_TYPE='Delete',字典表里没这一项。翻译层 fallback 返回原 code,绝不丢流水。
7. 备份表必须排除在白名单外
WMS_MATERIAL_ARRIVAL_BODY20250828、WMS_FG_STOCK_MASTER_DELETE 这类混进来会让统计翻倍。catalog.is_backup_table() 负责识别,单元测试里有守卫。
8. PRD_UNIT_HISTORY 这张表不存在
真名是 PRD_UNIT_PRD_HISTORY。现有 C# 版 Otd.MCP.Server 写错了,集成测试里有回归断言。
9. 只读守卫不能裸匹配关键词
工单状态实测有 'CREATE' 这个值,[ORDER_STATUS] = 'CREATE' 曾被守卫当成建表语句
整块拦掉(日报因此少一个板块)。_assert_readonly 现在会先抹掉字符串字面量与
[方括号标识符] 再查关键词。加新守卫规则时注意同样的陷阱。
测试
pytest tests/unit # 纯逻辑,随时可跑
pytest tests # 含连库冒烟,需 .env
验收硬要求:每个工具都必须在客户A与客户B两个库上分别跑通。 这两家模块启用度近乎互补(一个 MES 重、一个 WMS 重),是天然的双向回归测试集。只在单库测的工具,换客户必翻车。
安全边界
MES 是生产库,边界比 ERP 更严:
- 数据库账号只授
db_datareader(集成测试里有断言验证) db._assert_readonly静态拒绝一切非 SELECT 语句与拼接语句- 表名只能来自
catalog白名单,且经assert_identifier校验 - 业务值一律参数绑定,不做字符串拼接
- 不提供裸 SQL 工具——现有 C# 版的
query_tables(where_clause=...)让 LLM 自由写 WHERE,本项目不继承这个设计 - 强制 TOP 上限与查询超时
- 零业务写操作。唯一会改环境的是
mes_self_update——它只动本机的 Python 包, 不碰任何 MES 数据,且强制两步确认(首次调用只预览命令)。MES_VERSION_PIN可在客户现场冻结版本,设了之后一律拒绝升级。
CI 与发布
.github/workflows/ci.yml 每次 push / PR 跑:
- 单元测试(130 条,Python 3.10 / 3.11 / 3.12)
- 集成测试在无
.env时必须全部 skip —— 否则谁在没配库的机器上跑一次pytest就会看到一片红 - 43 个工具能全部注册且都有 docstring
- ontology 里点名的工具必须真的存在 —— 否则 Agent 照着它调会扑空
- 独立分发的图谱页面自带
charset(在前 1024 字节内)
集成测试连的是客户内网 SQL Server,CI 上跑不了,也不该把生产库凭据放进 GitHub Secrets —— 那些在本地对着真库跑,且必须三个库都跑。
.github/workflows/release.yml 在打 v* tag 时发布到 PyPI(Trusted Publishing,
仓库里不放 token),发布前校验 tag 与 __version__ 一致 —— 打错 tag 会把
1.2.0 的代码发成 1.3.0,事后很难查。
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file mes_mcp-0.1.0.tar.gz.
File metadata
- Download URL: mes_mcp-0.1.0.tar.gz
- Upload date:
- Size: 110.4 kB
- Tags: Source
- Uploaded using Trusted Publishing? Yes
- Uploaded via:
twine/7.0.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
8c54d1247f7a36571175c5b943d6983b1932962ce4f3169b5d405671cc83f532
|
|
| MD5 |
5f1ccd68d9abe29b79a58d6ec29e6e40
|
|
| BLAKE2b-256 |
d5ce3318fcf381d4affa70fc3f6be9cd3fb58c2df487e91b8adfc5de44960836
|
Provenance
The following attestation bundles were made for mes_mcp-0.1.0.tar.gz:
Publisher:
release.yml on stevendingliujian-collab/MES-MCP
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
mes_mcp-0.1.0.tar.gz -
Subject digest:
8c54d1247f7a36571175c5b943d6983b1932962ce4f3169b5d405671cc83f532 - Sigstore transparency entry: 2608792742
- Sigstore integration time:
-
Permalink:
stevendingliujian-collab/MES-MCP@1ca5275b14cc86d5491fe83fb2fd49a74f8fc0f7 -
Branch / Tag:
refs/tags/v0.1.0 - Owner: https://github.com/stevendingliujian-collab
-
Access:
private
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
release.yml@1ca5275b14cc86d5491fe83fb2fd49a74f8fc0f7 -
Trigger Event:
push
-
Statement type:
File details
Details for the file mes_mcp-0.1.0-py3-none-any.whl.
File metadata
- Download URL: mes_mcp-0.1.0-py3-none-any.whl
- Upload date:
- Size: 122.9 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? Yes
- Uploaded via:
twine/7.0.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
7b079952504e82de349f697d32eda8b137aef2b361efe52c66bc1e32614ac5dd
|
|
| MD5 |
80f01e4880bca250f46985638eb8ec56
|
|
| BLAKE2b-256 |
455b8259ea81bd122ccda988e31056315a0a345fa3629355898a8d8c41678543
|
Provenance
The following attestation bundles were made for mes_mcp-0.1.0-py3-none-any.whl:
Publisher:
release.yml on stevendingliujian-collab/MES-MCP
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
mes_mcp-0.1.0-py3-none-any.whl -
Subject digest:
7b079952504e82de349f697d32eda8b137aef2b361efe52c66bc1e32614ac5dd - Sigstore transparency entry: 2608793392
- Sigstore integration time:
-
Permalink:
stevendingliujian-collab/MES-MCP@1ca5275b14cc86d5491fe83fb2fd49a74f8fc0f7 -
Branch / Tag:
refs/tags/v0.1.0 - Owner: https://github.com/stevendingliujian-collab
-
Access:
private
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
release.yml@1ca5275b14cc86d5491fe83fb2fd49a74f8fc0f7 -
Trigger Event:
push
-
Statement type: