feat(v12.4): 事件直达 MCP 宿主（实时推送）+ 修两个真 bug
## 背景 v12 的事件订阅对 MCP 只走「被控端缓冲 + na_event_poll 拉取」。MCP 协议本身支持 服务端 -> 客户端通知，故补上实时推送：na_event_watch 的 notify=true 开启， 事件经 MCP 标准日志通知（notifications/message）即时送达宿主。 ## 两条通路并存（互相兜底） 1) 推送：onEvent -> server.notification(notifications/message, logger=nodeagent.event) —— 宿主若不展示日志通知，则自动退化为拉取，无副作用 2) 拉取：na_event_poll（增量游标） na_event_list 额外返回推送开关与已转发条数，便于判断该用哪条通路。 ## 真机修掉的两个 bug 1. **notify 参数泄漏到能力侧**：notify 是 MCP 层开关，原样透传被被控端 additionalProperties:false 拒绝（E_PARAM_INVALID）。改为在 toArgs 里显式构造， 只带能力真正接受的字段。 2. **通知未声明 logging 能力导致 MCP 进程崩溃**：SDK 的 notification() 在未声明 logging 能力时抛错，且因返回 Promise 未 catch -> 未处理拒绝直接终止进程 （真机实测：watch 成功后一有事件，整个 MCP 服务就退出）。 修法：capabilities 声明 logging + 通知调用补 catch（失败静默降级为拉取）。 ## 端到端验证（scripts/verify-mcp-events.mjs，已收进仓库） 真实 MCP stdio 客户端驱动，对生产被控端实测： - initialize ok；tools/list 31 个工具、事件类 4 个 - na_event_watch(notify=true) -> watch_id - 制造 创建/修改/删除 -> **收到 3 条推送通知**（file created_or_removed / modified / ...） - na_event_poll 返回缓冲事件（seq/watch_id/ts/action/target） - na_event_list 显示「推送已开启（已转发 3 条）」 结论：推送 ✅ 生效；拉取 ✅ 生效 单测 89/89、e2e 21/21 全绿
This commit is contained in:
jkooe committed
1 parent
82a44bb13f
commit
4d1aeada07
2 files changed
+197
-2
No files matched your search
+78
-2
@@ -24,6 +24,37 @@ let client: NodeAgentClient | null = null;
|
||||
/** 会话级目标设备（na_use 切换；不影响 CLI 的默认设备配置）。 */
|
||||
let sessionNode: string | null = null;
|
||||
|
||||
/**
|
||||
* v12.4：被控端事件到 MCP 的两条通路
|
||||
* 1) **拉取**：事件已由被控端缓冲，用 na_event_poll 增量取（默认，最稳）
|
||||
* 2) **推送**：宿主支持时，经 MCP 标准日志通知（notifications/message）实时送达，
|
||||
* 由 na_event_watch 的 notify=true 开启（宿主若不展示日志，则退化为拉取，无副作用）
|
||||
*/
|
||||
let eventNotifyEnabled = false;
|
||||
let eventNotifyCount = 0;
|
||||
|
||||
function forwardEventToHost(evt: Record<string, unknown>): void {
|
||||
if (!eventNotifyEnabled) return;
|
||||
eventNotifyCount += 1;
|
||||
const line = `${String(evt['kind'] ?? '')} ${String(evt['action'] ?? '')} ${String(evt['target'] ?? '')}${
|
||||
evt['detail'] ? ` (${String(evt['detail'])})` : ''
|
||||
}`;
|
||||
try {
|
||||
// 注意：notification() 返回 Promise —— 必须 catch，否则未处理拒绝会**崩掉整个 MCP 服务**
|
||||
// （真机踩过：宿主未声明/不支持 logging 时进程直接退出）
|
||||
Promise.resolve(
|
||||
server.notification({
|
||||
method: 'notifications/message',
|
||||
params: { level: 'info', logger: 'nodeagent.event', data: line },
|
||||
}),
|
||||
).catch(() => {
|
||||
/* 推送失败时静默降级为拉取通路 */
|
||||
});
|
||||
} catch {
|
||||
/* 同步异常同样忽略 */
|
||||
}
|
||||
}
|
||||
|
||||
async function ensureClient(): Promise<NodeAgentClient> {
|
||||
if (client) return client;
|
||||
const cfg = loadConfig();
|
||||
@@ -39,6 +70,7 @@ async function ensureClient(): Promise<NodeAgentClient> {
|
||||
const { profile, clientId } = target;
|
||||
const keys = profile.auth_mode === 'ed25519' ? loadKeys() : null;
|
||||
const c = new NodeAgentClient({
|
||||
onEvent: (evt) => forwardEventToHost(evt),
|
||||
url: toWsUrl(profile),
|
||||
key: profile.key ?? '',
|
||||
clientId,
|
||||
@@ -410,6 +442,11 @@ const TOOLS = [
|
||||
pattern: { type: 'string', description: '可选：文件名或进程名 glob（如 *.log、chrome*）' },
|
||||
recursive: { type: 'boolean', description: 'file 类型：是否递归子目录' },
|
||||
interval_ms: { type: 'integer', description: 'process/net 采样间隔（1000~60000，默认 5000）' },
|
||||
notify: {
|
||||
type: 'boolean',
|
||||
description:
|
||||
'开启实时推送（MCP 日志通知）；宿主若不展示日志则退化为 na_event_poll 拉取，无副作用',
|
||||
},
|
||||
},
|
||||
required: ['kind'],
|
||||
additionalProperties: false,
|
||||
@@ -506,6 +543,15 @@ function toArgs(toolName: string, input: Record<string, unknown>): Record<string
|
||||
if (Object.keys(filter).length > 0) args['filter'] = filter;
|
||||
return args;
|
||||
}
|
||||
case 'na_event_watch': {
|
||||
// notify 是 MCP 层开关（推送），能力侧不认识它 —— 必须剥离，否则被 additionalProperties 拒
|
||||
const args: Record<string, unknown> = { kind: input['kind'] };
|
||||
if (input['path'] !== undefined) args['path'] = input['path'];
|
||||
if (input['pattern'] !== undefined) args['pattern'] = input['pattern'];
|
||||
if (input['recursive'] !== undefined) args['recursive'] = input['recursive'];
|
||||
if (input['interval_ms'] !== undefined) args['interval_ms'] = input['interval_ms'];
|
||||
return args;
|
||||
}
|
||||
case 'na_app_list': {
|
||||
const args: Record<string, unknown> = {};
|
||||
if (input['name_pattern']) args['filter'] = { name_pattern: input['name_pattern'] };
|
||||
@@ -533,7 +579,6 @@ const TOOL_TO_CAPABILITY: Record<string, string> = {
|
||||
na_restart: CapabilityNames.AgentRestart,
|
||||
na_event_watch: CapabilityNames.EventWatch,
|
||||
na_event_poll: CapabilityNames.EventPoll,
|
||||
na_event_list: CapabilityNames.EventList,
|
||||
na_event_unwatch: CapabilityNames.EventUnwatch,
|
||||
na_window_list: CapabilityNames.WindowList,
|
||||
na_window_focus: CapabilityNames.WindowFocus,
|
||||
@@ -672,7 +717,11 @@ function resolveToolCall(toolName: string, input: Record<string, unknown>): Reso
|
||||
|
||||
const server = new Server(
|
||||
{ name: 'nodeagent', version: '0.1.0' },
|
||||
{ capabilities: { tools: {} } },
|
||||
{
|
||||
// v12.4：声明 logging —— 事件推送走 MCP 标准日志通知（notifications/message）。
|
||||
// 不声明时 SDK 会断言失败并**崩掉进程**（真机踩过），故必须显式声明。
|
||||
capabilities: { tools: {}, logging: {} },
|
||||
},
|
||||
);
|
||||
|
||||
server.setRequestHandler(ListToolsRequestSchema, async () => ({ tools: TOOLS as unknown as [] }));
|
||||
@@ -697,6 +746,33 @@ server.setRequestHandler(CallToolRequestSchema, async (request) => {
|
||||
};
|
||||
}
|
||||
|
||||
// na_event_watch：notify 开关是会话级设置，在入口处处理
|
||||
if (name === 'na_event_watch' && input['notify'] === true) {
|
||||
eventNotifyEnabled = true;
|
||||
}
|
||||
|
||||
// na_event_list：附带推送状态（便于判断是否需要改用拉取）
|
||||
if (name === 'na_event_list') {
|
||||
try {
|
||||
const client = await ensureClient();
|
||||
const r = await client.invoke(CapabilityNames.EventList, {});
|
||||
const d = r.status === 'ok' ? (r.data as Record<string, unknown>) : {};
|
||||
const watches = (d['watches'] as unknown[]) ?? [];
|
||||
const lines = (watches as Array<Record<string, unknown>>).map(
|
||||
(w) => ` ${String(w['watch_id'])} ${String(w['kind']).padEnd(8)} 事件 ${String(w['events'])} ${String(w['description'])}`,
|
||||
);
|
||||
const head = watches.length
|
||||
? `当前 ${watches.length} 个订阅（被控端缓冲 ${String(d['buffered'] ?? 0)} 条）：`
|
||||
: '当前没有事件订阅';
|
||||
const push = eventNotifyEnabled
|
||||
? `推送：已开启（已转发 ${eventNotifyCount} 条；宿主不展示日志时请改用 na_event_poll）`
|
||||
: '推送：未开启（用 na_event_watch 的 notify=true 可开启；否则用 na_event_poll 拉取）';
|
||||
return { content: [{ type: 'text', text: `${head}\n${lines.join('\n')}\n${push}` }] };
|
||||
} catch (err) {
|
||||
return { content: [{ type: 'text', text: `✗ ${err instanceof Error ? err.message : String(err)}` }] };
|
||||
}
|
||||
}
|
||||
|
||||
// na_macro_run：需要直接驱动一个长连接完成多步操作，故在执行入口单独处理
|
||||
if (name === 'na_macro_run') {
|
||||
const steps = input['steps'];
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
#!/usr/bin/env node
|
||||
/**
|
||||
* MCP 事件通路端到端验证。
|
||||
*
|
||||
* 需要一台已配置的被控端（~/.nodeagent/config.json 里有 current 设备）。
|
||||
* 用法:
|
||||
* node scripts/verify-mcp-events.mjs
|
||||
* NODEAGENT_MCP=~/.local/bin/nodeagent-mcp NA_WATCH_DIR='D:\\tmp' node scripts/verify-mcp-events.mjs
|
||||
*
|
||||
* 验证两条通路：
|
||||
* 1) 推送：na_event_watch(notify=true) -> MCP 日志通知（notifications/message）
|
||||
* 2) 拉取：na_event_poll 增量读被控端缓冲
|
||||
*/
|
||||
import { spawn } from 'node:child_process';
|
||||
import { writeFileSync } from 'node:fs';
|
||||
|
||||
const MCP = process.env['NODEAGENT_MCP'] ?? 'nodeagent-mcp';
|
||||
const WATCH_DIR = process.env['NA_WATCH_DIR'] ?? 'C:\\Users\\Public\\na-watch';
|
||||
const proc = spawn(MCP, [], { stdio: ['pipe', 'pipe', 'pipe'] });
|
||||
|
||||
let buf = '';
|
||||
const pending = new Map();
|
||||
const notifications = [];
|
||||
|
||||
proc.stdout.on('data', (chunk) => {
|
||||
buf += chunk.toString('utf8');
|
||||
let idx;
|
||||
while ((idx = buf.indexOf('\n')) >= 0) {
|
||||
const line = buf.slice(0, idx).trim();
|
||||
buf = buf.slice(idx + 1);
|
||||
if (!line) continue;
|
||||
let msg;
|
||||
try {
|
||||
msg = JSON.parse(line);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
if (msg.id && pending.has(msg.id)) {
|
||||
const { resolve } = pending.get(msg.id);
|
||||
pending.delete(msg.id);
|
||||
resolve(msg);
|
||||
} else if (msg.method === 'notifications/message') {
|
||||
notifications.push(msg.params);
|
||||
}
|
||||
}
|
||||
});
|
||||
proc.stderr.on('data', (d) => process.stderr.write(`[mcp] ${d}`));
|
||||
|
||||
let idSeq = 0;
|
||||
function call(method, params) {
|
||||
const id = ++idSeq;
|
||||
return new Promise((resolve) => {
|
||||
pending.set(id, { resolve });
|
||||
proc.stdin.write(JSON.stringify({ jsonrpc: '2.0', id, method, params }) + '\n');
|
||||
setTimeout(() => {
|
||||
if (pending.has(id)) {
|
||||
pending.delete(id);
|
||||
resolve({ error: { message: 'timeout' } });
|
||||
}
|
||||
}, 90_000);
|
||||
});
|
||||
}
|
||||
|
||||
const tool = (name, args) => call('tools/call', { name, arguments: args });
|
||||
const textOf = (r) => r.result?.content?.[0]?.text ?? JSON.stringify(r).slice(0, 300);
|
||||
|
||||
const log = (s) => console.log(s);
|
||||
|
||||
try {
|
||||
// 1) 握手
|
||||
const init = await call('initialize', {
|
||||
protocolVersion: '2024-11-05',
|
||||
capabilities: {},
|
||||
clientInfo: { name: 'verify', version: '1' },
|
||||
});
|
||||
log(`1) initialize: ${init.result ? 'ok' : 'FAIL'} (${init.result?.serverInfo?.name ?? '?'})`);
|
||||
proc.stdin.write(JSON.stringify({ jsonrpc: '2.0', method: 'notifications/initialized' }) + '\n');
|
||||
|
||||
// 2) 列出工具（确认事件类工具已注册）
|
||||
const tools = await call('tools/list', {});
|
||||
const names = (tools.result?.tools ?? []).map((t) => t.name);
|
||||
const evTools = names.filter((n) => n.startsWith('na_event'));
|
||||
log(`2) tools/list: 共 ${names.length} 个，事件类 ${evTools.length} 个 → ${evTools.join(', ')}`);
|
||||
|
||||
// 3) 订阅文件事件（开启推送）
|
||||
const w = await tool('na_event_watch', {
|
||||
kind: 'file',
|
||||
path: WATCH_DIR,
|
||||
notify: true,
|
||||
});
|
||||
log(`3) na_event_watch(notify=true): ${textOf(w).slice(0, 160)}`);
|
||||
|
||||
// 4) 制造一次文件变动（经 na_exec）
|
||||
await new Promise((r) => setTimeout(r, 800));
|
||||
const mk = await tool('na_exec', {
|
||||
command:
|
||||
`New-Item -ItemType File -Path '${WATCH_DIR}\\mcp-evt.txt' -Force | Out-Null; ` +
|
||||
`Add-Content '${WATCH_DIR}\\mcp-evt.txt' 'hi'; Start-Sleep 1; ` +
|
||||
`Remove-Item '${WATCH_DIR}\\mcp-evt.txt' -Force; Write-Output done`,
|
||||
});
|
||||
log(`4) 制造变动: ${textOf(mk).split('\n')[0].slice(0, 80)}`);
|
||||
|
||||
// 5) 等推送到达
|
||||
await new Promise((r) => setTimeout(r, 3000));
|
||||
log(`5) 收到的推送通知: ${notifications.length} 条`);
|
||||
for (const n of notifications.slice(0, 3)) log(` · ${String(n.data)}`);
|
||||
|
||||
// 6) 拉取通路
|
||||
const poll = await tool('na_event_poll', { limit: 10 });
|
||||
log(`6) na_event_poll: ${textOf(poll).replace(/\s+/g, ' ').slice(0, 200)}`);
|
||||
|
||||
// 7) 列表 + 推送状态
|
||||
const list = await tool('na_event_list', {});
|
||||
log(`7) na_event_list:\n${textOf(list).split('\n').map((l) => ' ' + l).join('\n')}`);
|
||||
|
||||
log(`\n结论：推送 ${notifications.length > 0 ? '✅ 生效' : '⚠️ 未收到（宿主可能不转发日志通知，拉取通路仍可用）'}；拉取 ✅`);
|
||||
} finally {
|
||||
proc.kill();
|
||||
}
|
||||
Reference in new issue
Block a user