Chrome AI Helper Agent Skill

本文档是 AI Agent 的 Skill 文档。Agent 通过访问 http://localhost:3000/ 学习如何接入和使用 Chrome AI Helper 系统。

文档包含：连接方式、MCP 工具列表、JSON-RPC 协议格式、交互规范、确认机制、关键约束。

不包含：plumber/task_watcher 完整源码（这些脚本部署在本地 ~/.chrome-ai-helper/ 目录，Agent 不需要逐行阅读源代码）。 脚本的部署指引见下方「脚本部署」章节。

🔴 最高优先级：所有浏览器操作必须通过用户侧 Chrome 浏览器完成。
Agent 严禁使用任何内置浏览器、web_search、web_fetch、或自行拼接 URL 访问网页。 你必须调用 browser_* MCP 工具（browser_navigate、browser_click、browser_type 等） 通过 WebSocket 连接操控用户正在使用的 Chrome 浏览器。 用户的侧边栏发来任务时，你的浏览器操作必须在用户眼前的 Chrome 浏览器中执行。

⚠️ 本 profile 已禁用 Hermes 内置 browser toolset。所有浏览器操作必须走 MCP 注册的 browser_* 工具，通过 Proxy 桥端转发到用户侧 Chrome。

简单判断：如果用户看不到你打开的网页 → 你错了，你没有操控用户侧浏览器。

禁止猜测路径。Agent 不得通过臆测 URL 拼接、枚举路径等方式尝试访问页面。 所有页面操作（跳转、查询、表单提交、翻页等）必须通过模拟人类操控方式完成： 点击链接/按钮 → 填写表单 → 提交 → 等待页面响应。

Skill 更新同步。每次学习本 Skill 后，Agent 必须 对比之前的逻辑， 检查是否有不兼容的变更。如果 Skill 有更新，Agent 必须 根据最新版本修复本地相关逻辑。

Token 变更。当用户提供新的 Token，Agent 必须 先更新本地 token 文件 （~/.chrome-ai-helper/token），然后重新建立连接。plumber 的 watchdog 会自动检测 token 文件变更并重连。

⛔ 工具使用限制（严格遵守）：
• browser_execute_js — 最后手段，禁止模拟点击/输入，每个任务最多 5 次
• browser_screenshot — 仅在 DOM 无法回答问题时使用，每个任务最多 3 次
• browser_type — 使用 Input.insertText 一次性插入文本，不会重复字符
• browser_click — 真实 CDP 鼠标事件，selector 为必填 CSS 选择器字符串（不支持坐标）
• browser_get_dom — 优先于 screenshot，先用 DOM 文本理解页面内容
• browser_wait — 仅支持 element_visible 和 timeout 两种 condition

所有 Agent 类型使用统一流程：plumber 维持 WebSocket 连接，watchdog 事件驱动消息收发。 通信机制：plumber 通过 watchdog 监听 replies.jsonl 文件变更（fallback: 1s 轮询） 和 token 文件变更（fallback: 3s 轮询）。不同 Agent 的差异仅在于任务触发方式。

重要：plumber 只负责收发，不包含智能逻辑。智能逻辑（意图分类、工具调用）完全由 Agent 层完成。 task_watcher.py 的 prompt 是纯协议层，只负责让 Agent 调用 MCP 工具。用户任务的详细步骤和验证条件由用户从侧边栏提供。

Agent 学习完本 Skill 后，必须立即自动执行以下操作，不得等待用户指令：

MCP 管道层 vs Agent 智能层 — 严格分离。

MCP 管道脚本（纯收发层）

管道脚本只负责网络收发，禁止包含任何智能逻辑：

管道脚本 必须做:
  ✅ WebSocket 长连接 + 实时接收消息（ws://localhost:3000/agent-ws）
  ✅ 将收到的消息写入本地文件（~/.chrome-ai-helper/messages.jsonl）
  ✅ watchdog 监听 replies.jsonl 变更 → 通过 WebSocket 发送 Agent 回复
  ✅ watchdog 监听 token 文件变更 → 自动断开用新 token 重连
  ✅ token 管理（读取/保存）

管道脚本 禁止做:
  ❌ 意图分类（聊天 vs 任务判断）
  ❌ 生成回复内容（文字、总结、分析）
  ❌ 硬编码任何基于消息内容的业务逻辑
  ❌ 自动调用 browser_* 工具（必须由 Agent 层决策后调用）

Agent 智能层（LLM 会话）

Agent（LLM）是唯一的智能层，负责：

Agent 层 负责:
  ✅ 检查信号文件 /tmp/mcp_pending_mtime.txt 获知新消息
  ✅ 用自己的模型（LLM）进行意图识别
  ✅ 生成聊天回复内容
  ✅ 决策调用哪些 browser_* 工具
  ✅ 将 Agent 回复写入 replies.jsonl（plumber 自动通过 WS 发送）

❌ 禁止: 把意图判断逻辑写在管道脚本中

推荐目录结构

~/.chrome-ai-helper/
├── token                         # token 存储
├── sessions//token   # 多会话 token
├── messages.jsonl               # plumber 写入，Agent 读取
├── mcp-plumber.py               # 纯收发管道脚本（后台常驻）
├── task_watcher.py              # 调度层（仅 Hermes，后台常驻）
└── replies.jsonl                # Agent 写入回复，plumber 自动发送

5.1 框架式 Agent 工作流

plumber 通过 WebSocket 实时接收消息 → 写入 messages.jsonl + 更新信号文件。 Agent 自身负责检测信号文件并处理消息。

loop (推荐 watchdog 检测，fallback: 5s 轮询):
  1. 检查 plumber 进程是否存活
     pid_file = /tmp/mcp_plumber.pid
     if not pid_file.exists() 或进程不存活:
         → 重新启动 plumber

  2. 检测信号文件 /tmp/mcp_pending_mtime.txt
     if signal_file 无变化:
         continue

  3. 增量读取 messages.jsonl（用 last_read_offset 记录位置）

  4. 对每条新消息，使用 LLM 进行意图分类
     intent = classify_with_llm(msg)  # "chat" 或 "task"

  5a. "chat" → 生成回复，写入 replies.jsonl
      {"jsonrpc":"2.0","method":"notifications/agent_message",
       "params":{"type":"text","content":"..."}}
      # plumber watchdog 自动检测 replies.jsonl 变化并通过 WS 发送

  5b. "task" → 执行浏览器任务，参照下方交互规范
      # 每步写 agent_action 到 replies.jsonl
      # 完成后写 agent_message + task_completed

  6. 当前批次处理完毕 → 删除 /tmp/mcp_pending_mtime.txt → 回到步骤 1

5.2 Hermes Agent 工作流（被动启动模式）

plumber 负责 WS 收发，task_watcher 负责任务检测和 Agent 启动。 Agent 自身不需要轮询——由 task_watcher 被动启动。

流程:
  1. plumber WS 收到 user_task → 写入 messages.jsonl + 更新信号文件
  2. task_watcher watchdog 检测到信号文件变更（fallback: 3s 轮询）
  3. task_watcher 增量读取 messages.jsonl，找到 user_task
  4. task_watcher 写入 task_accepted + agent_message + agent_action 到 replies.jsonl
  5. task_watcher 构建 prompt，启动 hermes chat -q 子进程：
     hermes chat -s mcp-protocol-testing -m Lo-Mo -q 
  6. Hermes Agent 子进程：
     - 通过 MCP server (http://localhost:3000/mcp) 调用 browser_* 工具
     - 操作完成后写入 agent_message + task_completed 到 replies.jsonl
  7. plumber watchdog 检测 replies.jsonl 变更 → 通过 WS 发送到侧边栏

读写约定

每当 Agent 收到一条消息，必须通过 WebSocket 或 POST /mcp 回复到侧边栏聊天窗口。 侧边栏不会主动轮询，Agent 必须主动推送。

⚠️ 任务完成判定（必须遵守）

消息分类规则

Agent 必须使用 LLM 进行意图识别，不要依赖关键词匹配。

分类标准：

多意图拆解与序列执行

用户一条消息可能包含多个意图。Agent 必须用 LLM 拆解为意图列表，逐条执行。

示例（用户："打开百度，搜索今天的 AI 新闻，把第一条标题告诉我"）：

LLM 拆解意图:
  [
    {type:"task", content:"打开百度", depends_on:null},
    {type:"task", content:"搜索今天的AI新闻", depends_on:0},
    {type:"chat", content:"返回第一条新闻标题", depends_on:1}
  ]

Agent 执行:
  → agent_message("收到，共3个步骤：
1.打开百度
2.搜索AI新闻
3.返回第一条标题")
  → [步骤1] agent_action("navigate","打开百度首页","started") → completed
  → agent_message("百度首页已打开")
  → [步骤2] agent_action("type","输入搜索关键词","started") → completed
  → agent_message("搜索完成，找到5条结果")
  → [步骤3] agent_message("第一条标题：xxx")
  → task_completed

浏览器任务处理流程

⚠️ 通知方法差异（重要）

关键规则：要显示文字内容，必须使用 agent_message。task_completed/task_failed 只是状态标记。

agent_message 参数格式

# 正确格式
{"jsonrpc":"2.0","method":"notifications/agent_message",
 "params":{"type":"text","content":"任务已完成: 百度首页已打开"}}

# ❌ 错误（缺少 type 字段）
{"jsonrpc":"2.0","method":"notifications/agent_message",
 "params":{"text":"任务已完成"}}

agent_action 使用规则

# 动作开始前
{"jsonrpc":"2.0","method":"notifications/agent_action",
 "params":{"action":"navigate","description":"打开百度首页","status":"started"}}

# 动作完成后
{"jsonrpc":"2.0","method":"notifications/agent_action",
 "params":{"action":"navigate","description":"打开百度首页","status":"completed"}}

# 动作失败时
{"jsonrpc":"2.0","method":"notifications/agent_action",
 "params":{"action":"click","description":"点击搜索按钮","status":"failed"}}

协议版本：MCP 2025-11-25（JSON-RPC 2.0）
通信模式：Agent 通过 WebSocket (ws://localhost:3000/agent-ws) 双向通信
备用模式：Agent 也可通过 HTTP POST /mcp 发送请求
⚠️ 必须先建立 WebSocket 连接并认证，才能收发消息。

# Step 1: 连接 Agent WebSocket
wscat -c ws://localhost:3000/agent-ws
> {"type":"agent_auth","token":"your-token-here"}
# 收到 auth_result 后即可收发 JSON-RPC 消息

# Step 2: 初始化
{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"agent","version":"1.0"}}}

# Step 3: 获取工具列表
{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}

# Step 4: 调用工具
{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"browser_start_session","arguments":{"url":"https://example.com"}}}

# 备用：HTTP POST
curl -s -X POST http://localhost:3000/mcp \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/call",...}'

Agent 发送以下通知后，Proxy 会识别并转发到 Extension（不会继续转发到对端）：

Token 可能被 AI 平台内容过滤截断/丢失。建议将 Token 保存到本地文件，不依赖对话上下文传递。

存储路径：~/.chrome-ai-helper/token（推荐，单会话）或 ~/.chrome-ai-helper/sessions/<session_id>/token（多会话）。 plumber 的 watchdog 会自动检测 token 文件变更并重连。

以下两个脚本部署在 ~/.chrome-ai-helper/ 目录，源码在项目仓库 scripts/ 中版本管理。 依赖：pip install websocket-client watchdog

mcp-plumber.py — WebSocket 收发层

启动：python3 ~/.chrome-ai-helper/mcp-plumber.py --token-file ~/.chrome-ai-helper/token &
PID：/tmp/mcp_plumber.pid
职责：纯收发层。WebSocket 连接 → 消息写入 messages.jsonl → 更新信号文件；watchdog 监听 replies.jsonl 变更 → 通过 WS 发送 Agent 回复；watchdog 监听 token 文件变更 → 自动用新 token 重连。禁止包含任何智能逻辑。

#!/usr/bin/env python3
"""
Chrome AI Helper — MCP 管道脚本（纯收发层）
保存为 ~/.chrome-ai-helper/mcp-plumber.py

职责:
  - WebSocket 连接 (ws://localhost:3000/agent-ws) 实时接收消息 → messages.jsonl
  - 写入新消息后创建信号文件通知 Agent
  - watchdog 监听 replies.jsonl 变更 → 通过 WebSocket 发送 Agent 回复
  - watchdog 监听 token 文件变更 → 自动断开用新 token 重连
  - 禁止包含任何智能逻辑

启动: python3 ~/.chrome-ai-helper/mcp-plumber.py --token-file ~/.chrome-ai-helper/token
依赖: pip install websocket-client watchdog
"""

import json, os, sys, time, signal, threading
from pathlib import Path

# -- 依赖检查 --
try:
    import websocket
except ImportError:
    print("[plumber] ERROR: websocket-client not installed. pip install websocket-client", file=sys.stderr)
    sys.exit(1)

_WATCHDOG_AVAILABLE = False
try:
    from watchdog.observers import Observer
    from watchdog.events import FileSystemEventHandler
    _WATCHDOG_AVAILABLE = True
except ImportError:
    pass

# ─── 配置 ───────────────────────────────────────────────

def get_config():
    import argparse
    p = argparse.ArgumentParser(description="MCP Plumber - pure I/O pipe")
    p.add_argument("--session-id")
    p.add_argument("--token-file", help="Path to token file")
    p.add_argument("--proxy-url", default="ws://localhost:3000/agent-ws")
    p.add_argument("--data-dir", default=os.path.expanduser("~/.chrome-ai-helper"))
    args = p.parse_args()

    data_dir = Path(args.data_dir)
    if args.token_file:
        token_file = Path(args.token_file)
    elif args.session_id:
        token_file = data_dir / "sessions" / args.session_id / "token"
    else:
        sessions_dir = data_dir / "sessions"
        token_file = None
        if sessions_dir.exists():
            for d in sorted(sessions_dir.iterdir(), reverse=True):
                tf = d / "token"
                if tf.exists():
                    token_file = tf
                    break
        if not token_file:
            print("[plumber] ERROR: No token file found.", file=sys.stderr)
            sys.exit(1)

    return {
        "token_file": token_file.absolute(),
        "proxy_url": args.proxy_url.rstrip("/"),
        "messages_file": data_dir / "messages.jsonl",
        "replies_file": data_dir / "replies.jsonl",
        "mtime_file": Path("/tmp/mcp_pending_mtime.txt"),
        "pid_file": Path("/tmp/mcp_plumber.pid"),
    }


def load_token(token_file):
    if token_file.exists():
        return token_file.read_text().strip()
    return None

# ─── WebSocket 客户端 ────────────────────────────────────

class PlumberWsClient:
    def __init__(self, token_file: Path, proxy_url: str, messages_file: Path, mtime_file: Path):
        self.token_file = token_file
        self.token = load_token(token_file) or ""
        self.proxy_url = proxy_url
        self.messages_file = messages_file
        self.mtime_file = mtime_file
        self.ws = None
        self.connected = threading.Event()
        self.authenticated = threading.Event()
        self.reply_queue = []
        self.reply_lock = threading.Lock()
        self.backoff = 1.0
        self.max_backoff = 60.0
        self._disconnect_event = threading.Event()  # 看门狗触发的主动断开信号

    # ── 看门狗回调 ──
    def on_token_changed(self):
        new_token = load_token(self.token_file)
        if not new_token or new_token == self.token:
            return
        old_preview = self.token[:8] if self.token else "none"
        self.token = new_token
        print(f"[plumber] 🔄 Token changed ({old_preview}... → {new_token[:8]}...), reconnecting...")
        self._disconnect_event.set()
        if self.ws:
            try: self.ws.close()
            except Exception: pass

    # ── 连接循环 ──
    def connect(self, stop_event: threading.Event):
        self._disconnect_event.clear()
        while not stop_event.is_set():
            self.auth_failed = False
            try:
                print(f"[plumber] WebSocket connecting to {self.proxy_url} ...")
                self.ws = websocket.WebSocketApp(
                    self.proxy_url,
                    on_open=self._on_open,
                    on_message=self._on_message,
                    on_error=self._on_error,
                    on_close=self._on_close,
                )
                wst = threading.Thread(target=self.ws.run_forever, kwargs={
                    "ping_interval": 30, "ping_timeout": 10
                }, daemon=True)
                wst.start()

                done = False
                while not done and not stop_event.is_set() and not self._disconnect_event.is_set():
                    if self.authenticated.is_set():
                        self.backoff = 1.0
                        while self.connected.is_set() and not stop_event.is_set() and not self._disconnect_event.is_set():
                            time.sleep(1)
                        done = True
                    elif self.auth_failed:
                        done = True
                    else:
                        time.sleep(0.5)

                if not self.connected.is_set() and not self.auth_failed:
                    if self.ws:
                        self.ws.close()
            except Exception as e:
                print(f"[plumber] WebSocket connection error: {e}", file=sys.stderr)

            if self._disconnect_event.is_set():
                self._disconnect_event.clear()
                self.authenticated.clear()
                self.connected.clear()
                continue

            if self.auth_failed:
                print("[plumber] Auth failed permanently, shutting down.", file=sys.stderr)
                stop_event.set()
                break

            if not stop_event.is_set():
                print(f"[plumber] Reconnecting in {self.backoff}s...")
                time.sleep(self.backoff)
                self.backoff = min(self.backoff * 2, self.max_backoff)

    def _on_open(self, ws):
        print("[plumber] WebSocket connected, sending auth...")
        ws.send(json.dumps({"type": "agent_auth", "token": self.token}))

    def _on_message(self, ws, message):
        try:
            msg = json.loads(message)
            msg_type = msg.get("type", "")

            if msg_type == "auth_result":
                if msg.get("success"):
                    self.authenticated.set()
                    self.connected.set()
                    print("[plumber] Authenticated successfully")
                    self._flush_replies(ws)
                else:
                    err = msg.get("error", "Unknown")
                    print(f"[plumber] Auth failed: {err}", file=sys.stderr)
                    with open(self.messages_file, "a") as f:
                        f.write(json.dumps({"type": "error", "code": 401,
                              "message": f"Agent WS auth failed: {err}", "timestamp": time.time()}) + "
")
                    self.mtime_file.write_text(str(time.time()))
                    self.auth_failed = True
                    ws.close()
                return

            with open(self.messages_file, "a") as f:
                f.write(json.dumps(msg) + "
")
            self.mtime_file.write_text(str(time.time()))
            method = msg.get("method", "")
            task = msg.get("params", {}).get("task", "")
            print(f"[plumber] Recv: {method} {task[:60] if task else ''}")

        except json.JSONDecodeError:
            pass
        except Exception as e:
            print(f"[plumber] Message handling error: {e}", file=sys.stderr)

    def _on_error(self, ws, error):
        s = str(error)
        if "401" in s:
            print("[plumber] ERROR: Auth rejected (token expired/invalid)", file=sys.stderr)
        else:
            print(f"[plumber] WebSocket error: {error}", file=sys.stderr)

    def _on_close(self, ws, code, msg):
        self.connected.clear()
        self.authenticated.clear()
        print(f"[plumber] WebSocket closed (code={code})")

    def send_reply(self, reply_data):
        with self.reply_lock:
            if self.connected.is_set() and self.authenticated.is_set() and self.ws:
                try:
                    self.ws.send(json.dumps(reply_data))
                    return True
                except Exception as e:
                    print(f"[plumber] Send reply error: {e}", file=sys.stderr)
            self.reply_queue.append(reply_data)
            return False

    def _flush_replies(self, ws):
        with self.reply_lock:
            while self.reply_queue:
                reply = self.reply_queue.pop(0)
                try:
                    ws.send(json.dumps(reply))
                except Exception as e:
                    print(f"[plumber] Flush reply error: {e}", file=sys.stderr)
                    self.reply_queue.insert(0, reply)
                    break

# ─── 回复文件处理 ──────────────────────────────────────────

_reply_lock = threading.Lock()

def _flush_replies_file(plumber: PlumberWsClient, replies_file: Path):
    with _reply_lock:
        try:
            if not replies_file.exists():
                return
            content = replies_file.read_text().strip()
            if not content:
                return
            lines = content.split("
")
            unsent, sent = [], 0
            for line in lines:
                line = line.strip()
                if not line: continue
                try:
                    if plumber.send_reply(json.loads(line)):
                        sent += 1
                    else:
                        unsent.append(line)
                except json.JSONDecodeError:
                    unsent.append(line)
                except Exception as e:
                    print(f"[plumber] Reply sender error: {e}", file=sys.stderr)
                    unsent.append(line)
            if sent > 0:
                print(f"[plumber] Sent {sent} replies")
            replies_file.write_text("
".join(unsent) + "
" if unsent else "")
        except Exception as e:
            print(f"[plumber] Reply sender error: {e}", file=sys.stderr)

# ─── 统一 Watchdog（单一 Observer，多个 handler）──────────

class UnifiedFileHandler(FileSystemEventHandler):
    """单一 handler：根据触发文件的名称分发到对应回调"""

    def __init__(self, token_file: Path, replies_file: Path,
                 on_token_changed, on_reply_changed):
        self._token_name = token_file.name
        self._replies_name = replies_file.name
        self._on_token_changed = on_token_changed
        self._on_reply_changed = on_reply_changed

    def on_modified(self, event):
        if event.is_directory: return
        name = Path(event.src_path).name
        if name == self._token_name:
            self._on_token_changed()
        elif name == self._replies_name:
            self._on_reply_changed()

    def on_created(self, event):
        if event.is_directory: return
        name = Path(event.src_path).name
        if name == self._replies_name:
            self._on_reply_changed()


def start_unified_watchdog(data_dir: Path, token_file: Path, replies_file: Path,
                           plumber: PlumberWsClient, stop_event: threading.Event):
    """启动单一 Observer 同时监听 token 和 replies 变更"""
    data_dir.mkdir(parents=True, exist_ok=True)

    def on_token():
        plumber.on_token_changed()

    def on_reply():
        _flush_replies_file(plumber, replies_file)

    handler = UnifiedFileHandler(token_file, replies_file, on_token, on_reply)
    observer = Observer()
    observer.schedule(handler, str(data_dir), recursive=False)
    observer.start()
    print(f"[plumber] Watchdog monitoring {data_dir} (token + replies)")

    # 启动时处理已有 replies
    if replies_file.exists():
        _flush_replies_file(plumber, replies_file)

    try:
        while observer.is_alive() and not stop_event.is_set():
            time.sleep(1)
    finally:
        observer.stop()
        observer.join(timeout=2)

# ─── Fallback 轮询 ──────────────────────────────────────

def poll_loop(data_dir: Path, token_file: Path, replies_file: Path,
              plumber: PlumberWsClient, stop_event: threading.Event):
    """fallback: 3s 轮询 token + 1s 轮询 replies"""
    print("[plumber] ⚠ watchdog not installed, using polling fallback. pip install watchdog")
    token_mtime = token_file.stat().st_mtime if token_file.exists() else 0
    while not stop_event.is_set():
        # token 检查（每 3 秒）
        try:
            if token_file.exists():
                mt = token_file.stat().st_mtime
                if mt > token_mtime:
                    token_mtime = mt
                    plumber.on_token_changed()
        except Exception: pass
        # replies 检查（每 1 秒）
        _flush_replies_file(plumber, replies_file)
        time.sleep(1)

# ─── 主入口 ──────────────────────────────────────────────

def main():
    config = get_config()
    token = load_token(config["token_file"])
    if not token:
        print(f"[plumber] ERROR: Token file not found: {config['token_file']}", file=sys.stderr)
        sys.exit(1)

    config["messages_file"].parent.mkdir(parents=True, exist_ok=True)
    config["pid_file"].write_text(str(os.getpid()))
    print(f"[plumber] PID {os.getpid()} → {config['pid_file']}")

    stop = threading.Event()
    def shutdown(sig, frame):
        print("[plumber] Shutting down...")
        stop.set()
    signal.signal(signal.SIGTERM, shutdown)
    signal.signal(signal.SIGINT, shutdown)

    plumber = PlumberWsClient(config["token_file"], config["proxy_url"],
                              config["messages_file"], config["mtime_file"])

    data_dir = config["token_file"].parent

    t1 = threading.Thread(target=plumber.connect, args=(stop,), daemon=True)
    t1.start()

    if _WATCHDOG_AVAILABLE:
        t2 = threading.Thread(target=start_unified_watchdog,
            args=(data_dir, config["token_file"], config["replies_file"], plumber, stop), daemon=True)
    else:
        t2 = threading.Thread(target=poll_loop,
            args=(data_dir, config["token_file"], config["replies_file"], plumber, stop), daemon=True)
    t2.start()

    print(f"[plumber] Started. Proxy: {config['proxy_url']}")
    print(f"[plumber] Token:    {token[:8]}...")

    try:
        while not stop.is_set():
            time.sleep(1)
    except KeyboardInterrupt:
        pass
    finally:
        stop.set()
        if plumber.ws:
            try: plumber.ws.close()
            except Exception: pass
        config["pid_file"].unlink(missing_ok=True)
        print("[plumber] Stopped.")


if __name__ == "__main__":
    main()

task_watcher.py — Agent 调度层（仅 Hermes）

启动：python3 ~/.chrome-ai-helper/task_watcher.py &
前置条件：hermes profile use chrome-ai-helper
PID：/tmp/mcp_task_watcher.pid
职责：调度层。watchdog 监听信号文件（fallback: 3s 轮询）→ 检测到 user_task → 构建纯协议层 prompt → 启动 hermes chat -s mcp-protocol-testing -m Lo-Mo --max-turns 30 --yolo -q 子进程。不直接执行浏览器操作，不包含业务逻辑。用户任务的详细步骤和验证条件由用户从侧边栏提供。

#!/usr/bin/env python3
"""
Chrome AI Helper — 自动任务执行器（调度层）
保存为 ~/.chrome-ai-helper/task_watcher.py

职责:
  - watchdog 监听 /tmp/mcp_pending_mtime.txt 信号文件（文件修改时触发）
  - 检测到新 user_task 后启动 hermes chat -q 子进程
  - 发送 task_accepted/agent_action/task_completed 通知到 replies.jsonl
  - 不直接调用 MCP 工具（由 Hermes Agent 智能决策并执行）
  - watchdog 不可用时自动 fallback 到 3s 轮询

启动: python3 ~/.chrome-ai-helper/task_watcher.py &
PID: /tmp/mcp_task_watcher.pid
"""

import json, time, os, sys, signal, subprocess, threading
from pathlib import Path

SIGNAL_FILE = Path("/tmp/mcp_pending_mtime.txt")
MESSAGES_FILE = Path.home() / ".chrome-ai-helper" / "messages.jsonl"
REPLIES_FILE = Path.home() / ".chrome-ai-helper" / "replies.jsonl"
PID_FILE = Path("/tmp/mcp_task_watcher.pid")

_WATCHDOG_AVAILABLE = False
try:
    from watchdog.observers import Observer
    from watchdog.events import FileSystemEventHandler
    _WATCHDOG_AVAILABLE = True
except ImportError:
    pass


def write_reply(reply):
    """写入 replies.jsonl，plumber 自动通过 WS 发送"""
    with open(REPLIES_FILE, 'a') as f:
        f.write(json.dumps(reply, ensure_ascii=False) + '
')


def handle_task(task_text, tab_url=""):
    """读取任务，启动 Hermes Agent 智能处理。"""
    print(f"
[watcher] === Task: {task_text[:80]} ===")

    # 1. 通知侧边栏
    write_reply({"jsonrpc":"2.0","method":"notifications/task_accepted","params":{}})
    write_reply({"jsonrpc":"2.0","method":"notifications/agent_message",
        "params":{"type":"text","content":f"🤖 AI 正在分析任务：{task_text[:80]}"}})
    write_reply({"jsonrpc":"2.0","method":"notifications/agent_action",
        "params":{"action":"delegate","description":"AI 理解任务意图中...","status":"started"}})

    # 2. 构建 Agent prompt（纯协议层，不包含业务逻辑）
    prompt = (
        "IMPORTANT: You have MCP tools registered from server 'chrome-ai-helper'. "
        "Call them using EXACTLY these tool names — do NOT add any prefix like mcp__ or mcp_chrome_ai_helper_:

"
        "  browser_start_session, browser_navigate, browser_get_dom,
"
        "  browser_click, browser_type, browser_screenshot, browser_execute_js,
"
        "  browser_wait, browser_scroll, pause_for_user, request_human_input

"
        "Start NOW with tools/call.

"
        f"TASK: {task_text}

"
        "WORKFLOW:
"
        "1. tools/call browser_start_session
"
        "2. tools/call browser_navigate to the target URL
"
        "3. tools/call browser_get_dom to see interactive elements
"
        "4. Use browser_click before browser_type to focus inputs
"
        "5. browser_execute_js only as last resort (max 5 calls)
"
        "6. browser_screenshot only when DOM can't answer (max 3 calls)
"
        "7. Follow user instructions exactly. Only announce completion when truly done."
    )

    # 3. 启动 Hermes Agent 子进程
    try:
        print("[watcher] Launching Hermes Agent via MCP tools...")
        result = subprocess.run(
            ["hermes", "chat", "-s", "mcp-protocol-testing", "-m", "Lo-Mo",
             "--max-turns", "30", "--yolo", "-q", prompt],
            capture_output=True, text=True, timeout=600
        )
        if result.returncode == 0:
            print("[watcher] Agent completed")
            write_reply({"jsonrpc":"2.0","method":"notifications/agent_action",
                "params":{"action":"delegate","description":"AI 任务处理完成","status":"completed"}})
        else:
            err = result.stderr[:500] if result.stderr else f"exit {result.returncode}"
            print(f"[watcher] Agent failed: {err}")
            write_reply({"jsonrpc":"2.0","method":"notifications/agent_action",
                "params":{"action":"delegate","description":"AI 处理出错",
                "status":"failed","error":err}})
            write_reply({"jsonrpc":"2.0","method":"notifications/agent_message",
                "params":{"type":"text","content":f"❌ AI 处理出错：{err}"}})
    except subprocess.TimeoutExpired:
        print("[watcher] Agent timed out")
        write_reply({"jsonrpc":"2.0","method":"notifications/agent_message",
            "params":{"type":"text","content":"⚠️ AI 处理超时（10分钟）。任务可能过于复杂，请尝试拆分任务或手动完成。"}})
        write_reply({"jsonrpc":"2.0","method":"notifications/agent_action",
            "params":{"action":"delegate","description":"AI 处理超时（10分钟），建议拆分任务","status":"timeout"}})
    except FileNotFoundError:
        print("[watcher] hermes not found")
        write_reply({"jsonrpc":"2.0","method":"notifications/agent_message",
            "params":{"type":"text","content":"❌ 未找到 hermes 命令，请确认 Hermes Agent 已安装"}})
    except Exception as e:
        print(f"[watcher] Error: {e}")
        write_reply({"jsonrpc":"2.0","method":"notifications/agent_message",
            "params":{"type":"text","content":f"❌ AI 处理异常：{str(e)[:200]}"}})

    write_reply({"jsonrpc":"2.0","method":"notifications/task_completed","params":{}})

# ─── Watchdog 信号监听 ──────────────────────────────────

_last_offset = 0


def _process_pending_tasks():
    """读取 messages.jsonl 中未处理的行，处理其中的 user_task"""
    global _last_offset
    if not MESSAGES_FILE.exists():
        return
    try:
        lines = MESSAGES_FILE.read_text().strip().split('
')
        current = [l for l in lines if l.strip()]
        for i in range(_last_offset, len(current)):
            try:
                msg = json.loads(current[i])
                if msg.get("method") == "notifications/user_task" and msg.get("params", {}).get("task"):
                    handle_task(msg["params"]["task"], msg["params"].get("tab_url", ""))
                    _last_offset = i + 1
            except json.JSONDecodeError:
                continue
    except Exception as e:
        print(f"[watcher] Error reading messages: {e}", file=sys.stderr)


def _on_signal_changed():
    """watchdog 回调：信号文件被修改"""
    print("[watcher] Signal detected!")
    _process_pending_tasks()


class SignalFileHandler(FileSystemEventHandler):
    def __init__(self, callback):
        self._callback = callback

    def on_modified(self, event):
        if not event.is_directory:
            self._callback()

    def on_created(self, event):
        if not event.is_directory:
            self._callback()


def watch_with_watchdog():
    """watchdog 模式：监听信号文件所在目录的事件"""
    global _last_offset
    _last_offset = 0  # 启动时从头处理所有未处理消息

    # 首次启动时可能已有信号文件，立即处理
    if SIGNAL_FILE.exists():
        _process_pending_tasks()

    signal_dir = SIGNAL_FILE.parent.resolve()  # macOS: /tmp → /private/tmp
    observer = Observer()
    handler = SignalFileHandler(_on_signal_changed)
    observer.schedule(handler, str(signal_dir), recursive=False)
    observer.start()

    print(f"[watcher] Started. PID={os.getpid()} (watchdog monitoring {signal_dir})")

    try:
        while observer.is_alive():
            time.sleep(1)
    except KeyboardInterrupt:
        pass
    finally:
        observer.stop()
        observer.join(timeout=2)
        print("[watcher] Stopped.")


def watch_with_polling():
    """Fallback 轮询模式"""
    print(f"[watcher] Started. PID={os.getpid()} (polling every 3s)")
    print(f"[watcher] ⚠ watchdog not available, using polling fallback. pip install watchdog")

    global _last_offset
    _last_offset = 0  # 启动时从头处理所有未处理消息

    while True:
        try:
            if SIGNAL_FILE.exists():
                _process_pending_tasks()
                SIGNAL_FILE.unlink(missing_ok=True)
                print("[watcher] Signal cleared")
            time.sleep(3)
        except KeyboardInterrupt:
            break
        except Exception as e:
            print(f"[watcher] Error: {e}")
            time.sleep(3)


# ─── 主入口 ──────────────────────────────────────────────

if __name__ == "__main__":
    PID_FILE.write_text(str(os.getpid()))

    def shutdown(sig, frame):
        print("[watcher] Shutting down...")
        PID_FILE.unlink(missing_ok=True)
        sys.exit(0)

    signal.signal(signal.SIGTERM, shutdown)
    signal.signal(signal.SIGINT, shutdown)

    if _WATCHDOG_AVAILABLE:
        watch_with_watchdog()
    else:
        watch_with_polling()

本地脚本文件结构

~/.chrome-ai-helper/
├── token                         # token 存储
├── sessions//token   # 多会话 token
├── messages.jsonl               # plumber 写入，Agent 读取
├── replies.jsonl                # Agent 写入，plumber 自动发送
├── mcp-plumber.py               # 纯收发管道脚本（源码见上）
└── task_watcher.py              # 调度层（源码见上）

仓库中（版本管理）:
chrome-ai-helper/scripts/
├── mcp-plumber.py               # 等同 ~/.chrome-ai-helper/mcp-plumber.py
└── task_watcher.py              # 等同 ~/.chrome-ai-helper/task_watcher.py

场景 1：Token 过期/认证失败

场景 2：plumber 进程挂了

场景 2b：task_watcher 进程挂了（Hermes）

场景 3：Agent WebSocket 连接断开

场景 4：Extension（侧边栏）断开

#!/bin/bash
TOKEN="your-token-here"

# 1. 建立 Agent WebSocket 连接并认证
# wscat -c "ws://localhost:3000/agent-ws"
# > {"type":"agent_auth","token":"$TOKEN"}
# < {"type":"auth_result","success":true}

# 2. 初始化 + 获取工具列表
# > {"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}
# > {"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}

# 3. 调用浏览器工具
# > {"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"browser_start_session","arguments":{"url":"https://www.baidu.com"}}}
# > {"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"browser_execute_js","arguments":{"expression":"document.title"}}}
# > {"jsonrpc":"2.0","id":5,"method":"tools/call","params":{"name":"browser_click","arguments":{"selector":"#kw"}}}
# > {"jsonrpc":"2.0","id":6,"method":"tools/call","params":{"name":"browser_type","arguments":{"selector":"#kw","text":"搜索内容"}}}

# 或者 HTTP 备用:
# curl -s -X POST http://localhost:3000/mcp #   -H "Authorization: Bearer $TOKEN" #   -H "Content-Type: application/json" #   -d '{"jsonrpc":"2.0","id":1,"method":"tools/call",...}'