AI Agent接入物理设备：Anthropic plumbing spec解读与最小工程实践

Anthropic 最近因为一个看似不起眼的动作，把 AI Agent 社区的目光从"模型参数"拉回到了"连接方式"上——它提出了一份用来连接 AI Agent 与实验室设备、机器人的 plumbing spec。

如果你不了解前因后果，可能会以为这只是又一份 API 文档。但如果把这件事放到 AI Agent 的演化路径里看，它其实指向了一个很多人早就该面对、却一直没人替行业回答的问题：当 AI Agent 想操作一台真实的实验设备或机器人时，代码和物理世界之间的"管道"由谁来定义？

这篇文章会从三个层面展开。先把 plumbing spec 到底是什么讲清楚，再说明它为什么比模型能力本身更值得关注，最后给出一个能在你自己的设备接入场景里直接复用的最小工程模式——哪怕你暂时用不上 Anthropic 的方案，这套"设备代理层 + 工具能力描述 + Agent 调用"的思路也能帮你少踩很多坑。

1. 为什么 AI Agent 需要一套"管道规范"

现在的 AI Agent 能写代码、能查数据库、能操作浏览器、能在企业内部系统里帮你发起审批流程，本质都是在"信息世界"里工作。信息世界的接口相对抽象，API、函数调用、数据库查询语言都已经高度标准化，Agent 借助工具调用能力可以快速接入。

但一旦把目标从"操作软件"变成"操作物理设备"，事情就完全变了。

实验室里常见的高通量移液工作站、温控仪、离心机、光谱仪，工厂里的机械臂、AGV 小车、PLC 控制器，虽然都叫"设备"，但彼此之间的通信协议、指令格式、状态表达方式千差万别。有的走串口，有的走 Modbus，有的走 OPC UA，有的干脆只能通过厂商私有的上位机软件来控制。

没有规范的时候，想让 Agent 控制一台设备，通常只有三条路：

第一，针对每台设备写一段专用的控制脚本。这只能做到"完成任务"，完全谈不上"能力复用"。换一台设备，代码全部重来。

第二，让 Agent 通过人类操作界面去点按钮。这本质上是在模拟人工操作，既不稳定，也不适合对安全要求高的场景。

第三，由平台方绑定一套封闭的设备协议。这能解决小范围问题，但会形成新的烟囱，设备厂商、软件开发者、集成商之间无法互通。

所以真正的问题不是"模型不够聪明，看不懂设备说明书"，而是"设备的能力根本没有一种通用的、可被程序化理解的语言"。

Anthropic 提出的 plumbing spec，正是在这个缺口上给出一个方向：在 AI Agent 和物理设备之间定义一套标准化的连接方式，让设备如何描述自己、Agent 如何发起操作、操作结果如何回传，都有一套双方都能理解的"通用语言"。

"plumbing"（管道/管路）这个词用得相当精准。水管系统之所以能覆盖一座城市，不是因为所有家庭的水龙头长得一模一样，而是因为接口尺寸和管路连接方式有统一标准。任何一台符合标准的设备，只要接上管道，就能进入整个系统。

plumbing spec 想做的是同一件事：让任意一台支持该规范的实验室设备或机器人，通过标准接口接入 Agent 系统，被模型调用。这比让模型更聪明更能触及当下的瓶颈。

2. 什么是 plumbing spec：水管与接口的隐喻

要理解 plumbing spec，不能把它当成又一个传输协议。

在软件工程里，我们已经有 HTTP、gRPC、MQTT 这些成熟的通信协议，它们解决的是"数据包怎么从一个节点到另一个节点"的问题。plumbing spec 站在更高的层次，它关心的不是数据怎么传，而是物理设备的能力如何被一个智能体理解、规划、调用和验证。

可以把它拆成三层：

第一层是"能力描述"。设备需要告诉 Agent：我是谁，我支持哪些操作，每个操作需要什么参数，执行后会返回什么状态。这类似于工具调用里的 function schema，只不过这里的工具是真实世界里的一台机器。

第二层是"执行协议"。Agent 发起操作后，设备如何接收指令、如何确认执行、如何返回进度和结果。这里尤其要处理"异步""耗时操作""部分成功"这些真实世界里绕不开的问题。

第三层是"安全与约束"。物理设备不是数据库接口，执行错误会造成真实代价。规范需要定义权限边界、操作确认机制、紧急停止路径、审计日志等要求。

用一个更贴近开发的比喻：plumbing spec 不要求所有设备说同一种方言，而是给设备定义了一本"翻译手册"。每个设备的厂商可以把自家协议继续保留，但在接入层做一次适配，把自己的能力翻译成规范描述的标准结构。Agent 不需要理解每一台设备的私有协议，它只需要学会读这本"翻译手册"。

所以更准确地说，plumbing spec 不是"接口的协议"，而是"接口的接口"。它解决的是生态碎片化的问题。

很多人容易有一个误解：以为这是要让所有设备厂商放弃自己的协议，统一用一种新协议。这种理解是错误的，也没有必要。更好的方式恰恰相反——底层协议百花齐放，但在 Agent 接入层形成标准，让模型侧只需要面对一套稳定的接口描述。

如果把这个思路落地到工程上，其实就是我们后面要讲的：在 Agent 与设备之间插入一个"设备代理层"，把异构设备包装成统一能力。

3. 谁最需要这个规范：场景拆解

理解了概念之后，很重要的问题是：这套东西到底能用在哪些地方，解决了什么真实痛点。

第一个场景是实验室自动化。很多生物实验室、化学实验室里已经有自动化设备，比如自动移液工作站、酶标仪、PCR 仪。但问题是，这些设备往往各自为战，每台设备都有自己的软件和脚本语言。实验人员要做的是把"设计实验 → 写方法 → 配置设备 → 运行 → 收集数据 → 分析"串成自动化流程。plumbing spec 如果能统一设备接入方式，Agent 就可以像编排 API 调用一样编排整个实验流程，而不需要在每台设备的脚本语言之间手动切换。

第二个场景是机器人任务编排。机器人领域的异构性比实验室更明显。一台机械臂的控制器协议、一台移动底盘的运动控制接口、一台视觉系统的触发机制，可能来自三个不同厂商。想让 Agent 完成"把 A 位置的物料移动到 B 位置并检测缺陷"这类跨设备任务，首先得解决编排问题。plumbing spec 的价值在于把每台机器人的能力抽象成标准操作原语，让 Agent 在规划时就拥有统一的行动空间。

第三个场景是无人值守数据采集。气象站、水质监测站、环境传感器阵列等设备，通常部署在偏远位置。过去每次改变采集逻辑，都需要人工到现场修改设备配置。如果设备接入统一的 Agent 控制管道，就可以通过自然语言或任务描述远程调整采集策略、验证数据有效性、处理异常事件。

第四个场景是科研流水线的复现与共享。学术研究里有一个普遍痛点：一篇论文的实验配置往往无法被其他团队复现，因为设备型号不同、脚本不同、参数格式不同。如果设备能力描述标准化，论文的实验流程就可以从"人话描述"变成"机器可执行的任务描述"，其他团队只要把设备接入同一个规范，就能直接运行。

四个场景放在一起，能看出一个共同点：它们都需要把"设备能力"从一个需要人工理解的东西，变成可以被程序化调用的东西。这正是 plumbing spec 的核心价值。

下表可以更直观地对比：

这里有一个很关键的工程判断：plumbing spec 本身不会直接让设备"变聪明"，但它能让 Agent 的聪明真正落得到设备上。没有这层管道，再强的模型也只能隔空对话。

4. 技术架构推演：从 Agent 到物理设备之间的分层

如果你想在自己的环境里提前按这个思路落地，可以参考下面的分层架构。这套分层不绑定 Anthropic 的具体实现，而是通用的工程模式，在任何 Agent + 物理设备项目中都适用。

整个链路可以拆成四层：

Agent（模型 + 决策） ↕ 工具调用协议（JSON 能力描述 + 执行请求） 设备代理层（Device Proxy） ↕ 厂商自定义协议（Modbus / 串口 / HTTP / OPC UA...） 物理设备（仪器 / 机器人）

第一层是 Agent 本体，负责理解任务、拆解步骤、决定调用哪个工具。它只认识标准化的工具描述，不关心设备底层协议。

第二层是工具调用协议，也就是"Agent 与设备代理层"之间的接口。这一层通常使用 JSON 格式描述工具：工具名称、参数列表、返回值、错误码。Agent 通过模型的原生工具调用能力发起请求。

第三层是设备代理层，也是整个架构里最关键的一层。它的职责可以概括为八个字：统一描述，向下适配。每个设备对应一个代理服务，代理内部翻译设备私有协议，对外暴露标准接口。在这一层，你要处理设备状态管理、指令排队、超时重试、错误归一化。

第四层是物理设备本身，继续使用它自己的通信协议。对于已有的设备，你不需要改造它，只需要为它写一个代理。

如果对 Anthropic 生态比较熟悉，会知道 MCP（Model Context Protocol）这类思路也被广泛讨论。这里可以做个保守判断：无论未来 plumbing spec 的具体格式如何演进，"Agent 只面向标准能力描述、设备通过代理适配接入"这个分层的核心思想，都会延续下去。

为什么设备代理层如此重要？

因为设备状态和 API 调用有着本质区别。API 调用默认是幂等的、可重试的，你调两次同一个查询接口，结果应该一致。但物理设备不是这样：你让机械臂执行一次搬运动作，执行成功后再次下指令，设备可能已经处于新的位置，盲目重试会造成碰撞或损坏。

所以在设备代理层，必须做到三件事：

第一，状态跟踪。代理要记住设备当前处于什么状态，是空闲、执行中、暂停还是故障。Agent 发来的指令只有在空闲状态才允许进入执行队列。

第二，校验。在把指令翻译成设备指令之前，代理要先做参数合法性校验、位置边界校验、工况安全校验。

第三，结果归一化。设备返回的原始报文可能是各种稀奇古怪的格式，代理要把它们转换成统一的结构化结果，比如：执行成功、执行失败、超时、执行中、已被取消。

这样的分层设计，也天然形成了安全边界：Agent 永远不能直接触达设备，所有操作都经过代理层。一旦发现异常，你可以在代理层切断链路、暂停任务，而不是去追着模型请求做限制。

5. 最小接入示例：把一台实验室设备变成 Agent 可调用的工具

下面我们用一个最小示例，跑通"Agent → 设备代理层 → 物理设备"的完整链路。

为了演示方便，我用 FastAPI 模拟一台温控设备。这台设备有一个"设定目标温度"的操作，和一个"读取当前温度"的查询操作。真实项目里，这些操作会通过串口或厂商 SDK 访问真实设备，示例中直接用内存变量模拟。

5.1 定义能力描述

首先，用一个 JSON 文件描述这台设备暴露给 Agent 的能力。这是设备代理层对外输出的"翻译手册"。

{ "device": "thermal_controller_01", "description": "实验室温控设备，支持温度设定与温度读取", "tools": [ { "name": "set_temperature", "description": "设置目标温度。温度范围 10 到 200 摄氏度。设定成功后设备开始升温或降温。", "parameters": { "type": "object", "properties": { "target_temp": { "type": "number", "minimum": 10, "maximum": 200, "description": "目标温度，单位摄氏度" }, "hold_time": { "type": "number", "minimum": 0, "description": "达到目标温度后的保持时间，单位分钟" } }, "required": ["target_temp"] } }, { "name": "read_temperature", "description": "读取设备当前温度", "parameters": { "type": "object", "properties": {} } } ] }

这个文件里的结构，基本就是 AGent 侧能理解的"工具定义"。模型会依据 description 和 parameters 来决定何时调用、传什么参数。

5.2 设备代理服务

接下来写设备代理服务。这里的关键点是：对外暴露标准工具调用接口，对内维护设备状态。

# 文件路径：device_proxy/app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field app = FastAPI(title="Thermal Controller Proxy") # 模拟设备状态 device_state = { "current_temp": 25.0, "target_temp": None, "status": "idle", # idle / heating / cooling / holding / error } class SetTemperatureRequest(BaseModel): target_temp: float = Field(ge=10, le=200, description="目标温度，单位摄氏度") hold_time: float = Field(default=0, ge=0, description="保持时间，单位分钟") @app.get("/health") def health(): """供 Agent 或运维系统探测设备代理服务状态""" return {"status": "ok", "device": "thermal_controller_01"} @app.get("/capabilities") def capabilities(): """返回设备能力描述，Agent 启动时拉取该列表""" return { "device": "thermal_controller_01", "description": "实验室温控设备，支持温度设定与温度读取", "tools": [...] } @app.post("/tools/set_temperature") def set_temperature(req: SetTemperatureRequest): if device_state["status"] not in ("idle", "holding"): raise HTTPException(status_code=409, detail="设备忙，当前无法执行新的温度设定") if req.target_temp > device_state["current_temp"]: device_state["status"] = "heating" elif req.target_temp < device_state["current_temp"]: device_state["status"] = "cooling" else: device_state["status"] = "holding" device_state["target_temp"] = req.target_temp return { "status": "accepted", "device_status": device_state["status"], "current_temp": device_state["current_temp"], "target_temp": req.target_temp, "message": f"正在向目标温度 {req.target_temp} 摄氏度调节" } @app.get("/tools/read_temperature") def read_temperature(): return { "status": "ok", "current_temp": device_state["current_temp"], "device_status": device_state["status"] }

代码里最关键的是set_temperature接口中的状态判断：只有当设备处于空闲或保持状态时，才接受新的温度设定。这个判断对应真实场景里的"防止设备冲突"。

5.3 Agent 侧调用

Agent 侧不直接请求device_proxy，而是通过大模型的工具调用（function calling）能力发起。为了方便演示，下面用 Python 脚本模拟这个调用过程：

# 文件路径：agent_demo.py import json import urllib.request PROXY_BASE = "http://127.0.0.1:8000" def call_tool(tool_name: str, parameters: dict): """模拟 Agent 收到模型工具调用结果后，向设备代理发起请求""" if tool_name == "set_temperature": url = f"{PROXY_BASE}/tools/set_temperature" data = json.dumps(parameters).encode("utf-8") req = urllib.request.Request( url, data=data, headers={"Content-Type": "application/json"}, method="POST" ) elif tool_name == "read_temperature": url = f"{PROXY_BASE}/tools/read_temperature" req = urllib.request.Request(url, method="GET") else: raise ValueError(f"未知工具: {tool_name}") with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode("utf-8")) if __name__ == "__main__": # 模拟模型决策：先读取当前温度，再设定目标温度 current = call_tool("read_temperature", {}) print("当前温度:", current["current_temp"]) result = call_tool("set_temperature", {"target_temp": 120, "hold_time": 30}) print("执行结果:", result)

这里的call_tool就是 Agent 工具调用的最小模型。真实项目中，这一层会由 LangChain、LlamaIndex 或 Anthropic 等平台提供的工具调用框架接管，但核心逻辑不变：模型输出工具名和参数，你的代码负责转发给设备代理。

5.4 启动与验证

启动设备代理服务：

cd device_proxy pip install fastapi uvicorn pydantic uvicorn app:app --host 0.0.0.0 --port 8000

再执行 Agent 调用脚本：

python agent_demo.py

正常情况下，你会看到类似输出：

当前温度: 25.0 执行结果: {'status': 'accepted', 'device_status': 'heating', 'current_temp': 25.0, 'target_temp': 120.0, 'message': '正在向目标温度 120 摄氏度调节'}

这个最小的链路已经跑通了 Agent 读取设备状态、向设备下发控制指令的完整闭环。把它换成真实设备，你只需要在设备代理层替换掉模拟状态读写逻辑，改成通过串口、Modbus 或厂商 SDK 操作真实硬件。

6. 运行结果与效果验证

上一步只是程序跑通，还不能算"接入成功"。要判断 Agent 是否真的能够可靠地控制设备，需要从三个层面验证。

第一层，能力发现是否正常。Agent 启动时能不能通过/capabilities接口拿到完整的工具定义，并且工具的参数约束（temperature 范围、必填项）能被模型正确理解。实际测试方法是让 Agent 直接输出下一次调用要使用的工具名和参数，检查它是否生成了合法的 JSON。

第二层，执行链路是否正确。调用set_temperature后，检查返回里的device_status是否按预期切换为heating或cooling，目标温度是否设置正确。如果设备代理返回409，要确认是因为设备忙触发了状态保护，还是因为参数越界被 pydantic 拦截。

第三层，状态一致性是否保持。连续调用两次read_temperature，观察当前温度是否按向目标温度靠近的逻辑变化。如果真实设备带有传感器，还应该核对代理返回的数据与设备本地显示是否一致。

如果运行时出现问题，建议按下面的顺序排查：

第一步，查看启动日志。uvicorn 启动的终端会打印每次请求的路由、状态码和异常堆栈，大多数问题在这一步就能定位。

第二步，确认设备代理服务健康。访问http://127.0.0.1:8000/health，检查是否返回{"status": "ok"}。

第三步，用命令行直接测试工具接口，绕开 Agent，确认为什么设备代理本身是否正常。比如：

curl -X POST http://127.0.0.1:8000/tools/set_temperature \ -H "Content-Type: application/json" \ -d '{"target_temp": 150}'

如果 curl 直接请求成功，而 Agent 脚本失败，问题通常出在参数格式或调用方式上，而不是设备代理本身。

7. 常见问题与排查思路

把 Agent 接入物理设备，踩坑点比普通接口开发多得多。下面列几个典型问题。

8. 最佳实践与工程建议

如果在真实项目里按 plumbing spec 的思路接入设备，有几个工程建议值得提前考虑。

第一，主动设计状态机。物理设备的操作不能像普通 API 那样随意重试。建议在设备代理层维护一个明确的状态机，至少包含空闲、执行中、暂停、故障、已完成几种状态。每次收到工具调用，先检查状态机是否允许当前操作，再往下执行。

第二，为每个物理操作引入唯一 ID。Agent 调用设备时，建议通过request_id标识一次物理操作。如果请求超时，重试时带上同一个request_id，代理层可以做去重，避免机械臂执行两次同类动作。

第三，建立"人工确认"的安全闸门。对运动控制、高温调节、大功率操作等高风险动作，代理层可以增加require_confirmation: true的配置。Agent 发起这类操作时，代理先进入"待确认"状态，由现场人员按下工控屏上的确认按钮，或者调用一次专门的确认 API 后再执行。这在实验室和工厂场景里非常重要。

第四，日志与审计不能省略。每一笔设备操作都要记录：谁（哪个 Agent、哪个用户）发起的，操作了哪台设备，参数是什么，设备返回什么结果，是否成功。事故追溯的时候，这些日志是唯一的证据链。

第五，紧急停止要独立于 Agent 链路。不要在系统里只依赖 Agent 的暂停指令，设备代理层和物理设备侧必须保留独立的急停通道。因为 Agent 可能程序卡死、模型可能产生幻觉、代理层可能崩溃，物理世界的最后一道安全网不能建立在任何一个智能组件之上。

第六，渐进式接入。不要一上来就想把所有设备接入统一管道。建议先选一台控制逻辑简单、风险较低的设备，比如温控器或数据采集器，跑通 Agent → 设备代理 → 真实硬件的完整链路。确认稳定后，再逐步接入运动控制类设备。这样出现问题的时候，排查范围会小很多。

第七，注意工具描述的质量。模型是否能正确调用设备，很大程度上取决于你在能力描述里写得多清楚。参数边界、单位、调用前提、失败返回方式都要写明白，否则模型很容易生成不合法的参数。这相当于你在给模型写"设备说明书"，写得越规范，模型越不容易犯错。

9. 给开发者的下一步建议

Anthropic 提出 plumbing spec，真正的信号不是某个具体协议的诞生，而是 AI Agent 的发展正从"信息空间"走向"物理空间"。模型已经可以理解自然语言里的复杂任务，接下来制约 Agent 应用的，恰恰是它与真实设备之间那套繁琐、异构、脆弱的连接管道。

对开发者来说，现在是一个值得提前布局的窗口期。你不一定需要等待规范完全落地，可以先按照"Agent 面向标准能力描述、设备通过代理适配接入"的思路，在自己的实验或项目里搭一个最小闭环。当你亲手把一台设备变成一个 Agent 可调用的工具之后，就会明白"管道"到底卡在哪些地方。

下一步值得关注的几个方向：一是 plumbing spec 本身的具体细则和生态支持情况，二是主流 Agent 框架是否会内置相关工具协议，三是设备厂商是否会主动提供符合标准的接入适配器。在这三者都还不确定之前，稳健的做法是把设备代理层做成标准化、可替换的模块，这样未来无论规范走向如何，你手里这套分层架构都不会过时。

如果这篇文章对你有用，建议先收藏备用。等你准备把自己的设备接入 Agent 时，再回来按这个最小模型跑通一次，后续的扩展就不会是无从下手了。