工业 AI

MCP 已经接进 PLC：但写设定值这一步，必须由人按下

agent 能读产线数据这件事已经发生了；能不能写，取决于服务器代码里那几行限额，以及有没有人替它按下按钮。

2026 年，Ignition、CODESYS 和 OPC UA 都有了 MCP server，Rockwell 和西门子则拿出了工程侧的 agent。原文给出一台在产线上试过的 Python MCP server：从 tag 白名单读，设定值只能变成提案，由人批准后才回写 PLC——以及这样一台 server 在厂里该放在什么位置。

10 月 21 日，Rockwell Automation 发布 FactoryTalk Conductor，一个用来在从设计到测试的工程工作中协调 AI agent 的平台。它为这一年收尾：AI agent 从聊天窗口搬进了自动化工具——PLC 开发环境、SCADA 平台、OPC UA server。这些集成大多说同一种协议：Model Context Protocol（MCP）。

所以对一座厂来说，问题不再是"agent 会不会拿到产线数据"，而是按什么条件拿。下面先梳理 2026 年变了什么，再给出一台跑得通的产线 MCP server：agent 读指定的 tag，只能提一条设定值改动建议，改由人做。

一、2026 年带来了什么

自动化工具里的 MCP server。 Inductive Automation 在 2 月 13 日把 Ignition 的 MCP 模块放出来做 early access。在 Ignition 8.3.5 发布之前它需要 nightly build，访问由权限和 API token 管，厂商也直说这个版本不带支持。CODESYS 给自己的开发系统出了 MCP server：4 月 28 日的 1.0，7 月 22 日的 1.1。按它的数据表，工具能浏览工程树、读取并替换 Structured Text、返回设备与 I/O 组态、检查错误。OPC Foundation 在 10 月 7 日把 OPC UA MCP server 的稳定版 2.0.0 发到 NuGet，五天后跟了 2.0.1。它把 OPC UA Part 4 的服务调用暴露成 MCP 工具，工具集可以用 profile（比如 --profile core）收窄。

工程 agent。 西门子一直在扩 Eigen Engineering Agent。6 月在巴黎 VivaTech 上，它加了两个能力：从 ECAD 文件生成 PLC 标签，以及从一句话的机器描述生成符合 Siemens Automation Framework 的项目。西门子报的数字是 19 个国家 100 多家客户、比手工快 2 到 5 倍——这是厂商口径，未核实。Rockwell 计划 2027 年一季度推出 FactoryTalk Conductor，从 Studio 5000 Logix Designer、Logix Echo、Emulate3D 和 FactoryTalk Optix 起步，客户可以接自己的语言模型，包括本地部署的。

标准。 4 月 20 日，OPC Foundation 表示要把 430 多个 OPC UA companion specification 转成适合 RAG 与 MCP 的格式。MCP 自己在 7 月 28 日出了新版规范：不再有协议 session，也不再有 initialize 握手，每个请求自带协议版本与客户端能力。

2026 时间线：Ignition MCP 模块、为 RAG/MCP 优化的 OPC UA 规范、CODESYS MCP server、西门子 Eigen agent、MCP 新规范、OPC Foundation 的 MCP server、Rockwell FactoryTalk Conductor

背景数字。 Rockwell 的 2026 State of Smart Manufacturing 报告覆盖 17 个国家 1,500 多家制造商，它给出的 AI 支撑运营占比是今天 34%，2030 年过半。同一份调查里，46% 的企业至少出过一次网络事件。

二、为什么"把助手接到 PLC"不够

MCP 规范说得很直接：应当始终有人在回路里，并且这个人有能力拒绝工具调用。客户端必须把 readOnlyHint 这类工具注解当作不可信，除非它来自可信服务器。服务器要校验输入、执行访问控制、限流、清洗输出；客户端要记录工具调用以备审计。

网络安全机构的措辞更进一步。CISA、澳大利亚 ACSC 等九个机构 2025 年 12 月 3 日的联合指引要求：限制 AI 在没有人在回路的情况下主动控制 OT 基础设施。它也倾向 push 或 broker 架构——数据离开 OT 时不保留常驻入站通道，跨进业务网时用单向传输和可审计的暂存区。

由此得到三条规则，下面的例子全部遵守：

agent 看到的是 tag 名单，不是控制器。 服务器拒绝任何不在名单上的名字。
一次写就是一条提案。 agent 的工具只把请求排进队列，由操作员运行的另一个程序写入 PLC。
限额由服务器执行，不由模型执行。 设定值范围和权限写在服务器代码里，不写在工具描述里。

三、MCP server 该放在哪

架构：OT 区的 PLC 与网关经 MQTT Sparkplug B 把数据推到 MES 区的 broker 与 TimescaleDB，MCP server 读这份副本；设定值改动经提案队列和操作员回到 PLC

在生产环境里，MCP server 根本不该和控制器说话。 产线数据经网关、MQTT Sparkplug B 到 broker，再落进 OT 网外的时间序列数据库，agent 读的是那份副本。OmniMES 就是这么建的：系统从不连控制器，遥测走 MQTT Sparkplug B 落到 TimescaleDB，MCP server 用的就是这份数据。设定值改动回 OT 只有一条路——经过人。

下面的例子直连 OPC UA，因为那是在试验台上最快能跑起来的方式。规则不变，只是读的来源变了。

四、一步一步做一台产线的 MCP server

我们在 Python 3.11 上、用 mcp 2.3.0 和 asyncua 2.1.0 测过这段代码，对着的是一台带 4 个过程变量和 300 个诊断寄存器的 OPC UA 模拟器。

pip install "mcp==2.3.0" "asyncua==2.1.0"

4.1 服务器：从名单读，以提案写

"""产线 1 的 MCP server：只读白名单内的 tag，写只产出待人批准的提案。"""
import sqlite3
import time
import uuid

from asyncua import Client as OpcClient
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations

ENDPOINT = "opc.tcp://127.0.0.1:4840/line1/"
DB = "proposals.db"

# agent 只看得见这些 tag，控制器地址空间里的其他东西它看不到
READABLE = {
    "press1.state": "ns=2;s=Press1.State",
    "press1.count": "ns=2;s=Press1.Count",
    "press1.oil_temp": "ns=2;s=Press1.OilTemp",
    "press1.pressure_setpoint": "ns=2;s=Press1.PressureSetpoint",
}
# agent 可以提议的设定值，硬限位在服务器侧校验
PROPOSABLE = {"press1.pressure_setpoint": (150.0, 200.0)}

mcp = MCPServer("line1")
_opc = None


async def opc():
    global _opc
    if _opc is None:
        _opc = OpcClient(ENDPOINT)
        await _opc.connect()
    return _opc


def db():
    con = sqlite3.connect(DB)
    con.execute(
        "create table if not exists proposals (id text primary key, tag text, value real, reason text,"
        " created real, status text, approved_by text, executed real, before real, after real)"
    )
    return con


@mcp.tool(annotations=ToolAnnotations(read_only_hint=True, open_world_hint=False))
async def read_tags(tags: list[str]) -> dict:
    """读取产线 1 这些 tag 的当前值。

    允许：press1.state、press1.count、press1.oil_temp、press1.pressure_setpoint。
    """
    unknown = sorted(set(tags) - READABLE.keys())
    if unknown:
        raise ToolError(f"Tags not allowed: {unknown}. Allowed: {sorted(READABLE)}")
    client = await opc()
    result = {}
    for tag in tags:
        dv = await client.get_node(READABLE[tag]).read_data_value()
        stamp = dv.SourceTimestamp or dv.ServerTimestamp
        result[tag] = {
            "value": dv.Value.Value,
            "status": dv.StatusCode.name,
            "time": stamp.isoformat() if stamp else None,
        }
    return result


@mcp.tool(annotations=ToolAnnotations(read_only_hint=False, destructive_hint=False, open_world_hint=False))
async def propose_setpoint(tag: str, value: float, reason: str) -> dict:
    """把一条设定值改动排队，等人批准。在那之前产线上什么都不会变。"""
    if tag not in PROPOSABLE:
        raise ToolError(f"No proposals allowed for {tag}. Allowed: {sorted(PROPOSABLE)}")
    low, high = PROPOSABLE[tag]
    if not low <= value <= high:
        raise ToolError(f"{value} is outside the permitted range {low}-{high}")
    proposal_id = uuid.uuid4().hex[:8]
    with db() as con:
        con.execute(
            "insert into proposals (id, tag, value, reason, created, status) values (?, ?, ?, ?, ?, 'pending')",
            (proposal_id, tag, value, reason, time.time()),
        )
    return {"proposal_id": proposal_id, "status": "pending human approval"}


if __name__ == "__main__":
    mcp.run(transport="stdio")

read_tags 带 read_only_hint=True，propose_setpoint 没有——但客户端没有义务相信注解。真正的保护是 READABLE 和 PROPOSABLE：模型看不见名单外的寄存器，也没法提一个 260 的压力，因为服务器在入队之前就把范围查了。ns=2 这个命名空间索引来自模拟器，硬编码 node id 之前要在真控制器上核对。

4.2 审批：操作员的程序，不是 agent 的

"""操作员侧：批准一条提案并写入 PLC。agent 永远不会运行这段。"""
import asyncio
import sys
import time

from asyncua import Client as OpcClient, ua

from line_mcp import ENDPOINT, PROPOSABLE, db

WRITABLE = {"press1.pressure_setpoint": "ns=2;s=Press1.PressureSetpoint"}


async def approve(proposal_id: str, operator: str):
    with db() as con:
        row = con.execute("select tag, value, status from proposals where id = ?", (proposal_id,)).fetchone()
    if row is None or row[2] != "pending":
        raise SystemExit(f"No pending proposal {proposal_id}")
    tag, value, _ = row
    low, high = PROPOSABLE[tag]
    if not low <= value <= high:  # 再查一次：数据库不是 PLC 的安全层
        raise SystemExit(f"{value} is outside {low}-{high}")
    async with OpcClient(ENDPOINT) as opc:
        node = opc.get_node(WRITABLE[tag])
        before = await node.read_value()
        await node.write_value(value, ua.VariantType.Double)
        after = await node.read_value()
    with db() as con:
        con.execute(
            "update proposals set status = 'executed', approved_by = ?, executed = ?, before = ?, after = ? where id = ?",
            (operator, time.time(), before, after, proposal_id),
        )
    print(f"{tag}: {before} -> {after}, approved by {operator}")


if __name__ == "__main__":
    asyncio.run(approve(sys.argv[1], sys.argv[2]))

程序把范围再查一遍，因为提案数据库不是安全层。它同时记下改动前后的值，以及批准人是谁。在真厂里，这段应该长在带登录的 HMI 面板或应用里，而不是一个控制台脚本。

4.3 不用语言模型也能测

SDK 的客户端可以在同一个进程里连上服务器，所以整条链路不用模型也能跑：

import asyncio

from mcp import Client

import line_mcp


async def main():
    async with Client(line_mcp.mcp) as client:
        result = await client.call_tool("read_tags", {"tags": ["press1.oil_temp"]})
        print(result.is_error, result.content[0].text)


asyncio.run(main())

接真客户端时，用 python line_mcp.py 启动服务器，它走标准输入输出通信。测试模拟器是一台 asyncua 的 OPC UA server：一个 Press1 对象（State、Count、OilTemp，以及可写的 PressureSetpoint），外加 Diagnostics 对象下的 300 个寄存器。

五、SDK 2.x 的两个坑

FastMCP 没了。 2.x 里类叫 MCPServer，从 mcp.server.mcpserver 导入。从 mcp.server.fastmcp 导入会报错，错误信息指向迁移指南——所以网上大量示例需要更新。

普通异常会把错误文本吞掉。 如果工具抛 ValueError，模型只看到 Error executing tool read_tags，永远不知道哪些 tag 是允许的。只有抛 ToolError，这条消息才会传到模型。两种都试过：用 ToolError 时，agent 能拿到允许的名字列表，并改正自己的调用。在 SDK 的结果对象里，这个字段是 is_error。

六、有多少数据会进模型

工具返回体积：控制器整个命名空间 304 个变量 37,397 字节，4 个白名单 tag 465 字节

一个"把所有 tag 都给我"的工具很诱人，因为它一上来就能用。在我们的模拟器上，把控制器整个命名空间（304 个变量）导出来是 37,397 字节；读那 4 个名单内的 tag 是 465 字节，差了约 80 倍。这些字节每一个都会进模型上下文，既花钱又稀释注意力。真控制器的变量数不同，倍数会变，但方向不变。

七、给 agent 多大自由

agent 访问权限的四级：从 tag 名单读、提案由人批准、窄限位内无批准写入、完全自主

图源：OmniMES 原文配图，随文编译。图注由译者译自原文 alt。

原文建议分四级：

级别	做法	适用
1	从 tag 名单读	每条产线都该从这里起步
2	提案由人批准	多数场景的合理目标
3	在窄限位内自动写、不经批准	只在做过风险评估之后，安全功能永远不行
4	完全自主	不建议用在 OT 里

这是原文的建议，不是标准——依据就是前面引的 CISA 指引与 MCP 规范。

八、障碍与限制

工具成熟度。 Ignition 的 MCP 模块是 early access 且不带支持；CODESYS 的 MCP server 不含技术支持，且同一时间只能挂一个开发环境；FactoryTalk Conductor 得等到 2027 年。
厂商口径。 Eigen Engineering Agent 那"2 到 5 倍"是西门子的数字，在自己的项目上量过再决定。
连接安全。 例子里的 OPC UA 没有加密。真厂要 SignAndEncrypt 加证书、给 MCP server 一个专用账号，并在 OPC UA server 侧只给读权限。
数据即模型输入。 tag 值、报警文本、操作员备注，都是以文本形式进模型的。MCP 规范要求：工具结果在交给模型之前先校验。
例子省掉了什么。 它没有限流，也没有读操作日志——而规范要求限流；它也不重连掉线的 OPC UA 会话。用 SQLite 当队列、用脚本当操作面板，做测试可以，轮班干活不行。