Sparkplug Tap

核心功能：通过ot-aiops MCP服务器消费MQTT/Sparkplug B/UNS数据，支持主题读取、Sparkplug B载荷解码、边缘节点/设备发现、命名空间树浏览，默认关闭高风险发布指令。典型场景：对接MQTT broker或Sparkplug B边缘设备，用于数据采样、节点状态排查。输入为配置参数，输出为结构化JSON数据，适配工业设备数据采集，不适用于IT/网络类设备。

zw008

热度 16.10

最近更新 2026-06-28 14:48:46

分类：开发依赖：MCP 服务应用场景：API 集成 / 数据处理

安装技能

复制下面的话术发给你的 Agent，它会通过 secure-skills 检索并安装该技能。

执行命令安装此 skill： npx secure-skills-cli add sparkplug-tap

Metadata

name: sparkplug-tap
description: Consume MQTT / Sparkplug B / Unified Namespace (UNS) data — bounded plain-MQTT reads, Sparkplug B sample (topic parsed + payload decoded), edge-node/device discovery from NBIRTH/DBIRTH, and live topic-tree (UNS) browse. A governed, MOC-gated publish/command exists but is OFF by default. Use when the task names MQTT, Sparkplug, UNS, a broker, or an spBv1.0 topic. Routes to the ot-aiops MCP server. Consume-first. For OPC-UA/Modbus/Siemens/Mitsubishi/MTConnect use the sibling *-tap skills; not for IT/network gear, Kubernetes, hypervisors, or backups.

sparkplug-tap

MQTT / Sparkplug B / UNS consume-first telemetry via the ot-aiops MCP server, using paho-mqtt (pure Python). Preview — validated against synthetic protobuf payloads, NOT a live broker. Sparkplug B is fully decoded (vendored Eclipse Tahu protobuf schema, depends only on protobuf): per metric you get name, alias (resolved via the BIRTH model), datatype (Int/Float/Bool/String/DateTime/ DataSet/Template…), value, timestamp, and is_historical/is_null. A birth/death + seq model tracks node/device online state, applies NDATA/DDATA by alias, and flags seq gaps; STATE topics surface primary-host status.

When to use

An MQTT broker / Sparkplug B edge / UNS (host:1883 or :8883 TLS, topic).
Sample topics, discover edge nodes, browse the namespace tree.

When NOT to use (routing)

OPC-UA → opcua-tap; Modbus → modbus-tap; Siemens → s7-tap; Mitsubishi → mc-tap; CNC → mtconnect-tap; cross-protocol triage → industrial-diagnostics.
IT/network/Kubernetes/hypervisor/backup → not this tool.

Read/consume tools (bounded — never an open loop)

Tool	Params	Returns
`mqtt_read_topic`	`endpoint?, topic="", count=25, timeout_s=10`	`{messages:[{topic,payload:{encoding,json\|text\|sparkplug_b}}]}`
`sparkplug_subscribe_sample`	`endpoint?, topic="", count=25, timeout_s=10`	`{historical_metric_count, seq_gap_count, samples:[{topic, sparkplug:{group_id,message_type,edge_node_id,device_id}, payload:{metrics:[{name,alias,datatype,value,is_historical}]}}]}`
`sparkplug_decode_payload`	`payload, encoding="base64", alias_map?`	`{seq, metric_count, historical_count, metrics:[{name,alias,datatype,value,timestamp,is_historical,is_null}]}`
`sparkplug_node_list`	`endpoint?, timeout_s=10, count=500`	`{nodes:[{group_id,edge_node_id,online,born,devices[...],metric_aliases_known,seq_gap_count}], primary_hosts:[{host_id,state}]}`
`uns_browse`	`endpoint?, topic="#", timeout_s=10, count=500`	`{topic_count, topics[...], tree{...}}`

sparkplug_decode_payload decodes a single raw payload (base64 or hex) — useful to inspect a captured NBIRTH/NDATA without a live broker; pass alias_map to resolve alias-only NDATA/DDATA metrics to names.

Example: sparkplug_node_list(endpoint="uns", timeout_s=15) → {"node_count":2,"nodes":[{"group_id":"Plant1","edge_node_id":"Edge1","online":true,"born":true,"devices":["DevA"],"seq_gap_count":0}],"primary_hosts":[{"host_id":"primaryHost","state":"ONLINE"}]}

Command tool (HIGH risk · MOC · off by default)

mqtt_publish(topic, payload, endpoint?, qos=0, retain=false, dry_run=true)

OT-DANGEROUS. 未经授权勿对生产控制系统下发指令. A command (e.g. Sparkplug NCMD/DCMD) can change a live system and has NO automatic inverse. Defaults to dry_run=true.
Apply: dry_run=false + OPCUA_AUDIT_APPROVED_BY. CLI: ot-aiops mqtt publish factory/cmd '{"sp":50}' --apply (double-confirm prompt).

Setup

ot-aiops init (mqtt: broker/port/topic/TLS/user) · ot-aiops doctor · ot-aiops mcp. Test against a local mosquitto broker (+ a Sparkplug edge). Full Sparkplug B decode is bundled (no extra install). MQTT username/password is stored encrypted; TLS optional. 缺功能提 issue/PR 欢迎留言.