#!/usr/bin/env python3 """ scripts/cli.py - MyAgent CLI 工具集 ==================================== 轻量级命令行工具，LLM 通过 command 工具调用这些命令完成所有操作。 用法: myagent-ai [args...] 命令分类: 感知: ocr, analyze-image, transcribe 搜索: search, read-url, fetch-url 文件: send-file 文档: docx-create, docx-read, xlsx-create, xlsx-read, xlsx-edit, ppt-create, ppt-read, pdf-create, pdf-read 系统: sysinfo, ps, env, pathinfo 浏览器: browser-open, browser-screenshot, browser-close, browser-click, browser-fill, browser-eval, browser-navigate GUI: screenshot, mouse-click, mouse-drag, type-text, hotkey, window-list, window-focus 记忆: memory 通信: chat 帮助: help, -h, --help """ from __future__ import annotations import asyncio import json import os import sys import time from pathlib import Path # 确保可以导入 myagent 模块 _SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) _PROJECT_DIR = os.path.dirname(_SCRIPT_DIR) if _PROJECT_DIR not in sys.path: sys.path.insert(0, _PROJECT_DIR) # ============================================================================= # 工具函数 # ============================================================================= def _print_result(result: dict): """打印 SkillResult / 命令结果""" if result.get("success"): if result.get("message"): print(result["message"]) data = result.get("data") if data: print(json.dumps(data, ensure_ascii=False, indent=2, default=str)) else: print(f"错误: {result.get('error', '未知错误')}", file=sys.stderr) sys.exit(1) def _read_stdin() -> str: """从 stdin 读取内容""" if not sys.stdin.isatty(): return sys.stdin.read() return "" def _parse_json_arg(arg: str) -> dict: """解析 JSON 参数""" try: return json.loads(arg) except json.JSONDecodeError as e: print(f"JSON 解析失败: {e}", file=sys.stderr) sys.exit(1) async def _run(): """CLI 主入口""" args = sys.argv[1:] if not args: args = ["help"] cmd = args[0] cmd_args = args[1:] # [v1.23.75] LLM 有时把参数放在子命令前面（如 --agent 3 chat --message "hi"） # 此时 cmd 会是 "--agent" 而不是 "chat"，导致命令路由失败。 # 防御：当 cmd 以 - 开头时，找到第一个非 flag 参数作为子命令。 if cmd and cmd.startswith("-"): for i, a in enumerate(args): if not a.startswith("-"): cmd = a cmd_args = args[:i] + args[i + 1:] # flags 归入 cmd_args 传给子命令 break # ── 命令路由 ── handlers = { # 感知 "ocr": cmd_ocr, "analyze-image": cmd_analyze_image, "transcribe": cmd_transcribe, # 搜索 "search": cmd_search, "read-url": cmd_read_url, "fetch-url": cmd_fetch_url, # 文件操作 "send-file": cmd_send_file, # 文档 "docx-create": cmd_docx_create, "docx-read": cmd_docx_read, "xlsx-create": cmd_xlsx_create, "xlsx-read": cmd_xlsx_read, "xlsx-edit": cmd_xlsx_edit, "ppt-create": cmd_ppt_create, "ppt-read": cmd_ppt_read, "pdf-create": cmd_pdf_create, "pdf-read": cmd_pdf_read, # 系统 "pathinfo": cmd_pathinfo, # 浏览器 (ChromeDev MCP) "browser-open": cmd_browser_open, "browser-screenshot": cmd_browser_screenshot, "browser-close": cmd_browser_close, "browser-click": cmd_browser_click, "browser-fill": cmd_browser_fill, "browser-eval": cmd_browser_eval, "browser-navigate": cmd_browser_navigate, # 反检测浏览器 (DrissionPage) "stealth-open": cmd_stealth_open, "stealth-navigate": cmd_stealth_navigate, "stealth-click": cmd_stealth_click, "stealth-fill": cmd_stealth_fill, "stealth-screenshot": cmd_stealth_screenshot, "stealth-content": cmd_stealth_content, "stealth-cookies": cmd_stealth_cookies, "stealth-wait": cmd_stealth_wait, "stealth-wait-manual": cmd_stealth_wait_manual, "stealth-close": cmd_stealth_close, "stealth-eval": cmd_stealth_eval, # Profile 管理 "profile-list": cmd_profile_list, "profile-create": cmd_profile_create, "profile-delete": cmd_profile_delete, # Site 管理 (v1.31.0) "site-manage": cmd_site_manage, # GUI "screenshot": cmd_screenshot, "mouse-click": cmd_mouse_click, "mouse-drag": cmd_mouse_drag, "type-text": cmd_type_text, "hotkey": cmd_hotkey, "window-list": cmd_window_list, "window-focus": cmd_window_focus, "screen-element": cmd_screen_element, # 记忆 "memory": cmd_memory, # 媒体播放 "playaudio": cmd_playaudio, "playvideo": cmd_playvideo, # 通信 "chat": cmd_chat, # 压缩解压 "zip": cmd_zip, "unzip": cmd_unzip, # 帮助 "help": cmd_help, "-h": cmd_help, "--help": cmd_help, } handler = handlers.get(cmd) if handler: await handler(cmd_args) else: print(f"未知命令: {cmd}", file=sys.stderr) print(f"运行 'myagent-ai help' 查看所有可用命令", file=sys.stderr) sys.exit(1) def _filter_unknown_args(args, allowed_prefixes=None): """[v1.23.46] 通用参数过滤器 — 去掉不认识的系统命令参数。 LLM 经常把系统命令的参数（如 ls -la, grep -n）传给 myagent-ai 子命令。 此函数只保留: 位置参数（不以 - 开头） + allowed_prefixes 匹配的选项。 Args: args: 原始参数列表 allowed_prefixes: 允许的选项前缀集合（如 {"-p", "--pattern", "-r", "--max"}） Returns: 过滤后的参数列表 """ if allowed_prefixes is None: allowed_prefixes = set() # 常见系统命令参数（标志，不带值） _sys_flags = { "-l", "-a", "-h", "-L", "-d", "-F", "-f", "-i", "-s", "-R", "-S", "-t", "-r", "-u", "-c", "-1", "-m", "-x", "-o", "-g", "-G", "-A", "-n", "-w", "-p", "-q", "-v", "-e", "-E", "-z", "-b", "-B", "-T", "-H", "-I", "-k", "-N", "-W", # GNU long options "--color", "--group-directories-first", "--hide", "--indicator-style", "--quoting-style", "--show-control-chars", "--sort", "--time", "--time-style", "--format", "--human-readable", "-lh", "-la", "-al", "-lr", "-lt", "-ls", "-ld", "-li", "-shal", "-lah", } # 带值的系统参数 _sys_opts = { "--sort", "--time", "--time-style", "--format", "--hide", "--indicator-style", "--quoting-style", "--context", "--width", "--tabs", "--block-size", } filtered = [] skip_next = False for arg in args: if skip_next: skip_next = False continue # 以 - 开头的是选项参数 if arg.startswith("-"): # 检查是否在允许列表中 matched = any(arg.startswith(prefix) for prefix in allowed_prefixes) if matched: filtered.append(arg) # 如果参数不含 =，且不是标志模式，可能需要跳过下一个 if "=" not in arg and arg not in allowed_prefixes: # 检查是否有对应的 long option 带 = pass continue # 不在允许列表中，检查是否是系统参数 stripped = arg.lstrip("-").rstrip("=") if arg in _sys_flags or stripped in _sys_flags: continue # 标志，跳过 if arg in _sys_opts or stripped in _sys_opts: skip_next = True # 带值选项，跳过下一个 continue # 未知选项，跳过 continue else: # 位置参数，保留 filtered.append(arg) return filtered if filtered else args # ============================================================================= # 感知命令 # ============================================================================= async def cmd_ocr(args): """OCR 文字识别""" import argparse p = argparse.ArgumentParser(prog="myagent-ai ocr", description="OCR 文字识别 — 从图片中提取文字") p.add_argument("image", help="图片路径") p.add_argument("lang", nargs="?", default="ch", help="语言: ch=中英混合(默认), en=英文") a = p.parse_args(args) from core.tool_dispatcher import ToolDispatcher d = ToolDispatcher() result = await d._exec_image_ocr({"image_path": a.image, "lang": a.lang}, "") _print_result(result) async def cmd_analyze_image(args): """图片分析 (VLM)""" import argparse p = argparse.ArgumentParser(prog="myagent-ai analyze-image", description="图片内容分析 — 使用 VLM 分析图片") p.add_argument("image", help="图片路径") p.add_argument("prompt", nargs="?", default="请详细描述这张图片的内容。", help="分析提示词") a = p.parse_args(args) from core.tool_dispatcher import ToolDispatcher d = ToolDispatcher() result = await d._exec_image_analyze({"image_path": a.image, "prompt": a.prompt}, "") _print_result(result) async def cmd_transcribe(args): """语音转文字""" import argparse p = argparse.ArgumentParser(prog="myagent-ai transcribe", description="语音转文字 — 将音频转录为文本") p.add_argument("audio", help="音频文件路径") p.add_argument("language", nargs="?", default="zh", help="语言代码: zh(默认), en, ja") a = p.parse_args(args) from core.tool_dispatcher import ToolDispatcher d = ToolDispatcher() result = await d._exec_audio_transcribe({"audio_path": a.audio, "language": a.language}, "") _print_result(result) # ============================================================================= # 搜索命令 # ============================================================================= async def cmd_search(args): """网络搜索""" import argparse p = argparse.ArgumentParser(prog="myagent-ai search", description="网络搜索 — 搜索互联网信息") p.add_argument("query", nargs="+", help="搜索关键词") p.add_argument("-n", "--num", type=int, default=10, help="返回结果数量 (默认10, 最大20)") a = p.parse_args(args) from aiskills.search_skill import WebSearchSkill skill = WebSearchSkill() result = await skill.execute(query=" ".join(a.query), num=a.num) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_read_url(args): """读取网页正文""" import argparse p = argparse.ArgumentParser(prog="myagent-ai read-url", description="读取网页正文内容") p.add_argument("url", help="网页 URL") p.add_argument("--raw", action="store_true", help="返回原始 HTML") a = p.parse_args(args) from aiskills.search_skill import WebReadSkill skill = WebReadSkill() result = await skill.execute(url=a.url, extract_text=not a.raw) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_fetch_url(args): """获取 URL 原始内容 (API 调用)""" import argparse p = argparse.ArgumentParser(prog="myagent-ai fetch-url", description="获取 URL 原始内容 (API 调用)") p.add_argument("url", help="URL 地址") p.add_argument("-m", "--method", default="GET", choices=["GET", "POST", "PUT", "DELETE"]) p.add_argument("-H", "--header", action="append", default=[], help="请求头 (格式: Key:Value)") p.add_argument("-d", "--data", default="", help="请求体 (POST/PUT)") a = p.parse_args(args) headers = {} for h in a.header: if ":" in h: k, v = h.split(":", 1) headers[k.strip()] = v.strip() from aiskills.search_skill import URLReadSkill skill = URLReadSkill() result = await skill.execute(url=a.url, method=a.method, headers=headers, body=a.data) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) # ============================================================================= # 文件操作命令 # ============================================================================= async def cmd_send_file(args): """发送文件给用户""" import argparse p = argparse.ArgumentParser(prog="myagent-ai send-file", description="发送文件给用户") p.add_argument("path", help="文件路径") p.add_argument("description", nargs="?", default="", help="文件描述") a = p.parse_args(args) from aiskills.file_send import FileSendSkill skill = FileSendSkill() result = await skill.execute(file_path=a.path, description=a.description) if result.get("success"): print(f"文件已发送: {result.get('name', '')} ({result.get('size', 0)} 字节)") if result.get("url"): print(f"URL: {result['url']}") else: print(f"错误: {result.get('error', '发送失败')}", file=sys.stderr) sys.exit(1) # ============================================================================= # 文档命令 # ============================================================================= async def cmd_docx_create(args): """创建 Word 文档""" import argparse p = argparse.ArgumentParser(prog="myagent-ai docx-create", description="创建 Word 文档") p.add_argument("-c", "--content", required=True, help="内容 JSON (结构见 DocxSkill)") p.add_argument("-t", "--title", default="文档", help="文档标题") p.add_argument("-o", "--output", "--output_path", default="", help="输出路径 (默认自动生成)") a = p.parse_args(args) content = _parse_json_arg(a.content) from aiskills.docx_skill import DOCXCreateSkill skill = DOCXCreateSkill() result = await skill.execute(content=content, title=a.title, output_path=a.output) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) if result.success and result.files: print(f"\n__SEND_FILE__{result.files[0]}|{result.message or 'Word 文档'}__END__") async def cmd_docx_read(args): """读取 Word 文档""" import argparse p = argparse.ArgumentParser(prog="myagent-ai docx-read", description="读取 Word 文档内容") p.add_argument("path", help="Word 文件路径 (.docx)") a = p.parse_args(args) from aiskills.docx_skill import DOCXReadSkill skill = DOCXReadSkill() result = await skill.execute(path=a.path) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_xlsx_create(args): """创建 Excel 文件""" import argparse p = argparse.ArgumentParser(prog="myagent-ai xlsx-create", description="创建 Excel 文件") p.add_argument("-s", "--sheets", required=True, help="工作表数据 JSON") p.add_argument("-t", "--title", default="表格", help="文件标题") p.add_argument("-o", "--output", "--output_path", default="", help="输出路径 (默认自动生成)") a = p.parse_args(args) sheets = _parse_json_arg(a.sheets) from aiskills.xlsx_skill import XLSXCreateSkill skill = XLSXCreateSkill() result = await skill.execute(sheets=sheets, title=a.title, output_path=a.output) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) if result.success and result.files: print(f"\n__SEND_FILE__{result.files[0]}|{result.message or 'Excel 文档'}__END__") async def cmd_xlsx_read(args): """读取 Excel 文件""" import argparse p = argparse.ArgumentParser(prog="myagent-ai xlsx-read", description="读取 Excel 文件内容") p.add_argument("path", help="Excel 文件路径 (.xlsx)") p.add_argument("--sheet", default="", help="工作表名称 (默认全部)") a = p.parse_args(args) from aiskills.xlsx_skill import XLSXReadSkill skill = XLSXReadSkill() result = await skill.execute(path=a.path, sheet_name=a.sheet) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_xlsx_edit(args): """编辑 Excel 文件""" import argparse p = argparse.ArgumentParser(prog="myagent-ai xlsx-edit", description="编辑 Excel 文件") p.add_argument("path", help="Excel 文件路径") p.add_argument("action", choices=["append-row", "add-column", "update-cell"], help="操作类型") p.add_argument("-d", "--data", required=True, help="操作数据 JSON") p.add_argument("--sheet", default="", help="工作表名称 (默认活动表)") a = p.parse_args(args) data = _parse_json_arg(a.data) from aiskills.xlsx_skill import XLSXEditSkill skill = XLSXEditSkill() result = await skill.execute(path=a.path, action=a.action, data=data, sheet_name=a.sheet) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_ppt_create(args): """创建 PowerPoint 文件""" import argparse p = argparse.ArgumentParser(prog="myagent-ai ppt-create", description="创建 PowerPoint 演示文稿") p.add_argument("-s", "--slides", required=True, help="幻灯片数据 JSON") p.add_argument("--theme", default="", help="主题名称") p.add_argument("-o", "--output", "--output_path", default="", help="输出路径 (默认自动生成)") a = p.parse_args(args) slides = _parse_json_arg(a.slides) from aiskills.ppt_skill import PPTCreateSkill skill = PPTCreateSkill() result = await skill.execute(slides=slides, theme=a.theme, output_path=a.output) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) if result.success and result.files: print(f"\n__SEND_FILE__{result.files[0]}|{result.message or 'PPT 演示文稿'}__END__") async def cmd_ppt_read(args): """读取 PowerPoint 文件""" import argparse p = argparse.ArgumentParser(prog="myagent-ai ppt-read", description="读取 PowerPoint 演示文稿内容") p.add_argument("path", help="PPT 文件路径 (.pptx)") a = p.parse_args(args) from aiskills.ppt_skill import PPTReadSkill skill = PPTReadSkill() result = await skill.execute(path=a.path) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_pdf_create(args): """创建 PDF 文件""" import argparse p = argparse.ArgumentParser(prog="myagent-ai pdf-create", description="创建 PDF 文件") p.add_argument("-c", "--content", required=True, help="内容 JSON") p.add_argument("--palette", default="", help="配色方案") p.add_argument("-o", "--output", "--output_path", default="", help="输出路径 (默认自动生成)") a = p.parse_args(args) content = _parse_json_arg(a.content) from aiskills.pdf_skill import PDFCreateSkill skill = PDFCreateSkill() result = await skill.execute(content=content, palette=a.palette, output_path=a.output) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) if result.success and result.files: print(f"\n__SEND_FILE__{result.files[0]}|{result.message or 'PDF 文档'}__END__") async def cmd_pdf_read(args): """读取 PDF 文件""" import argparse p = argparse.ArgumentParser(prog="myagent-ai pdf-read", description="读取 PDF 文件内容") p.add_argument("path", help="PDF 文件路径") p.add_argument("--start", type=int, default=1, help="起始页 (默认1)") p.add_argument("--end", type=int, default=0, help="结束页 (默认全部)") a = p.parse_args(args) from aiskills.pdf_skill import PDFReadSkill skill = PDFReadSkill() result = await skill.execute(path=a.path, start_page=a.start, end_page=a.end) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) # ============================================================================= # 系统命令 # ============================================================================= async def cmd_pathinfo(args): """路径信息""" import argparse p = argparse.ArgumentParser(prog="myagent-ai pathinfo", description="获取路径详细信息") p.add_argument("path", help="文件/目录路径") a = p.parse_args(args) from aiskills.system_skill import PathExpandSkill skill = PathExpandSkill() result = await skill.execute(path=a.path) _print_result({"success": result.success, "data": result.data, "error": result.error}) # ============================================================================= # 浏览器命令 (chrome-devtools-mcp) # ============================================================================= async def _ensure_mcp_client(): """确保 MCP 客户端可用""" from aiskills.chromedev_mcp import get_mcp_client, _ensure_node_deps err = _ensure_node_deps() if err: print(f"错误: {err}", file=sys.stderr) sys.exit(1) client = await get_mcp_client() if not client: print("错误: 浏览器 MCP 客户端启动失败", file=sys.stderr) print("提示: 运行 'myagent-ai browser-open --diagnose' 查看诊断信息", file=sys.stderr) sys.exit(1) return client async def cmd_browser_open(args): """打开浏览器页面""" import argparse p = argparse.ArgumentParser(prog="myagent-ai browser-open", description="打开浏览器页面") p.add_argument("url", help="要打开的 URL") p.add_argument("--no-headless", action="store_true", help="使用有头模式 (可见浏览器)") p.add_argument("--diagnose", action="store_true", help="诊断浏览器环境") a = p.parse_args(args) if a.diagnose: from aiskills.chromedev_mcp import MCPClient diag = MCPClient.diagnose() print(json.dumps(diag, ensure_ascii=False, indent=2)) return client = await _ensure_mcp_client() result = await client.call_tool("navigate_page", {"url": a.url}, timeout=30) if result.get("error"): print(f"错误: {result['error']}", file=sys.stderr) sys.exit(1) print(f"已打开: {a.url}") if result.get("text"): print(result["text"][:2000]) async def cmd_browser_screenshot(args): """浏览器截图""" import argparse p = argparse.ArgumentParser(prog="myagent-ai browser-screenshot", description="浏览器页面截图") p.add_argument("-o", "--output", default="", help="截图保存路径 (默认自动生成)") a = p.parse_args(args) client = await _ensure_mcp_client() mcp_args = {} if a.output: mcp_args["filePath"] = a.output result = await client.call_tool("take_screenshot", mcp_args, timeout=30) if result.get("error"): print(f"错误: {result['error']}", file=sys.stderr) sys.exit(1) print("截图已保存") if result.get("text"): print(result["text"][:500]) async def cmd_browser_close(args): """关闭浏览器""" client = await _ensure_mcp_client() await client.stop() print("浏览器已关闭") async def cmd_browser_click(args): """浏览器点击元素""" import argparse p = argparse.ArgumentParser(prog="myagent-ai browser-click", description="点击页面元素") p.add_argument("selector", help="CSS 选择器或元素 ref") a = p.parse_args(args) client = await _ensure_mcp_client() result = await client.call_tool("click", {"selector": a.selector}, timeout=15) if result.get("error"): print(f"错误: {result['error']}", file=sys.stderr) sys.exit(1) print("已点击") if result.get("text"): print(result["text"][:1000]) async def cmd_browser_fill(args): """浏览器填写输入框""" import argparse p = argparse.ArgumentParser(prog="myagent-ai browser-fill", description="填写页面输入框") p.add_argument("selector", help="CSS 选择器") p.add_argument("value", help="要填写的值") a = p.parse_args(args) client = await _ensure_mcp_client() result = await client.call_tool("fill", {"selector": a.selector, "value": a.value}, timeout=15) if result.get("error"): print(f"错误: {result['error']}", file=sys.stderr) sys.exit(1) print("已填写") if result.get("text"): print(result["text"][:1000]) async def cmd_browser_eval(args): """浏览器执行 JavaScript""" import argparse p = argparse.ArgumentParser(prog="myagent-ai browser-eval", description="在页面中执行 JavaScript") p.add_argument("script", help="JavaScript 代码") a = p.parse_args(args) client = await _ensure_mcp_client() result = await client.call_tool("evaluate_script", {"script": a.script}, timeout=15) if result.get("error"): print(f"错误: {result['error']}", file=sys.stderr) sys.exit(1) if result.get("text"): print(result["text"][:2000]) async def cmd_browser_navigate(args): """浏览器标签页管理""" import argparse p = argparse.ArgumentParser(prog="myagent-ai browser-navigate", description="浏览器标签页管理") p.add_argument("action", choices=["list", "select", "new", "close", "back", "forward"], help="操作: list=列出标签页, select=切换, new=新建, close=关闭") p.add_argument("arg", nargs="?", default="", help="操作参数 (标签页索引/URL)") a = p.parse_args(args) client = await _ensure_mcp_client() if a.action == "list": result = await client.call_tool("list_pages", {}, timeout=10) elif a.action == "select": result = await client.call_tool("select_page", {"pageNumber": int(a.arg)}, timeout=10) elif a.action == "new": url = a.arg or "about:blank" result = await client.call_tool("new_page", {"url": url}, timeout=15) elif a.action == "close": idx = int(a.arg) if a.arg.isdigit() else 0 result = await client.call_tool("close_page", {"pageNumber": idx}, timeout=10) else: print(f"不支持的操作: {a.action}", file=sys.stderr) sys.exit(1) if result.get("error"): print(f"错误: {result['error']}", file=sys.stderr) sys.exit(1) if result.get("text"): print(result["text"][:2000]) # ============================================================================= # 反检测浏览器命令 (DrissionPage) # ============================================================================= async def _ensure_stealth_browser(profile_name: str = "default", headless: bool = False): """确保反检测浏览器可用""" from aiskills.browser_stealth import get_stealth_browser # Bug Fix: get_stealth_browser 是同步函数，不能用 await browser = get_stealth_browser(profile_name=profile_name, headless=headless) if not browser._started: result = await browser.start() if not result.success: print(f"错误: {result.error}", file=sys.stderr) sys.exit(1) return browser async def cmd_stealth_open(args): """启动反检测浏览器""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-open", description="启动反检测浏览器 (DrissionPage)") p.add_argument("profile", nargs="?", default="default", help="Profile 名称") p.add_argument("--no-headless", action="store_true", help="有头模式") a = p.parse_args(args) from aiskills.browser_stealth import get_stealth_browser # Bug Fix: get_stealth_browser 是同步函数 browser = get_stealth_browser(profile_name=a.profile, headless=not a.no_headless) result = await browser.start() _print_result({"success": result.success, "message": result.message, "error": result.error}) async def cmd_stealth_navigate(args): """反检测浏览器导航""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-navigate", description="反检测浏览器导航") p.add_argument("url", help="目标 URL") p.add_argument("-p", "--profile", default="default", help="Profile 名称") a = p.parse_args(args) browser = await _ensure_stealth_browser(a.profile) result = await browser.navigate(a.url) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_stealth_click(args): """反检测浏览器点击""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-click", description="反检测浏览器点击元素") p.add_argument("selector", help="元素选择器") p.add_argument("-p", "--profile", default="default", help="Profile 名称") a = p.parse_args(args) browser = await _ensure_stealth_browser(a.profile) result = await browser.click(a.selector) _print_result({"success": result.success, "message": result.message, "error": result.error}) async def cmd_stealth_fill(args): """反检测浏览器填写""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-fill", description="反检测浏览器填写输入框") p.add_argument("selector", help="输入框选择器") p.add_argument("value", help="要输入的文本") p.add_argument("-p", "--profile", default="default", help="Profile 名称") a = p.parse_args(args) browser = await _ensure_stealth_browser(a.profile) result = await browser.fill(a.selector, a.value) _print_result({"success": result.success, "message": result.message, "error": result.error}) async def cmd_stealth_screenshot(args): """反检测浏览器截图""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-screenshot", description="反检测浏览器截图") p.add_argument("-o", "--output", default="", help="保存路径") p.add_argument("-p", "--profile", default="default", help="Profile 名称") a = p.parse_args(args) browser = await _ensure_stealth_browser(a.profile) result = await browser.screenshot(save_path=a.output) if result.success: print(result.message) if result.files: print(f"\n__SEND_FILE__{result.files[0]}|反检测浏览器截图__END__") else: print(f"错误: {result.error}", file=sys.stderr) sys.exit(1) async def cmd_stealth_content(args): """获取反检测浏览器页面内容""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-content", description="获取反检测浏览器页面内容") p.add_argument("-p", "--profile", default="default", help="Profile 名称") a = p.parse_args(args) browser = await _ensure_stealth_browser(a.profile) result = await browser.get_content() if result.success: data = result.data or {} print(f"标题: {data.get('title', '')}") print(f"URL: {data.get('url', '')}") print(f"内容长度: {data.get('text_length', 0)} 字符") if data.get('truncated'): print("(内容已截断，仅显示前 10000 字符)") print("") print(result.output or "") else: print(f"错误: {result.error}", file=sys.stderr) sys.exit(1) async def cmd_stealth_cookies(args): """反检测浏览器 Cookie 管理""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-cookies", description="反检测浏览器 Cookie 管理") p.add_argument("action", choices=["save", "load", "get", "clear"], help="操作") p.add_argument("-p", "--profile", default="default", help="Profile 名称") a = p.parse_args(args) browser = await _ensure_stealth_browser(a.profile) if a.action == "save": result = await browser.save_cookies() elif a.action == "load": result = await browser.load_cookies() elif a.action == "get": result = await browser.get_cookies() elif a.action == "clear": result = await browser.clear_cookies() else: print(f"未知操作: {a.action}", file=sys.stderr) sys.exit(1) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_stealth_wait(args): """反检测浏览器等待元素""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-wait", description="反检测浏览器等待元素出现") p.add_argument("selector", help="元素选择器") p.add_argument("-t", "--timeout", type=int, default=10, help="超时秒数") p.add_argument("-p", "--profile", default="default", help="Profile 名称") a = p.parse_args(args) browser = await _ensure_stealth_browser(a.profile) result = await browser.wait_for(a.selector, timeout=float(a.timeout)) _print_result({"success": result.success, "message": result.message, "error": result.error}) async def cmd_stealth_wait_manual(args): """等待用户手动操作""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-wait-manual", description="等待用户手动操作 (验证码/2FA)") p.add_argument("reason", nargs="?", default="验证码", help="等待原因") p.add_argument("-t", "--timeout", type=int, default=300, help="最长等待秒数") p.add_argument("-p", "--profile", default="default", help="Profile 名称") a = p.parse_args(args) browser = await _ensure_stealth_browser(a.profile) print(f"等待用户手动操作: {a.reason} (最多 {a.timeout} 秒)") print("请通过 VNC 远程桌面或 Web Control 面板操作浏览器...") result = await browser.wait_for_manual(reason=a.reason, timeout=float(a.timeout)) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_stealth_close(args): """关闭反检测浏览器""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-close", description="关闭反检测浏览器") p.add_argument("profile", nargs="?", default="", help="Profile 名称（空=关闭所有）") a = p.parse_args(args) from aiskills.browser_stealth import close_stealth_browser # close_stealth_browser 现在是同步函数 close_stealth_browser(profile_name=a.profile) print("反检测浏览器已关闭") async def cmd_stealth_eval(args): """反检测浏览器执行 JS""" import argparse p = argparse.ArgumentParser(prog="myagent-ai stealth-eval", description="反检测浏览器执行 JavaScript") p.add_argument("script", help="JavaScript 代码") p.add_argument("-p", "--profile", default="default", help="Profile 名称") a = p.parse_args(args) browser = await _ensure_stealth_browser(a.profile) result = await browser.evaluate(a.script) if result.success: data = result.data or {} print(data.get("result", "")) else: print(f"错误: {result.error}", file=sys.stderr) sys.exit(1) # ============================================================================= # Profile 管理命令 # ============================================================================= async def cmd_profile_list(args): """列出所有浏览器 Profile""" from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() profiles = mgr.list_profiles() if not profiles: print("暂无已创建的 Profile") print(f"\n可用预置网站: {', '.join(mgr.get_preset_names())}") print("提示: 使用 'myagent-ai stealth-open <预置名>' 自动创建") return print(f"共 {len(profiles)} 个 Profile:\n") for pf in profiles: size = pf.get('size_mb', 0) cookies = 'yes' if pf.get('has_cookies') else 'no' print(f" {pf['name']:15s} {pf.get('display_name', ''):20s} " f"cookies={cookies:3s} size={size:.1f}MB") print(f"\n总占用: {mgr.get_total_size_mb():.1f}MB") async def cmd_profile_create(args): """创建浏览器 Profile""" import argparse p = argparse.ArgumentParser(prog="myagent-ai profile-create", description="创建浏览器 Profile") p.add_argument("name", help="Profile 名称") p.add_argument("--display", default="", help="显示名称") p.add_argument("--login-url", default="", help="登录页 URL") a = p.parse_args(args) from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() profile = mgr.create_profile( name=a.name, display_name=a.display, login_url=a.login_url, ) print(f"Profile 已创建: {a.name}") print(f" 目录: {profile.profile_dir}") if profile.login_url: print(f" 登录页: {profile.login_url}") async def cmd_profile_delete(args): """删除浏览器 Profile""" import argparse p = argparse.ArgumentParser(prog="myagent-ai profile-delete", description="删除浏览器 Profile") p.add_argument("name", help="Profile 名称") a = p.parse_args(args) from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() ok = mgr.delete_profile(a.name) if ok: print(f"Profile 已删除: {a.name}") else: print(f"删除失败: {a.name}", file=sys.stderr) sys.exit(1) # ============================================================================= # Site 管理命令 (v1.31.0) # ============================================================================= async def cmd_site_manage(args): """网站注册管理 — 列出/查看/添加/删除/初始化网站""" if not args: args = ["list"] subcmd = args[0] subargs = args[1:] if subcmd == "list": await _site_manage_list(subargs) elif subcmd == "show": await _site_manage_show(subargs) elif subcmd == "add": await _site_manage_add(subargs) elif subcmd == "remove": await _site_manage_remove(subargs) elif subcmd == "init": await _site_manage_init(subargs) else: print(f"未知子命令: {subcmd}", file=sys.stderr) print("可用子命令: list, show, add, remove, init", file=sys.stderr) sys.exit(1) async def _site_manage_list(args): """列出所有注册网站""" import argparse p = argparse.ArgumentParser(prog="myagent-ai site-manage list", description="列出所有注册网站") p.add_argument("--category", "-c", default="", help="按分类筛选 (email/social/cloud/ai_chat/dev/content)") a = p.parse_args(args) from core.site_registry import get_site_registry reg = get_site_registry() sites = reg.list_sites(category=a.category) if not sites: print("暂无注册网站") return # 按分类分组显示 current_cat = "" for s in sites: cat = s.get("category", "custom") if cat != current_cat: current_cat = cat from core.site_registry import SITE_CATEGORIES cat_info = SITE_CATEGORIES.get(cat, {"name": cat, "icon": "??"}) print(f"\n {cat_info.get('icon', '')} {cat_info.get('name', cat)}") print(f" {'─' * 50}") profile_mark = "*" if s.get("has_profile") else " " custom_mark = "[自定义]" if s.get("is_custom") else "" print(f" {profile_mark} {s['name']:20s} {s.get('display_name', ''):16s} {custom_mark}") print(f"\n * = 已创建 Profile 共 {len(sites)} 个网站") print(f"\n 使用 'myagent-ai site-manage show <名称>' 查看详情") print(f" 使用 'myagent-ai site-manage add <名称>' 添加自定义网站") async def _site_manage_show(args): """显示网站详细配置""" if not args: print("错误: 请提供网站名称", file=sys.stderr) print("用法: myagent-ai site-manage show <名称>", file=sys.stderr) sys.exit(1) name = args[0] from core.site_registry import get_site_registry reg = get_site_registry() site = reg.get_site(name) if not site: print(f"错误: 网站不存在: {name}", file=sys.stderr) print(f"提示: 使用 'myagent-ai site-manage list' 查看所有可用网站", file=sys.stderr) sys.exit(1) print(f" 网站: {site.get('display_name', name)}") print(f" 名称: {site.get('name', '')}") print(f" 分类: {site.get('category', '')}") print(f" 类型: {'内置' if site.get('is_builtin') else '自定义'}") if site.get("has_profile"): print(f" Profile: 已创建") else: print(f" Profile: 未创建") if site.get("login_url"): print(f" 登录页: {site['login_url']}") if site.get("home_url"): print(f" 首页: {site['home_url']}") if site.get("detect_url"): print(f" 检测页: {site['detect_url']}") urls = site.get("urls", {}) if urls: print(f"\n 可用页面:") for key, url in urls.items(): print(f" {key:12s} {url}") tips = site.get("tips", []) if tips: print(f"\n 操作提示:") for tip in tips: print(f" - {tip}") async def _site_manage_add(args): """添加自定义网站""" import argparse p = argparse.ArgumentParser(prog="myagent-ai site-manage add", description="添加自定义网站") p.add_argument("name", help="网站名称（英文，如 my_site）") p.add_argument("--display", "-d", default="", help="显示名称") p.add_argument("--login-url", "-l", default="", help="登录页 URL") p.add_argument("--home-url", "-u", default="", help="首页 URL") p.add_argument("--category", "-c", default="custom", help="分类") a = p.parse_args(args) from core.site_registry import get_site_registry reg = get_site_registry() try: site = reg.add_site({ "name": a.name, "display_name": a.display or a.name, "login_url": a.login_url, "home_url": a.home_url, "category": a.category, }) print(f"已添加网站: {a.name}") if site.get("display_name"): print(f" 显示名称: {site['display_name']}") if site.get("login_url"): print(f" 登录页: {site['login_url']}") print(f" 配置文件: {reg.base_dir / f'{a.name}.json'}") except ValueError as e: print(f"错误: {e}", file=sys.stderr) sys.exit(1) async def _site_manage_remove(args): """删除自定义网站""" if not args: print("错误: 请提供网站名称", file=sys.stderr) sys.exit(1) name = args[0] from core.site_registry import get_site_registry reg = get_site_registry() ok = reg.remove_site(name) if ok: print(f"已删除网站: {name}") else: print(f"删除失败: {name}（内置网站不可删除，或网站不存在）", file=sys.stderr) sys.exit(1) async def _site_manage_init(args): """批量初始化网站 Profile""" from core.site_registry import get_site_registry reg = get_site_registry() initialized = reg.init_profiles() if initialized: print(f"已初始化 {len(initialized)} 个网站的 Profile:") for name in initialized: print(f" - {name}") else: print("所有网站 Profile 已初始化（无新 Profile 创建）") # ============================================================================= # GUI 命令 # ============================================================================= async def cmd_screenshot(args): """屏幕截图""" import argparse p = argparse.ArgumentParser(prog="myagent-ai screenshot", description="屏幕截图") p.add_argument("region", nargs="?", default="", help="截图区域 'left,top,width,height' (空=全屏)") p.add_argument("-m", "--monitor", type=int, default=1, help="显示器编号 (默认1)") a = p.parse_args(args) from aiskills.gui_skill import ScreenShotSkill skill = ScreenShotSkill() result = await skill.execute(region=a.region, monitor=a.monitor) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) if result.success and result.files: print(f"\n__SEND_FILE__{result.files[0]}|屏幕截图__END__") async def cmd_mouse_click(args): """鼠标点击""" import argparse p = argparse.ArgumentParser(prog="myagent-ai mouse-click", description="鼠标点击") p.add_argument("x", type=int, help="X 坐标") p.add_argument("y", type=int, help="Y 坐标") p.add_argument("-b", "--button", default="left", choices=["left", "right", "middle"]) p.add_argument("-c", "--clicks", type=int, default=1, help="点击次数 (默认1)") a = p.parse_args(args) from aiskills.gui_skill import MouseClickSkill skill = MouseClickSkill() result = await skill.execute(x=a.x, y=a.y, button=a.button, clicks=a.clicks) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_mouse_drag(args): """鼠标拖拽""" import argparse p = argparse.ArgumentParser(prog="myagent-ai mouse-drag", description="鼠标拖拽") p.add_argument("start_x", type=int, help="起点 X") p.add_argument("start_y", type=int, help="起点 Y") p.add_argument("end_x", type=int, help="终点 X") p.add_argument("end_y", type=int, help="终点 Y") p.add_argument("-d", "--duration", type=float, default=0.3, help="持续时间 (秒)") a = p.parse_args(args) from aiskills.gui_skill import MouseDragSkill skill = MouseDragSkill() result = await skill.execute(start_x=a.start_x, start_y=a.start_y, end_x=a.end_x, end_y=a.end_y, duration=a.duration) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_type_text(args): """输入文本""" import argparse p = argparse.ArgumentParser(prog="myagent-ai type-text", description="在当前光标位置输入文本") p.add_argument("text", help="要输入的文本") p.add_argument("--clear", action="store_true", help="先清空已有内容") a = p.parse_args(args) from aiskills.gui_skill import TypeTextSkill skill = TypeTextSkill() result = await skill.execute(text=a.text, clear_first=a.clear) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_hotkey(args): """按下快捷键""" import argparse p = argparse.ArgumentParser(prog="myagent-ai hotkey", description="按下键盘快捷键") p.add_argument("action", help="快捷键 (如 copy, paste, ctrl+c, alt+tab)") a = p.parse_args(args) from aiskills.gui_skill import HotkeySkill skill = HotkeySkill() result = await skill.execute(action=a.action) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_window_list(args): """列出窗口""" import argparse p = argparse.ArgumentParser(prog="myagent-ai window-list", description="列出所有打开的窗口") p.add_argument("--filter", "-f", default="", help="窗口标题过滤") a = p.parse_args(args) from aiskills.gui_skill import WindowListSkill skill = WindowListSkill() result = await skill.execute(filter=a.filter) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) async def cmd_window_focus(args): """聚焦窗口""" import argparse p = argparse.ArgumentParser(prog="myagent-ai window-focus", description="聚焦/置顶窗口") p.add_argument("title", help="窗口标题 (模糊匹配)") p.add_argument("--maximize", action="store_true", help="最大化窗口") a = p.parse_args(args) from aiskills.gui_skill import WindowFocusSkill skill = WindowFocusSkill() result = await skill.execute(title=a.title, maximize=a.maximize) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) # ============================================================================= # 记忆命令 # ============================================================================= async def cmd_memory(args): """搜索记忆""" import argparse p = argparse.ArgumentParser(prog="myagent-ai memory", description="搜索历史记忆") p.add_argument("--keyword", "-k", default="", help="搜索关键词") p.add_argument("--time", "-t", default="", help="时间点筛选 (如 2025-01)") p.add_argument("--session", "-s", default="", help="会话 ID") p.add_argument("--limit", "-n", type=int, default=5, help="返回条数 (默认5)") a = p.parse_args(args) # memory 需要通过 ToolDispatcher 调用 (依赖 MemoryAgent) # CLI 模式下直接读取 SQLite 数据库 db_path = Path.home() / ".myagent" / "data" / "memory.db" if not db_path.exists(): print("未找到记忆数据库，记忆功能仅在 web 模式下可用") return try: import sqlite3 conn = sqlite3.connect(str(db_path)) conn.row_factory = sqlite3.Row cursor = conn.cursor() conditions = [] params = [] if a.keyword: conditions.append("content LIKE ?") params.append(f"%{a.keyword}%") if a.time: conditions.append("created_at LIKE ?") params.append(f"%{a.time}%") if a.session: conditions.append("session_id = ?") params.append(a.session) where = " AND ".join(conditions) if conditions else "1=1" query = f""" SELECT session_id, role, content, key, created_at, importance FROM memories WHERE {where} ORDER BY created_at DESC LIMIT ? """ params.append(a.limit) rows = cursor.execute(query, params).fetchall() conn.close() if not rows: print("未找到相关记忆") return print(f"找到 {len(rows)} 条记忆:\n") for i, row in enumerate(rows, 1): print(f"{i}. [{row['created_at']}] [{row['key']}] {row['content'][:200]}") if row['session_id']: print(f" 会话: {row['session_id'][:20]}") except Exception as e: print(f"记忆搜索失败: {e}", file=sys.stderr) sys.exit(1) # ============================================================================= # 帮助命令 # ============================================================================= async def cmd_screen_element(args): """屏幕元素识别""" import argparse p = argparse.ArgumentParser(prog="myagent-ai screen-element", description="屏幕元素识别 — 通过描述定位屏幕上的元素") p.add_argument("description", help="元素描述 (如 '确定按钮', '搜索框')") p.add_argument("region", nargs="?", default="", help="搜索区域 'left,top,width,height'") a = p.parse_args(args) from aiskills.gui_skill import ScreenElementSkill skill = ScreenElementSkill() result = await skill.execute(description=a.description, region=a.region) _print_result({"success": result.success, "message": result.message, "data": result.data, "error": result.error}) # ============================================================================= # 媒体播放命令 # ============================================================================= async def cmd_playaudio(args): """播放音频 — 通过标记触发 SSE 嵌入播放器""" import argparse p = argparse.ArgumentParser(prog="myagent-ai playaudio", description="在聊天中内嵌播放音频") group = p.add_mutually_exclusive_group(required=True) group.add_argument("--url", default="", help="音频 URL") group.add_argument("--file", default="", help="本地音频文件路径") p.add_argument("--title", default="", help="音频标题") a = p.parse_args(args) source = a.url or a.file title = a.title or (os.path.basename(a.file) if a.file else "音频") # 输出标记，由 ToolDispatcher._exec_command 检测并触发 SSE print(f"__EMBED_AUDIO__{source}|{title}__END__") async def cmd_playvideo(args): """播放视频 — 通过标记触发 SSE 嵌入播放器""" import argparse p = argparse.ArgumentParser(prog="myagent-ai playvideo", description="在聊天中内嵌播放视频") group = p.add_mutually_exclusive_group(required=True) group.add_argument("--url", default="", help="视频 URL") group.add_argument("--file", default="", help="本地视频文件路径") p.add_argument("--title", default="", help="视频标题") a = p.parse_args(args) source = a.url or a.file title = a.title or (os.path.basename(a.file) if a.file else "视频") # 输出标记，由 ToolDispatcher._exec_command 检测并触发 SSE print(f"__EMBED_VIDEO__{source}|{title}__END__") # ============================================================================= # 通信命令 # ============================================================================= async def cmd_chat(args): """Agent间通信 — 向群内其他Agent发送消息或文件""" import argparse import logging _chat_logger = logging.getLogger("myagent.chat") _chat_logger.setLevel(logging.INFO) if not _chat_logger.handlers: _h = logging.StreamHandler(sys.stdout) _h.setFormatter(logging.Formatter("[chat] %(message)s")) _chat_logger.addHandler(_h) p = argparse.ArgumentParser(prog="myagent-ai chat", description="向群内其他Agent发送消息或文件，消息会出现在群聊记录中") p.add_argument("--agent", "-a", required=True, help="目标Agent的ID或名称（群成员列表中显示）") p.add_argument("--message", "-m", default="", help="要发送的消息内容") p.add_argument("--file", "-f", default="", help="要发送的文件路径（可多次使用: -f file1 -f file2）") p.add_argument("--group", "-g", help="群ID（可选，默认使用最近活跃的群）") a = p.parse_args(args) if not a.message and not a.file: print("错误: 必须指定 --message 或 --file", file=sys.stderr) sys.exit(1) _chat_logger.info(f"发起私聊: 目标={a.agent}, 消息={'(文件)' if not a.message else a.message[:80]}, 群={a.group or '默认'}") # [v1.23.75] Agent 查找：从数据库按 ID 匹配 _config_file = Path.home() / ".myagent" / "config.json" _base_dir = Path.home() / ".myagent" _db_path = Path.home() / ".myagent" / "data" / "agents.db" try: if _config_file.exists(): _cfg_raw = json.loads(_config_file.read_text(encoding="utf-8")) _custom_dir = _cfg_raw.get("data_dir", "") if _custom_dir: _expanded = os.path.expanduser(_custom_dir) if not os.path.isabs(_expanded): _expanded = str(Path.home() / _expanded) _base_dir = Path(_expanded) except Exception: pass resolved_path = None found_ids = [] # 优先从数据库查询（支持 ID 或 名称匹配） if _db_path.exists(): try: import sqlite3 conn = sqlite3.connect(str(_db_path)) conn.row_factory = sqlite3.Row rows = conn.execute("SELECT path, id, name FROM agents WHERE id != ''").fetchall() conn.close() # 两轮匹配：第一轮按 ID 精确匹配，第二轮按名称兜底 _name_fallback = None for row in rows: aid = row["id"] aname = row["name"] or "" found_ids.append(aid) if aid == a.agent: resolved_path = row["path"] break if aname == a.agent and _name_fallback is None: _name_fallback = row["path"] if not resolved_path and _name_fallback: resolved_path = _name_fallback except Exception: pass # 数据库查不到，回退到文件系统（兼容旧版本，支持 ID 或 名称匹配） if not resolved_path: agents_data_dir = _base_dir / "data" / "agents" if agents_data_dir.exists(): # 两轮匹配：第一轮按 ID 精确匹配，第二轮按名称兜底 _name_fallback_fs = None for cfg_path in agents_data_dir.rglob("config.json"): try: ag_cfg = json.loads(cfg_path.read_text(encoding="utf-8")) aid = ag_cfg.get("id", "") aname = ag_cfg.get("name", "") if aid and aid not in found_ids: found_ids.append(aid) if aid == a.agent: resolved_path = cfg_path.parent.relative_to(agents_data_dir).as_posix() break if aname and aname == a.agent and _name_fallback_fs is None: _name_fallback_fs = cfg_path.parent.relative_to(agents_data_dir).as_posix() except Exception: pass if not resolved_path and _name_fallback_fs: resolved_path = _name_fallback_fs if not resolved_path: err_parts = [f"错误: Agent 不存在: {a.agent}"] if not _db_path.exists() and not (_base_dir / "data" / "agents").exists(): err_parts.append(f"agents 数据不存在（数据库和目录均未找到）") else: cfg_count = len(found_ids) err_parts.append(f"共扫描到 {cfg_count} 个 Agent") if cfg_count > 0: sample = found_ids[:5] err_parts.append(f"已有 ID: {', '.join(sample)}") err_parts.append(f"（请使用Agent的ID或名称，从群成员列表中获取）") print(" | ".join(err_parts), file=sys.stderr) sys.exit(1) # 从数据库获取 agent_name agent_name = resolved_path try: conn = sqlite3.connect(str(_db_path)) conn.row_factory = sqlite3.Row row = conn.execute("SELECT name FROM agents WHERE path = ?", (resolved_path,)).fetchone() conn.close() if row and row["name"]: agent_name = row["name"] except Exception: pass # 验证文件是否存在 files_info = "" if a.file: import glob as _glob # 支持多个 -f 参数 file_paths = a.file.split("|") valid_files = [] for fp in file_paths: fp = fp.strip() if not fp: continue # 支持通配符 matched = _glob.glob(fp) if not matched: print(f"错误: 文件不存在: {fp}", file=sys.stderr) sys.exit(1) for mf in matched: abs_mf = os.path.abspath(mf) if os.path.isfile(abs_mf): valid_files.append(abs_mf) if valid_files: files_info = "|".join(valid_files) # Output chat marker # 格式: __CHAT_AGENT__agent_path|agent_name|message|files__END__ # files 为空时使用3段格式（避免尾部 | 导致解析时消息带 | 后缀） if files_info: print(f"__CHAT_AGENT__{resolved_path}|{agent_name}|{a.message}|{files_info}__END__") else: print(f"__CHAT_AGENT__{resolved_path}|{agent_name}|{a.message}__END__") _chat_logger.info(f"私聊标记已输出: → {agent_name} ({resolved_path}), 消息: {a.message[:80]}") async def cmd_zip(args): """压缩文件或文件夹""" import argparse import zipfile p = argparse.ArgumentParser(prog="myagent-ai zip", description="压缩文件或文件夹为 zip") p.add_argument("paths", nargs="+", help="要压缩的文件或文件夹路径（支持多个）") p.add_argument("-o", "--output", "--output_path", default="", help="输出 zip 文件路径（默认: 第一个路径名.zip）") p.add_argument("--exclude", nargs="*", default=[], help="要排除的文件模式（如 __pycache__ *.pyc .git）") a = p.parse_args(args) import fnmatch excluded = set(a.exclude) if a.exclude else set() # 确定输出路径 if a.output: out_path = os.path.abspath(a.output) else: first = a.paths[0].rstrip(os.sep) out_path = os.path.abspath(first + ".zip") # 确保输出目录存在 os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True) file_count = 0 total_size = 0 with zipfile.ZipFile(out_path, 'w', zipfile.ZIP_DEFLATED) as zf: for src in a.paths: src = os.path.abspath(src) if os.path.isfile(src): # 检查是否排除 fname = os.path.basename(src) skip = False for pat in excluded: if fnmatch.fnmatch(fname, pat): skip = True break if skip: continue zf.write(src, fname) file_count += 1 total_size += os.path.getsize(src) elif os.path.isdir(src): for root, dirs, files in os.walk(src): # 过滤排除的目录 dirs[:] = [d for d in dirs if not any(fnmatch.fnmatch(d, p) for p in excluded)] for f in files: if any(fnmatch.fnmatch(f, p) for p in excluded): continue full = os.path.join(root, f) arcname = os.path.relpath(full, os.path.dirname(src)) zf.write(full, arcname) file_count += 1 total_size += os.path.getsize(full) else: print(f"错误: 路径不存在: {src}", file=sys.stderr) sys.exit(1) size_str = f"{total_size / 1024:.1f} KB" if total_size < 1024 * 1024 else f"{total_size / (1024 * 1024):.1f} MB" print(f"压缩完成: {out_path} ({file_count} 个文件, {size_str})") # 输出文件标记，自动发送给用户 print(f"__SEND_FILE__{out_path}|压缩文件__END__") async def cmd_unzip(args): """解压缩 zip 文件""" import argparse import zipfile p = argparse.ArgumentParser(prog="myagent-ai unzip", description="解压缩 zip 文件") p.add_argument("zipfile", help="zip 文件路径") p.add_argument("-o", "--output", "--output_path", default="", help="输出目录（默认: zip 文件所在目录，去除 .zip 后缀）") a = p.parse_args(args) zip_path = os.path.abspath(a.zipfile) if not os.path.isfile(zip_path): print(f"错误: 文件不存在: {zip_path}", file=sys.stderr) sys.exit(1) # 确定输出目录 if a.output: out_dir = os.path.abspath(a.output) else: out_dir = os.path.splitext(zip_path)[0] os.makedirs(out_dir, exist_ok=True) file_count = 0 with zipfile.ZipFile(zip_path, 'r') as zf: zf.extractall(out_dir) file_count = len(zf.namelist()) print(f"解压完成: {out_dir} ({file_count} 个文件)") # ============================================================================= # 帮助命令 # ============================================================================= async def cmd_help(args): """显示帮助信息""" help_text = """ MyAgent AI - 命令行工具集 ======================== 感知: myagent-ai ocr <图片路径> [ch|en] OCR 文字识别 myagent-ai analyze-image <图片路径> [提示词] 图片内容分析 (VLM) myagent-ai transcribe <音频路径> [zh|en|ja] 语音转文字 搜索: myagent-ai search <关键词> [-n 数量] 网络搜索 myagent-ai read-url [--raw] 读取网页正文 myagent-ai fetch-url [-m METHOD] [-d DATA] 获取 URL 原始内容 (API) 文件操作: myagent-ai read <文件路径> [--offset N] [--limit N] 读取文件 myagent-ai write <文件路径> [-c 内容] [--append] 写入文件 myagent-ai ls <目录> [-p *.py] [-r] [--max N] 列出目录 myagent-ai rm <路径> [-r] 删除文件/目录 myagent-ai grep <关键词> <目录> [-p *.py] 搜索文件内容 myagent-ai mv <源路径> <目标路径> 移动/重命名 myagent-ai send-file <文件路径> [描述] 发送文件给用户 文档: myagent-ai docx-create -c [-t 标题] 创建 Word 文档 myagent-ai docx-read <文件路径> 读取 Word 文档 myagent-ai xlsx-create -s [-t 标题] 创建 Excel 文件 myagent-ai xlsx-read <文件路径> [--sheet 名称] 读取 Excel 文件 myagent-ai xlsx-edit <文件路径> -d 编辑 Excel 文件 myagent-ai ppt-create -s [--theme] 创建 PowerPoint myagent-ai ppt-read <文件路径> 读取 PowerPoint myagent-ai pdf-create -c [--palette] 创建 PDF 文件 myagent-ai pdf-read <文件路径> [--start N] [--end N] 读取 PDF 文件 系统: myagent-ai sysinfo 系统信息 myagent-ai ps [--filter 名称] [--limit N] 列出进程 myagent-ai env [KEY] 环境变量 myagent-ai pathinfo <路径> 路径信息 浏览器 (chrome-devtools-mcp): myagent-ai browser-open [--no-headless] 打开网页 myagent-ai browser-screenshot [-o 输出路径] 浏览器截图 myagent-ai browser-click 点击元素 myagent-ai browser-fill <选择器> <值> 填写输入框 myagent-ai browser-eval 执行 JavaScript myagent-ai browser-navigate 标签页管理 myagent-ai browser-close 关闭浏览器 GUI (仅 Windows/macOS): myagent-ai screenshot [区域] [-m 显示器] 屏幕截图 myagent-ai mouse-click [-b 按键] [-c 次数] 鼠标点击 myagent-ai mouse-drag 鼠标拖拽 myagent-ai type-text <文本> [--clear] 输入文本 myagent-ai hotkey <快捷键> 按下快捷键 myagent-ai window-list [--filter 关键词] 列出窗口 myagent-ai window-focus <标题> [--maximize] 聚焦窗口 myagent-ai screen-element <描述> [区域] 屏幕元素识别 记忆: myagent-ai memory [--keyword 关键词] [--limit N] 搜索历史记忆 通信: myagent-ai chat --agent --message "消息" 向其他Agent发送消息 myagent-ai chat --agent --file <文件路径> 向其他Agent发送文件 myagent-ai chat --agent -m "消息" -f <文件> 同时发送消息和文件 压缩解压: myagent-ai zip <路径1> [路径2] ... [-o 输出.zip] 压缩文件或文件夹 myagent-ai unzip <文件.zip> [-o 输出目录] 解压 zip 文件 媒体播放: myagent-ai playaudio --url [--title] 播放在线音频 myagent-ai playaudio --file <路径> [--title] 播放本地音频 myagent-ai playvideo --url [--title] 播放在线视频 myagent-ai playvideo --file <路径> [--title] 播放本地视频 """.strip() print(help_text) # ============================================================================= # 入口 # ============================================================================= if __name__ == "__main__": asyncio.run(_run())