1. 从一个"反直觉"的硬件需求说起
功耗计这东西,做硬件和嵌入式的人都不陌生。IoT Power 这类设备本质上是一个带计量能力的 USB 供电监测模块,能实时读出电压、电流、功率、累计电量这些参数。平时我们用它测板子待机功耗、测充电曲线、测某个外设在满载时的电流尖峰,都是常规操作。
但问题来了:每次看数据都得打开上位机软件,或者盯着串口终端刷日志。测一个设备的功耗曲线,我得先手动跑测试脚本,再切到功耗计界面截图,最后把数据抄到表格里做对比。这套流程做一次两次还行,做十次二十次就非常烦。
于是就有了这个想法:能不能让 AI 自己去看功耗计?我把需求说清楚——"帮我测一下这块板子在空闲状态下的平均功耗,跑五分钟,顺便看看有没有异常尖峰"——AI 自己去调功耗计、采数据、算统计、给结论。这中间不需要我手动切窗口、抄数字。
要实现这个,核心不是让 AI 去"看屏幕",而是给它一个能直接调用的接口。这就是MCP(Model Context Protocol)派上用场的地方。MCP 本质上是一套让 AI 模型能够调用外部工具、读取外部资源的协议规范。你写一个 MCP 服务端,把功耗计的能力包装成几个工具函数暴露出去,AI 就能像调函数一样去读功耗数据。
这篇文章就围绕"给 IoT Power 写一个 MCP 服务端"这件事展开。我会讲清楚为什么选 Python + ctypes 这条路、MCP 服务端的核心结构怎么设计、功耗计 SDK 的调用有哪些坑、以及实测下来 AI 调用这套工具时的真实表现。适合有 Python 基础、对硬件测量和 AI Agent 工具链都感兴趣的人看。哪怕你之前没写过 MCP 服务端,跟着思路走也能理解整套逻辑。
2. 为什么是 MCP,而不是直接写个脚本给 AI 跑
2.1 脚本方案的天花板在哪里
最直接的做法是写一个 Python 脚本,读功耗计数据,输出 JSON,然后让 AI 去执行这个脚本、解析结果。这个方案能跑通,但很快会碰到几个问题。
第一,AI 不知道脚本能干什么。你得在提示词里把脚本的用法、参数、输出格式全部描述一遍。脚本一多,提示词就爆炸了。第二,参数传递不灵活。今天想测五分钟,明天想测十分钟,后天想按电流阈值触发采样,每次都得改脚本或者加命令行参数,AI 还得猜参数怎么传。第三,没有标准化的错误处理。脚本报错了,AI 只能看到一堆 traceback,不知道是设备没连上还是权限不够。
MCP 解决的正是这几个问题。它把"能力"抽象成**工具(Tool)和资源(Resource)**两类东西。工具是 AI 可以主动调用的函数,带明确的输入 schema 和输出格式;资源是 AI 可以读取的数据源,比如设备列表、历史测量记录。AI 通过协议自带的描述信息就能知道每个工具干什么、要传什么参数,不需要你在提示词里反复解释。
2.2 MCP 服务端的两种通信方式
MCP 服务端和客户端之间主要有两种通信方式:stdio和HTTP/SSE。stdio 方式是服务端作为子进程启动,通过标准输入输出和客户端通信,适合本地工具集成;HTTP 方式则是服务端独立运行,客户端通过网络连接,适合远程调用或多客户端共享。
对于功耗计这种本地 USB 设备,stdio 是更自然的选择。设备插在这台机器上,服务端也跑在这台机器上,没必要绕一圈网络。而且 stdio 方式下,服务端的生命周期由客户端管理,启动和关闭都很干净,不会出现设备句柄泄漏的问题。
不过要注意一点:stdio 模式下服务端不能往 stdout 打印任何非协议内容。很多人调试时习惯性print()一下,结果直接把协议流冲乱了,客户端解析失败。调试信息一律走 stderr,这是硬规矩。
2.3 工具粒度怎么切
设计 MCP 工具时,粒度是个关键决策。切得太粗,一个工具干太多事,AI 不好组合;切得太细,工具数量爆炸,AI 选择困难。
我的做法是按"测量动作"来切,而不是按"设备寄存器"来切。具体来说,暴露这几个工具:
| 工具名 | 作用 | 关键参数 |
|---|---|---|
list_devices | 列出当前连接的功耗计 | 无 |
read_once | 读取一次瞬时数据 | 设备索引 |
sample_for | 持续采样指定时长 | 设备索引、时长、采样间隔 |
get_statistics | 对采样结果做统计 | 采样数据或时间范围 |
detect_spikes | 检测电流尖峰 | 阈值、时间窗口 |
这样切的好处是,AI 可以先list_devices确认设备在不在,再sample_for采一段数据,然后get_statistics和detect_spikes做分析。每个工具职责单一,组合起来却能完成复杂任务。而且sample_for返回的是原始采样点,get_statistics接受这些点做计算,AI 可以在中间插入自己的判断逻辑,比如"如果平均电流超过 100mA 就再采一次"。
3. 用 ctypes 打通 Python 和功耗计 SDK
3.1 为什么不用现成的 Python 库
IoT Power 这类设备通常提供的是 C 语言写的动态库(Windows 下是.dll,Linux 下是.so),配套的 Python 封装要么没有,要么年久失修。我查过一圈,能找到的 Python 绑定基本都是个人维护的,接口不全,而且对新固件版本支持滞后。
这种情况下,ctypes是最务实的选择。它是 Python 标准库的一部分,不需要额外安装,直接加载动态库、声明函数签名、调用就行。虽然写起来比用现成库麻烦一点,但胜在可控——设备 SDK 更新了,我改几行声明就能跟上,不用等别人发新版。
3.2 加载动态库和声明函数原型
先看加载部分。假设 SDK 提供的库文件叫iotpower.dll(Linux 下对应libiotpower.so),核心代码大概是这样:
import ctypes import ctypes.util import platform import os def load_sdk(): system = platform.system() if system == "Windows": lib_name = "iotpower.dll" elif system == "Linux": lib_name = "libiotpower.so" elif system == "Darwin": lib_name = "libiotpower.dylib" else: raise RuntimeError(f"不支持的系统: {system}") lib_path = os.path.join(os.path.dirname(__file__), "sdk", lib_name) if not os.path.exists(lib_path): found = ctypes.util.find_library("iotpower") if found: lib_path = found else: raise FileNotFoundError(f"找不到功耗计 SDK: {lib_name}") return ctypes.CDLL(lib_path)这里有个细节:ctypes.CDLL和ctypes.WinDLL在 Windows 上行为不同。如果 SDK 用的是 stdcall 调用约定,得用WinDLL;如果是 cdecl,用CDLL。大部分现代 SDK 都是 cdecl,但保险起见,可以先试CDLL,报错再换。我实测这台设备的 SDK 是 cdecl,所以直接用CDLL没问题。
加载之后要声明函数原型。这一步非常关键,不声明原型直接调用,在 64 位系统上很容易出问题,因为指针会被截断成 32 位。声明方式:
lib = load_sdk() lib.iotpower_open.argtypes = [ctypes.c_int] lib.iotpower_open.restype = ctypes.c_void_p lib.iotpower_close.argtypes = [ctypes.c_void_p] lib.iotpower_close.restype = ctypes.c_int lib.iotpower_read.argtypes = [ctypes.c_void_p, ctypes.POINTER(Measurement)] lib.iotpower_read.restype = ctypes.c_intargtypes和restype一定要写全。我见过太多人只写argtypes不写restype,结果返回值是个指针的时候,Python 默认按int解析,高位丢失,拿到的地址是错的,后面一访问就段错误。
3.3 用 Structure 映射设备返回的数据结构
功耗计 SDK 返回的测量数据通常是个结构体。用ctypes.Structure来映射:
class Measurement(ctypes.Structure): _fields_ = [ ("voltage_mv", ctypes.c_uint32), ("current_ua", ctypes.c_uint32), ("power_mw", ctypes.c_uint32), ("timestamp_ms", ctypes.c_uint64), ]字段类型和顺序必须和 C 头文件里完全一致,一个字节都不能差。对齐方式也要注意,C 编译器默认会对齐结构体成员,ctypes默认也是对齐的,所以一般不用手动加_pack_。但如果 SDK 头文件里明确写了#pragma pack(1),那 Python 这边也得加_pack_ = 1,否则字段偏移全错。
提示:验证结构体大小是个好习惯。用
ctypes.sizeof(Measurement)打印出来,和 C 那边sizeof的结果对一下。对不上就说明字段类型或对齐有问题,别急着往下写。
3.4 设备打开与句柄管理
iotpower_open返回的是一个void*句柄,后续所有操作都要传这个句柄。这里有个容易踩的坑:句柄不是线程安全的。如果你在 MCP 服务端里用了多线程,多个线程同时用同一个句柄读数据,轻则数据错乱,重则直接崩溃。
我的处理方式是给每个设备句柄配一把锁:
import threading class DeviceHandle: def __init__(self, index): self.index = index self.handle = lib.iotpower_open(index) if not self.handle: raise RuntimeError(f"打开设备 {index} 失败") self.lock = threading.Lock() def read(self): with self.lock: m = Measurement() ret = lib.iotpower_read(self.handle, ctypes.byref(m)) if ret != 0: raise RuntimeError(f"读取失败,错误码 {ret}") return m def close(self): with self.lock: if self.handle: lib.iotpower_close(self.handle) self.handle = None锁的粒度控制在单次读取操作上,不要锁住整个采样循环。否则一个长时间采样会阻塞其他所有操作,AI 想同时查个设备状态都查不了。
4. MCP 服务端的骨架与工具注册
4.1 服务端初始化与协议握手
MCP 服务端的核心是处理客户端发来的 JSON-RPC 消息。用官方 Python SDK 的话,初始化大概是这样:
from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app = Server("iotpower-mcp") @app.list_tools() async def list_tools(): return [ Tool( name="list_devices", description="列出当前连接的 IoT Power 功耗计设备", inputSchema={"type": "object", "properties": {}}, ), Tool( name="sample_for", description="持续采样指定时长,返回原始采样点", inputSchema={ "type": "object", "properties": { "device_index": {"type": "integer", "description": "设备索引"}, "duration_sec": {"type": "number", "description": "采样时长(秒)"}, "interval_ms": {"type": "integer", "description": "采样间隔(毫秒)", "default": 100}, }, "required": ["device_index", "duration_sec"], }, ), # ... 其他工具 ]inputSchema用的是 JSON Schema 格式,AI 客户端会根据这个 schema 来生成调用参数。描述要写清楚,尤其是单位——duration_sec是秒不是毫秒,interval_ms是毫秒不是秒,这种地方不写明白,AI 很容易搞混。
4.2 工具调用的分发逻辑
工具调用的入口是一个统一的 handler:
@app.call_tool() async def call_tool(name: str, arguments: dict): if name == "list_devices": devices = scan_devices() return [TextContent(type="text", text=json.dumps(devices, ensure_ascii=False))] elif name == "sample_for": idx = arguments["device_index"] duration = arguments["duration_sec"] interval = arguments.get("interval_ms", 100) samples = await sample_device(idx, duration, interval) return [TextContent(type="text", text=json.dumps(samples, ensure_ascii=False))] # ... 其他工具 else: raise ValueError(f"未知工具: {name}")注意sample_for是个耗时操作。如果采样五分钟,这个 handler 就会阻塞五分钟。MCP 协议本身支持异步,所以采样循环里要定期await asyncio.sleep(0)让出控制权,否则整个服务端会卡死,客户端发来的其他请求都处理不了。
4.3 采样循环的节奏控制
采样间隔的控制比想象中微妙。你不能简单地time.sleep(interval),因为读取本身也要耗时。如果读取耗时 20ms,间隔设 100ms,实际采样周期就变成了 120ms,长期采样下来时间戳会漂移。
正确的做法是基于绝对时间计算下一次采样点:
import time async def sample_device(index, duration_sec, interval_ms): handle = get_handle(index) interval = interval_ms / 1000.0 start = time.monotonic() end = start + duration_sec samples = [] next_tick = start while time.monotonic() < end: m = handle.read() samples.append({ "t": round(time.monotonic() - start, 4), "voltage_v": m.voltage_mv / 1000.0, "current_ma": m.current_ua / 1000.0, "power_mw": m.power_mw, }) next_tick += interval sleep_time = next_tick - time.monotonic() if sleep_time > 0: await asyncio.sleep(sleep_time) else: # 落后了,跳过等待直接采下一个 pass return samples用time.monotonic()而不是time.time(),因为前者不受系统时间调整影响,采样时间轴更稳定。next_tick累加而不是基于当前时间重算,能避免累积误差。
4.4 资源暴露:让 AI 主动查设备状态
除了工具,MCP 还支持资源(Resource)。资源适合放那些 AI 可能需要反复读取、但不涉及动作的数据。比如当前设备列表、最近一次采样的摘要。
@app.list_resources() async def list_resources(): return [ Resource( uri="iotpower://devices", name="已连接设备列表", mimeType="application/json", ), Resource( uri="iotpower://last-sample/summary", name="最近一次采样摘要", mimeType="application/json", ), ]资源的好处是 AI 可以在不调用工具的情况下直接读取,减少不必要的工具调用开销。比如 AI 想确认设备还在不在,读一下iotpower://devices就行,不用调list_devices。
5. 实测中踩到的坑与排查过程
5.1 设备打开失败:权限与驱动问题
第一次跑的时候,iotpower_open一直返回空句柄。排查过程是这样的:
先确认设备在系统里能不能被识别。Windows 下看设备管理器,Linux 下lsusb。设备在,说明硬件连接没问题。然后检查权限,Linux 下普通用户默认没有 USB 设备访问权限,需要配 udev 规则。Windows 下一般是驱动没装对,设备被识别成了未知设备。
这一步的教训是:MCP 服务端启动时要做一次设备自检,把设备状态作为资源暴露出去。AI 调用工具失败时,能先读资源确认是设备问题还是逻辑问题,而不是拿到一个笼统的报错。
5.2 采样数据跳变:缓冲区没清
采样跑起来之后,发现数据偶尔会跳变——电压突然从 5V 跳到 0V,下一拍又跳回来。一开始怀疑是设备固件问题,后来发现是 SDK 内部有个环形缓冲区,如果读取速度跟不上设备上报速度,缓冲区会溢出,读到的就是旧数据和新数据混在一起。
解决办法是每次开始采样前先调一次iotpower_flush(如果 SDK 提供的话),把缓冲区清空。如果 SDK 没提供这个接口,就连续快速读几次丢弃,直到读到的数据时间戳是递增的。
def flush_buffer(handle, max_attempts=10): last_ts = 0 for _ in range(max_attempts): m = handle.read() if m.timestamp_ms > last_ts: last_ts = m.timestamp_ms else: break5.3 AI 调用参数格式错误
工具注册好之后,让 AI 试着调用。第一次它传的duration_sec是字符串"5"而不是数字5。JSON Schema 里写的是"type": "number",但 AI 有时候会按自己的理解传字符串。
处理方式是在 handler 里做一次类型转换和校验:
def coerce_number(value, name): try: return float(value) except (TypeError, ValueError): raise ValueError(f"参数 {name} 必须是数字,收到: {value!r}")不要指望 AI 每次都传对类型,服务端做一层防御性转换是必要的。同时错误信息要写清楚,AI 看到"参数 duration_sec 必须是数字"之后,下一次调用通常就能改对。
5.4 长时间采样导致客户端超时
采样五分钟,客户端等了两分钟就超时断开了。这是因为 MCP 客户端对工具调用有超时限制,具体时长取决于客户端实现。
解决方案有两个方向。一是把长采样拆成多次短采样,AI 自己循环调用;二是服务端支持进度通知,定期往客户端发进度消息,让客户端知道任务还在跑。MCP 协议里有notifications/progress这个机制,用起来大概是这样:
async def sample_with_progress(index, duration_sec, interval_ms, progress_token): total_ticks = int(duration_sec / (interval_ms / 1000.0)) for i in range(total_ticks): # ... 采样 if i % 10 == 0: await app.request_context.session.send_progress_notification( progress_token=progress_token, progress=i, total=total_ticks, )实测下来,带进度通知的长采样,客户端不会再超时断开。这个机制值得加上,尤其是采样时长超过一分钟的场景。
6. 让 AI 真正"看懂"功耗数据的几个设计细节
6.1 返回数据的单位要统一且明确
功耗计 SDK 返回的原始数据单位五花八门:电压是毫伏,电流是微安,功率是毫瓦。如果直接把这些数字丢给 AI,它很容易在换算上出错。比如把current_ua当成毫安,算出来的功耗就差了 1000 倍。
我的做法是在服务端统一换算成伏、毫安、毫瓦,并且在字段名里带上单位:
{ "t": 0.1, "voltage_v": 5.02, "current_ma": 123.4, "power_mw": 619.5 }字段名带单位,AI 就不会猜错。这个细节看起来小,但实测能显著减少 AI 的换算错误。
6.2 统计工具要返回分布信息,而不只是平均值
一开始get_statistics只返回平均值,结果 AI 拿到"平均电流 120mA"之后,完全没法判断功耗是否稳定。后来改成返回一组分布指标:
| 指标 | 含义 | 用途 |
|---|---|---|
| mean | 平均值 | 整体功耗水平 |
| median | 中位数 | 排除尖峰后的典型值 |
| std | 标准差 | 波动程度 |
| min / max | 极值 | 范围边界 |
| p95 / p99 | 95/99 分位 | 尖峰参考 |
有了这些,AI 就能说出"平均电流 120mA,但 p99 达到 350mA,说明存在周期性尖峰"这种有信息量的结论,而不是干巴巴一个平均数。
6.3 尖峰检测的阈值要可配置
detect_spikes的阈值不能写死。不同设备、不同场景,什么算"尖峰"完全不一样。待机电流 1mA 的设备,10mA 就是大尖峰;满载 2A 的设备,2.5A 才算异常。
所以阈值设计成参数,同时给一个基于统计的默认值:threshold = mean + 3 * std。AI 可以先不传阈值,用默认值跑一遍,看看结果再决定要不要调整。
def detect_spikes(samples, threshold=None, window=5): currents = [s["current_ma"] for s in samples] if threshold is None: mean = statistics.mean(currents) std = statistics.pstdev(currents) threshold = mean + 3 * std spikes = [] for i, s in enumerate(samples): if s["current_ma"] > threshold: spikes.append({ "t": s["t"], "current_ma": s["current_ma"], "threshold": round(threshold, 2), }) return spikes6.4 采样数据的返回量要控制
采样五分钟、间隔 100ms,就是 3000 个采样点。全量返回给 AI,token 消耗巨大,而且 AI 也看不过来。我的处理是:原始数据存到临时文件,返回给 AI 的是摘要 + 文件路径。AI 需要细节时,再通过资源读取文件。
def save_samples(samples, path): with open(path, "w", encoding="utf-8") as f: json.dump(samples, f, ensure_ascii=False) def summarize(samples): currents = [s["current_ma"] for s in samples] return { "count": len(samples), "duration_sec": samples[-1]["t"] - samples[0]["t"], "current_ma": { "mean": round(statistics.mean(currents), 2), "max": round(max(currents), 2), "min": round(min(currents), 2), }, "raw_file": "/tmp/iotpower_sample_xxx.json", }这样 AI 拿到的是几十个 token 的摘要,需要深入分析时再去读文件。实测这个设计让长采样的可用性提升了很多。
7. 这套 MCP 服务端实际用起来是什么体验
7.1 典型工作流
现在我的使用流程是这样的:打开支持 MCP 的 AI 客户端,连上这个服务端,然后直接说需求。
比如:"帮我测一下开发板在空闲状态下的功耗,跑三分钟,采样间隔 200ms,看看有没有超过 200mA 的尖峰。"
AI 会自己规划:先list_devices确认设备,再sample_for采三分钟,然后get_statistics算统计,最后detect_spikes查尖峰。整个过程我不需要碰任何界面,最后拿到一份带数据的结论。
再比如:"对比一下这块板子在 WiFi 开启和关闭两种状态下的功耗差异。"AI 会分别采两次数据,然后自己算差值。这种对比分析以前要手动做表格,现在一句话就搞定。
7.2 AI 调用工具时的真实表现
实测下来,AI 对这套工具的调用准确率相当高,但有几个地方需要留意。
一是设备索引。如果只连了一台设备,AI 有时会忘记传device_index,或者传个0但实际索引是1。解决办法是在list_devices的返回里明确写出索引,并且在工具描述里强调"索引来自 list_devices 的返回"。
二是采样时长的单位。虽然 schema 里写了秒,AI 偶尔还是会按毫秒理解。后来我在描述里加了一句"例如 300 表示 5 分钟",错误率明显下降。
三是多步任务的规划。AI 有时候会跳过list_devices直接调sample_for,如果设备没连上就会报错。这个不算大问题,报错之后 AI 通常会回头去查设备列表。但如果想让它一次成功,可以在sample_for的错误信息里提示"请先调用 list_devices 确认设备"。
7.3 性能与资源占用
服务端本身很轻,Python 进程常驻内存大概 30-50MB。采样时 CPU 占用取决于采样间隔,100ms 间隔下基本可以忽略。真正占资源的是数据存储——如果连续采样几小时,原始数据文件会很大。
我的做法是给采样数据加个自动清理:超过 24 小时的文件在服务端启动时删掉。同时限制单次采样的最大时长,比如 30 分钟,超过就拒绝,避免 AI 误传一个超大时长把磁盘写满。
8. 几个可以继续深挖的方向
这套东西跑通之后,能扩展的地方不少。
多设备并行采样。现在sample_for一次只采一台设备。如果同时测多块板子,可以加一个sample_multi工具,内部用线程池并行采样,返回一个设备到数据的映射。这样对比测试的效率会高很多。
触发式采样。现在的采样是定时的,但有些场景需要"电流超过阈值才开始记录"。可以加一个sample_on_trigger工具,参数里带触发条件,服务端内部轮询直到条件满足才开始正式采样。这对抓瞬态功耗特别有用。
和历史数据对比。把每次采样的摘要存到本地数据库,加一个compare_with_history工具,AI 就能说出"这次测的功耗比上周高了 15%"这种结论。做硬件迭代的时候,这个功能价值很大。
功耗曲线可视化。虽然 MCP 本身不直接支持返回图片,但可以生成一个临时 HTML 文件,把采样数据用图表渲染出来,然后把文件路径返回给 AI。AI 可以进一步处理这个文件,或者提示用户打开查看。
这套 MCP 服务端我从有想法到跑通大概花了一个周末,其中一半时间花在 ctypes 调 SDK 的调试上。如果你手头也有类似的硬件设备,SDK 只提供 C 接口,那这套"ctypes 封装 + MCP 暴露"的思路可以直接复用。核心就三件事:把 C 结构体映射对、把句柄管理好、把返回数据的单位标清楚。剩下的就是让 AI 自己去折腾了。