Sparkplug B

Sparkplug B MCP 服务

一个用于 Sparkplug B 的 模型上下文协议（MCP）服务。Sparkplug B 是 Eclipse 标准的 MQTT 主题/载荷约定，用于工业/SCADA 数据（出生/死亡证书、带序列号的指标、protobuf 载荷）。可连接承载 Sparkplug B 流量的任意 MQTT 代理；仅使用开放的 Sparkplug B 规范，不使用特定供应商的扩展。

此服务充当 Sparkplug B 主机应用程序：订阅整个 spBv1.0 命名空间，根据 Edge Node 和 Device 的 NBIRTH/DBIRTH 消息发现它们，跟踪其实时指标，并可向其发送命令（NCMD/DCMD）——这与实际部署中 SCADA/历史数据库主机应用程序所扮演的角色相同。

使用 FastMCP 和 pysparkplug 构建（后者基于 paho-mqtt 构建）。

工具

工具	描述
get_connection_status()	报告代理连接状态及此主机应用程序的身份。
set_host_online(online)	发布此连接器自己的保留型 Sparkplug STATE 消息。
list_edge_nodes(group_id?)	列出通过 NBIRTH/NDEATH 发现的 Edge Node。
list_devices(group_id?, edge_node_id?)	列出通过 DBIRTH/DDEATH 发现的 Device。
get_node_metrics(group_id, edge_node_id)	获取 Edge Node 的最新已知指标值。
get_device_metrics(group_id, edge_node_id, device_id)	获取 Device 的最新已知指标值。
send_node_command(group_id, edge_node_id, metrics)	发送 NCMD——写入节点的一个或多个指标。
send_device_command(group_id, edge_node_id, device_id, metrics)	发送 DCMD——写入设备的一个或多个指标。
request_node_rebirth(group_id, edge_node_id)	标准的 Node Control/Rebirth 命令——请求节点重新发布其完整的 NBIRTH。
start_watch(group_id?, edge_node_id?, device_id?, message_types?)	开始缓冲符合可选筛选条件的已解码 Sparkplug 流量。返回一个 session_id。
get_events(session_id, max_events, timeout_seconds)	从监视会话中取出缓冲的事件；可短暂等待事件到达。
stop_watch(session_id)	停止监视会话并丢弃其缓冲的事件。
list_watches()	列出所有活动的监视会话。

为什么监视采用基于会话的方式

与其底层的 MQTT 有线协议一样，Sparkplug B 是一个连续流，而非请求/响应 API——单次 MCP 工具调用无法永远等待下一条出生或数据消息。start_watch 会注册一个可选筛选条件（组/节点/设备/消息类型），并在后台缓冲匹配的已解码消息（此连接器始终订阅整个 spBv1.0 命名空间，因此每个监视会话无需额外订阅代理）；get_events 会取出这些缓冲内容，并可在有界时间内等待新数据。完成后调用 stop_watch。

支持的指标数据类型

INT8/INT16/INT32/INT64、UINT8/UINT16/UINT32/UINT64、FLOAT、DOUBLE、BOOLEAN、STRING、TEXT、UUID、DATETIME、BYTES、FILE 及其 *_ARRAY 变体。DATASET、TEMPLATE、PROPERTYSET 和 PROPERTYSETLIST 属于 Sparkplug B 规范的一部分，但底层的 pysparkplug 编解码器不支持这些类型，因此会明确报错，而不会静默地错误处理。

对于 send_node_command/send_device_command，value 的形式取决于 datatype：标量使用普通的 int/float/bool/str，DATETIME 使用 ISO 8601 字符串，BYTES/FILE 使用 base64 字符串，*_ARRAY 类型使用 JSON 列表。

配置

环境变量	必需	描述
SPARKPLUG_MQTT_HOST	是	Broker 主机名或 IP。
SPARKPLUG_HOST_ID	是	此连接器的 Sparkplug Host Application ID（用于 spBv1.0/STATE/{id} 主题）。
SPARKPLUG_MQTT_PORT	否	默认值为 1883。
SPARKPLUG_MQTT_USERNAME / SPARKPLUG_MQTT_PASSWORD	否	Broker 凭据。
SPARKPLUG_MQTT_CLIENT_ID	否	默认为 Broker 分配的 ID。
SPARKPLUG_MQTT_KEEPALIVE	否	秒数，默认值为 60。
SPARKPLUG_MQTT_CONNECT_TIMEOUT	否	等待初始连接的秒数，默认值为 10。
SPARKPLUG_MQTT_USE_TLS	否	true/false，默认值为 false。
SPARKPLUG_MQTT_TLS_CA_CERTS	否	CA 证书包的路径。省略时使用系统信任库。
SPARKPLUG_MQTT_TLS_CLIENT_CERT / SPARKPLUG_MQTT_TLS_CLIENT_KEY	否	用于双向 TLS。
SPARKPLUG_GROUP_ID	否	将发现/监听范围限制为单个 Sparkplug Group ID。省略时订阅所有组（spBv1.0/#）。

本地运行

uv sync
export SPARKPLUG_MQTT_HOST=localhost
export SPARKPLUG_MQTT_PORT=1883
export SPARKPLUG_HOST_ID=my-mcp-host
uv run sparkplugbmcpserver.py

参阅 mcp.json 获取可直接使用的客户端配置（stdio 和 Docker）。

针对真实 Broker 进行测试

本仓库的设计原则要求针对真实实例进行测试，而非使用模拟对象。docker-compose.yml 会启动Eclipse Mosquitto（2.x），并启用匿名监听器以便进行本地测试。测试固件使用 pysparkplug 自带的 EdgeNode/Device 类，模拟真实的 PLC/设备——发布真实的 NBIRTH/DBIRTH/NDATA/DDATA/NDEATH 消息，并通过网络接收真实的 NCMD/DCMD 命令——因此测试覆盖的是协议层，而非手工编写的模拟对象。由于没有可用于测试的免费真实 PLC/边缘节点硬件，因此仅模拟“物理”节点/设备；双方对 Sparkplug B 协议的处理都是真实的。

docker compose up -d       # start Mosquitto on localhost:1883 (anonymous, local testing only)
uv run pytest -v           # tests/conftest.py also starts/stops the stack automatically
docker compose down        # stop it manually if you started it yourself

许可证说明

docker-compose.yml 和 docker/ 下的文件配置的是开源 Eclipse Mosquitto Broker（EPL-2.0），仅用于本地测试——它们不属于 MCP 服务本身。