""" Browser-Use 原生工具适配器 Native Browser-Use Tools Adapter 直接使用 browser-use 的原生类（BrowserSession, Tools）实现所有浏览器操作工具。 不依赖 Playwright，完全基于 CDP 协议。 核心特性： 1. 浏览器会话持久化 - 只启动一次浏览器 2. 状态自动保持 - 登录状态、Cookie、LocalStorage 等 3. 完整的底层访问 - 可以直接使用 CDP 协议 4. 性能优异 - 避免频繁创建/销毁浏览器实例 5. 多种浏览器类型 - 支持 local、cloud、container 三种模式 支持的浏览器类型： 1. Local (本地浏览器): - 在本地运行 Chrome - 支持可视化调试 - 速度最快 - 示例: init_browser_session(browser_type="local") 2. Cloud (云浏览器): - 在云端运行 - 不占用本地资源 - 适合生产环境 - 示例: init_browser_session(browser_type="cloud") 3. Container (容器浏览器): - 在独立容器中运行 - 隔离性好 - 支持预配置账户 - 示例: init_browser_session(browser_type="container", container_url="https://example.com") 使用方法： 1. 在 Agent 初始化时调用 init_browser_session() 并指定 browser_type 2. 使用各个工具函数执行浏览器操作 3. 任务结束时调用 cleanup_browser_session() 文件操作说明： - 浏览器专用文件目录：.cache/.browser_use_files/ (在当前工作目录下) 用于存储浏览器会话产生的临时文件（下载、上传、截图等） - 一般文件操作：请使用 agent.tools.builtin 中的文件工具 (read_file, write_file, edit_file) 这些工具功能更完善，支持diff预览、智能匹配、分页读取等 """ import logging import sys import os import json import httpx import asyncio import aiohttp import re import base64 from urllib.parse import urlparse, parse_qs, unquote from typing import Optional, List, Dict, Any, Tuple, Union from pathlib import Path from langchain_core.runnables import RunnableLambda from argparse import Namespace # 使用 Namespace 快速构造带属性的对象 from langchain_core.messages import AIMessage from ....llm.openrouter import openrouter_llm_call # 将项目根目录添加到 Python 路径 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # 导入框架的工具装饰器和结果类 from agent.tools import tool, ToolResult from agent.tools.builtin.browser.sync_mysql_help import mysql # 导入 browser-use 的核心类 from browser_use import BrowserSession, BrowserProfile from browser_use.tools.service import Tools try: from browser_use.tools.views import ReadContentAction # type: ignore except Exception: from pydantic import BaseModel class ReadContentAction(BaseModel): goal: str source: str = "page" context: str = "" from browser_use.agent.views import ActionResult from browser_use.filesystem.file_system import FileSystem # ============================================================ # 无需注册的内部辅助函数 # ============================================================ # ============================================================ # 全局浏览器会话管理 # ============================================================ # 全局变量：浏览器会话和工具实例 _browser_session: Optional[BrowserSession] = None _browser_tools: Optional[Tools] = None _file_system: Optional[FileSystem] = None async def create_container(url: str, account_name: str = "liuwenwu") -> Dict[str, Any]: """ 创建浏览器容器并导航到指定URL 按照 test.md 的要求： 1.1 调用接口创建容器 1.2 调用接口创建窗口并导航到URL Args: url: 要导航的URL地址 account_name: 账户名称 Returns: 包含容器信息的字典: - success: 是否成功 - container_id: 容器ID - vnc: VNC访问URL - cdp: CDP协议URL（用于浏览器连接） - connection_id: 窗口连接ID - error: 错误信息（如果失败） """ result = { "success": False, "container_id": None, "vnc": None, "cdp": None, "connection_id": None, "error": None } try: async with aiohttp.ClientSession() as session: # 步骤1.1: 创建容器 print("📦 步骤1.1: 创建容器...") create_url = "http://47.84.182.56:8200/api/v1/container/create" create_payload = { "auto_remove": True, "need_port_binding": True, "max_lifetime_seconds": 900 } async with session.post(create_url, json=create_payload) as resp: if resp.status != 200: raise RuntimeError(f"创建容器失败: HTTP {resp.status}") create_result = await resp.json() if create_result.get("code") != 0: raise RuntimeError(f"创建容器失败: {create_result.get('msg')}") data = create_result.get("data", {}) result["container_id"] = data.get("container_id") result["vnc"] = data.get("vnc") result["cdp"] = data.get("cdp") print(f"✅ 容器创建成功") print(f" Container ID: {result['container_id']}") print(f" VNC: {result['vnc']}") print(f" CDP: {result['cdp']}") # 等待容器内的浏览器启动 print(f"\n⏳ 等待容器内浏览器启动...") await asyncio.sleep(5) # 步骤1.2: 创建页面并导航 print(f"\n📱 步骤1.2: 创建页面并导航到 {url}...") page_create_url = "http://47.84.182.56:8200/api/v1/browser/page/create" page_payload = { "container_id": result["container_id"], "url": url, "account_name": account_name, "need_wait": True, "timeout": 30 } # 重试机制：最多尝试3次 max_retries = 3 page_created = False last_error = None for attempt in range(max_retries): try: if attempt > 0: print(f" 重试 {attempt + 1}/{max_retries}...") await asyncio.sleep(3) # 重试前等待 async with session.post(page_create_url, json=page_payload, timeout=aiohttp.ClientTimeout(total=60)) as resp: if resp.status != 200: response_text = await resp.text() last_error = f"HTTP {resp.status}: {response_text[:200]}" continue page_result = await resp.json() if page_result.get("code") != 0: last_error = f"{page_result.get('msg')}" continue page_data = page_result.get("data", {}) result["connection_id"] = page_data.get("connection_id") result["success"] = True page_created = True print(f"✅ 页面创建成功") print(f" Connection ID: {result['connection_id']}") break except asyncio.TimeoutError: last_error = "请求超时" continue except aiohttp.ClientError as e: last_error = f"网络错误: {str(e)}" continue except Exception as e: last_error = f"未知错误: {str(e)}" continue if not page_created: raise RuntimeError(f"创建页面失败（尝试{max_retries}次后）: {last_error}") except Exception as e: result["error"] = str(e) print(f"❌ 错误: {str(e)}") return result async def init_browser_session( browser_type: str = "local", headless: bool = False, url: Optional[str] = None, profile_name: str = "default", user_data_dir: Optional[str] = None, browser_profile: Optional[BrowserProfile] = None, **kwargs ) -> tuple[BrowserSession, Tools]: global _browser_session, _browser_tools, _file_system if _browser_session is not None: return _browser_session, _browser_tools valid_types = ["local", "cloud", "container"] if browser_type not in valid_types: raise ValueError(f"无效的 browser_type: {browser_type}") # --- 核心：定义本地统一存储路径 --- save_dir = Path.cwd() / ".cache/.browser_use_files" save_dir.mkdir(parents=True, exist_ok=True) # 基础参数配置 session_params = { "headless": headless, # 告诉 Playwright 所有的下载临时流先存入此本地目录 "downloads_path": str(save_dir), } if browser_type == "container": print("🐳 使用容器浏览器模式") if not url: url = "about:blank" container_info = await create_container(url=url, account_name=profile_name) if not container_info["success"]: raise RuntimeError(f"容器创建失败: {container_info['error']}") session_params["cdp_url"] = container_info["cdp"] await asyncio.sleep(3) elif browser_type == "cloud": print("🌐 使用云浏览器模式") session_params["use_cloud"] = True if profile_name and profile_name != "default": session_params["cloud_profile_id"] = profile_name else: # local print("💻 使用本地浏览器模式") session_params["is_local"] = True if user_data_dir is None and profile_name: user_data_dir = str(Path.home() / ".browser_use" / "profiles" / profile_name) Path(user_data_dir).mkdir(parents=True, exist_ok=True) session_params["user_data_dir"] = user_data_dir # macOS 路径兼容 import platform if platform.system() == "Darwin": chrome_path = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" if Path(chrome_path).exists(): session_params["executable_path"] = chrome_path if browser_profile: session_params["browser_profile"] = browser_profile session_params.update(kwargs) # 创建会话 _browser_session = BrowserSession(**session_params) await _browser_session.start() _browser_tools = Tools() _file_system = FileSystem(base_dir=str(save_dir)) print(f"✅ 浏览器会话初始化成功 | 默认下载路径: {save_dir}") if browser_type in ["local", "cloud"] and url: await _browser_tools.navigate(url=url, browser_session=_browser_session) return _browser_session, _browser_tools async def get_browser_session() -> tuple[BrowserSession, Tools]: """ 获取当前浏览器会话，如果不存在或连接已断开则自动重新创建 Returns: (BrowserSession, Tools) 元组 """ global _browser_session, _browser_tools, _file_system if _browser_session is not None: # 检查底层 CDP 连接是否仍然存活 # 当 runner.stop() 暂停后用户在菜单停留较久，WebSocket 可能超时断开， # 但 _browser_session 对象仍然存在，导致后续操作抛出 ConnectionClosedError alive = False try: cdp_root = getattr(_browser_session, '_cdp_client_root', None) sess_mgr = getattr(_browser_session, 'session_manager', None) if cdp_root is not None and sess_mgr is not None: cdp_session = await _browser_session.get_or_create_cdp_session() await asyncio.wait_for( cdp_session.cdp_client.send.Runtime.evaluate( params={'expression': '1+1'}, session_id=cdp_session.session_id ), timeout=3.0, ) alive = True except Exception: pass if not alive: print("⚠️ 浏览器会话连接已断开，正在重新初始化...") try: await cleanup_browser_session() except Exception: _browser_session = None _browser_tools = None _file_system = None if _browser_session is None: await init_browser_session() return _browser_session, _browser_tools async def cleanup_browser_session(): """ 清理浏览器会话 优雅地停止浏览器但保留会话状态 """ global _browser_session, _browser_tools, _file_system if _browser_session is not None: await _browser_session.stop() _browser_session = None _browser_tools = None _file_system = None async def kill_browser_session(): """ 强制终止浏览器会话 完全关闭浏览器进程 """ global _browser_session, _browser_tools, _file_system if _browser_session is not None: await _browser_session.kill() _browser_session = None _browser_tools = None _file_system = None # ============================================================ # 辅助函数：ActionResult 转 ToolResult # ============================================================ def action_result_to_tool_result(result: ActionResult, title: str = None) -> ToolResult: """ 将 browser-use 的 ActionResult 转换为框架的 ToolResult Args: result: browser-use 的 ActionResult title: 可选的标题（如果不提供则从 result 推断） Returns: ToolResult """ if result.error: return ToolResult( title=title or "操作失败", output="", error=result.error, long_term_memory=result.long_term_memory or result.error ) return ToolResult( title=title or "操作成功", output=result.extracted_content or "", long_term_memory=result.long_term_memory or result.extracted_content or "", metadata=result.metadata or {} ) def _cookie_domain_for_type(cookie_type: str, url: str) -> Tuple[str, str]: if cookie_type: key = cookie_type.lower() if key in {"xiaohongshu", "xhs"}: return ".xiaohongshu.com", "https://www.xiaohongshu.com" parsed = urlparse(url or "") domain = parsed.netloc or "" domain = domain.replace("www.", "") if domain: domain = f".{domain}" base_url = f"{parsed.scheme}://{parsed.netloc}" if parsed.scheme and parsed.netloc else url return domain, base_url def _parse_cookie_string(cookie_str: str, domain: str, url: str) -> List[Dict[str, Any]]: cookies: List[Dict[str, Any]] = [] if not cookie_str: return cookies parts = cookie_str.split(";") for part in parts: if not part: continue if "=" not in part: continue name, value = part.split("=", 1) cookie = { "name": str(name).strip(), "value": str(value).strip(), "domain": domain, "path": "/", "expires": -1, "httpOnly": False, "secure": True, "sameSite": "None" } if url: cookie["url"] = url cookies.append(cookie) return cookies def _normalize_cookies(cookie_value: Any, domain: str, url: str) -> List[Dict[str, Any]]: if cookie_value is None: return [] if isinstance(cookie_value, list): return cookie_value if isinstance(cookie_value, dict): if "cookies" in cookie_value: return _normalize_cookies(cookie_value.get("cookies"), domain, url) if "name" in cookie_value and "value" in cookie_value: return [cookie_value] return [] if isinstance(cookie_value, (bytes, bytearray)): cookie_value = cookie_value.decode("utf-8", errors="ignore") if isinstance(cookie_value, str): text = cookie_value.strip() if not text: return [] try: parsed = json.loads(text) except Exception: parsed = None if parsed is not None: return _normalize_cookies(parsed, domain, url) return _parse_cookie_string(text, domain, url) return [] def _extract_cookie_value(row: Optional[Dict[str, Any]]) -> Any: if not row: return None # 优先使用 cookies 字段 if "cookies" in row: return row["cookies"] # 兼容其他可能的字段名 for key, value in row.items(): if "cookie" in key.lower(): return value return None def _fetch_cookie_row(cookie_type: str) -> Optional[Dict[str, Any]]: if not cookie_type: return None try: return mysql.fetchone( "select * from agent_channel_cookies where type=%s limit 1", (cookie_type,) ) except Exception: return None def _fetch_profile_id(cookie_type: str) -> Optional[str]: """从数据库获取 cloud_profile_id""" if not cookie_type: return None try: row = mysql.fetchone( "select profileId from agent_channel_cookies where type=%s limit 1", (cookie_type,) ) if row and "profileId" in row: return row["profileId"] return None except Exception: return None # ============================================================ # 需要注册的工具 # ============================================================ # ============================================================ # 导航类工具 (Navigation Tools) # ============================================================ @tool() async def browser_navigate_to_url(url: str, new_tab: bool = False) -> ToolResult: """ 导航到指定的 URL Navigate to a specific URL 使用 browser-use 的原生导航功能，支持在新标签页打开。 Args: url: 要访问的 URL 地址 new_tab: 是否在新标签页中打开（默认 False） Returns: ToolResult: 包含导航结果的工具返回对象 Example: navigate_to_url("https://www.baidu.com") navigate_to_url("https://www.google.com", new_tab=True) """ try: browser, tools = await get_browser_session() # 使用 browser-use 的 navigate 工具 result = await tools.navigate( url=url, new_tab=new_tab, browser_session=browser ) return action_result_to_tool_result(result, f"导航到 {url}") except Exception as e: return ToolResult( title="导航失败", output="", error=f"Failed to navigate to {url}: {str(e)}", long_term_memory=f"导航到 {url} 失败" ) @tool() async def browser_search_web(query: str, engine: str = "bing") -> ToolResult: """ 使用搜索引擎搜索 Search the web using a search engine Args: query: 搜索关键词 engine: 搜索引擎 (google, duckduckgo, bing) - 默认: google Returns: ToolResult: 搜索结果 Example: search_web("Python async programming", engine="google") """ try: browser, tools = await get_browser_session() # 使用 browser-use 的 search 工具 result = await tools.search( query=query, engine=engine, browser_session=browser ) return action_result_to_tool_result(result, f"搜索: {query}") except Exception as e: return ToolResult( title="搜索失败", output="", error=f"Search failed: {str(e)}", long_term_memory=f"搜索 '{query}' 失败" ) @tool() async def browser_go_back() -> ToolResult: """ 返回到上一个页面 Go back to the previous page 模拟浏览器的"后退"按钮功能。 Returns: ToolResult: 包含返回操作结果的工具返回对象 """ try: browser, tools = await get_browser_session() result = await tools.go_back(browser_session=browser) return action_result_to_tool_result(result, "返回上一页") except Exception as e: return ToolResult( title="返回失败", output="", error=f"Failed to go back: {str(e)}", long_term_memory="返回上一页失败" ) @tool() async def browser_wait(seconds: int = 3) -> ToolResult: """ 等待指定的秒数 Wait for a specified number of seconds 用于等待页面加载、动画完成或其他异步操作。 Args: seconds: 等待时间（秒），最大30秒 Returns: ToolResult: 包含等待操作结果的工具返回对象 Example: wait(5) # 等待5秒 """ try: browser, tools = await get_browser_session() result = await tools.wait(seconds=seconds, browser_session=browser) return action_result_to_tool_result(result, f"等待 {seconds} 秒") except Exception as e: return ToolResult( title="等待失败", output="", error=f"Failed to wait: {str(e)}", long_term_memory="等待失败" ) # ============================================================ # 元素交互工具 (Element Interaction Tools) # ============================================================ # 定义一个专门捕获下载链接的 Handler class DownloadLinkCaptureHandler(logging.Handler): def __init__(self): super().__init__() self.captured_url = None def emit(self, record): # 如果已经捕获到了（通常第一条是最完整的），就不再处理后续日志 if self.captured_url: return message = record.getMessage() # 寻找包含下载信息的日志 if "redirection?filename=" in message or "Failed to download" in message: # 使用更严格的正则，确保不抓取带省略号(...)的截断链接 # 排除掉末尾带有三个点的干扰 match = re.search(r"https?://[^\s]+(?!\.\.\.)", message) if match: url = match.group(0) # 再次过滤：如果发现提取出的 URL 确实包含三个点，说明依然抓到了截断版，跳过 if "..." not in url: self.captured_url = url # print(f"🎯 成功锁定完整直链: {url[:50]}...") # 调试用 @tool() async def browser_download_direct_url(url: str, save_name: str = "book.epub") -> ToolResult: save_dir = Path.cwd() / ".cache/.browser_use_files" save_dir.mkdir(parents=True, exist_ok=True) # 提取域名作为 Referer，这能骗过 90% 的防盗链校验 from urllib.parse import urlparse parsed_url = urlparse(url) base_url = f"{parsed_url.scheme}://{parsed_url.netloc}/" # 如果没传 save_name，自动从 URL 获取 if not save_name: import unquote # 尝试从 URL 路径获取文件名并解码（处理中文） save_name = Path(urlparse(url).path).name or f"download_{int(time.time())}" save_name = unquote(save_name) target_path = save_dir / save_name headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "*/*", "Referer": base_url, # 动态设置 Referer "Range": "bytes=0-", # 有时对大文件下载有奇效 } try: print(f"🚀 开始下载: {url[:60]}...") # 使用 follow_redirects=True 处理链接中的 redirection async with httpx.AsyncClient(headers=headers, follow_redirects=True, timeout=60.0) as client: async with client.stream("GET", url) as response: if response.status_code != 200: print(f"❌ 下载失败，HTTP 状态码: {response.status_code}") return # 获取实际文件名（如果服务器提供了） # 这里会优先使用你指定的 save_name with open(target_path, "wb") as f: downloaded_bytes = 0 async for chunk in response.aiter_bytes(): f.write(chunk) downloaded_bytes += len(chunk) if downloaded_bytes % (1024 * 1024) == 0: # 每下载 1MB 打印一次 print(f"📥 已下载: {downloaded_bytes // (1024 * 1024)} MB") print(f"✅ 下载完成！文件已存至: {target_path}") success_msg = f"✅ 下载完成！文件已存至: {target_path}" return ToolResult( title="直链下载成功", output=success_msg, long_term_memory=success_msg, metadata={"path": str(target_path)} ) except Exception as e: # 异常捕获返回 return ToolResult( title="下载异常", output="", error=f"💥 发生错误: {str(e)}", long_term_memory=f"下载任务由于异常中断: {str(e)}" ) @tool() async def browser_click_element(index: int) -> ToolResult: """ 点击页面元素，并自动通过拦截内部日志获取下载直链。 """ # 1. 挂载日志窃听器 capture_handler = DownloadLinkCaptureHandler() logger = logging.getLogger("browser_use") # 拦截整个 browser_use 命名空间 logger.addHandler(capture_handler) try: browser, tools = await get_browser_session() # 2. 执行原生的点击动作 result = await tools.click( index=index, browser_session=browser ) # 3. 检查是否有“意外收获” download_msg = "" if capture_handler.captured_url: captured_url = capture_handler.captured_url download_msg = f"\n\n⚠️ 系统检测到浏览器下载被拦截，已自动捕获准确直链：\n{captured_url}\n\n建议：你可以直接使用 browser_download_direct_url 工具下载此链接。" # 如果你想更激进一点，甚至可以在这里直接自动触发本地下载逻辑 # await auto_download_file(captured_url) # 4. 转换结果并附加捕获的信息 tool_result = action_result_to_tool_result(result, f"点击元素 {index}") if download_msg: # 关键：把日志里的信息塞进 output，这样 LLM 就能看到了！ tool_result.output = (tool_result.output or "") + download_msg tool_result.long_term_memory = (tool_result.long_term_memory or "") + f" 捕获下载链接: {captured_url}" return tool_result except Exception as e: return ToolResult( title="点击失败", output="", error=f"Failed to click element {index}: {str(e)}", long_term_memory=f"点击元素 {index} 失败" ) finally: # 5. 务必移除监听器，防止内存泄漏和日志污染 logger.removeHandler(capture_handler) @tool() async def browser_input_text(index: int, text: str, clear: bool = True) -> ToolResult: """ 在指定元素中输入文本 Input text into an element Args: index: 元素索引（从浏览器状态中获取） text: 要输入的文本内容 clear: 是否先清除现有文本（默认 True） Returns: ToolResult: 包含输入操作结果的工具返回对象 Example: input_text(index=0, text="Hello World", clear=True) """ try: browser, tools = await get_browser_session() result = await tools.input( index=index, text=text, clear=clear, browser_session=browser ) return action_result_to_tool_result(result, f"输入文本到元素 {index}") except Exception as e: return ToolResult( title="输入失败", output="", error=f"Failed to input text into element {index}: {str(e)}", long_term_memory=f"输入文本失败" ) @tool() async def browser_send_keys(keys: str) -> ToolResult: """ 发送键盘按键或快捷键 Send keyboard keys or shortcuts 支持发送单个按键、组合键和快捷键。 Args: keys: 要发送的按键字符串 - 单个按键: "Enter", "Escape", "PageDown", "Tab" - 组合键: "Control+o", "Shift+Tab", "Alt+F4" - 功能键: "F1", "F2", ..., "F12" Returns: ToolResult: 包含按键操作结果的工具返回对象 Example: send_keys("Enter") send_keys("Control+A") """ try: browser, tools = await get_browser_session() result = await tools.send_keys( keys=keys, browser_session=browser ) return action_result_to_tool_result(result, f"发送按键: {keys}") except Exception as e: return ToolResult( title="发送按键失败", output="", error=f"Failed to send keys: {str(e)}", long_term_memory="发送按键失败" ) @tool() async def browser_upload_file(index: int, path: str) -> ToolResult: """ 上传文件到文件输入元素 Upload a file to a file input element Args: index: 文件输入框的元素索引 path: 要上传的文件路径（绝对路径） Returns: ToolResult: 包含上传操作结果的工具返回对象 Example: upload_file(index=7, path="/path/to/file.pdf") Note: 文件必须存在且路径必须是绝对路径 """ try: browser, tools = await get_browser_session() result = await tools.upload_file( index=index, path=path, browser_session=browser, available_file_paths=[path], file_system=_file_system ) return action_result_to_tool_result(result, f"上传文件: {path}") except Exception as e: return ToolResult( title="上传失败", output="", error=f"Failed to upload file: {str(e)}", long_term_memory=f"上传文件 {path} 失败" ) # ============================================================ # 滚动和视图工具 (Scroll & View Tools) # ============================================================ @tool() async def browser_scroll_page(down: bool = True, pages: float = 1.0, index: Optional[int] = None) -> ToolResult: try: # 限制单次滚动幅度，避免 agent 一次滚 100 页 MAX_PAGES = 10 if pages > MAX_PAGES: pages = MAX_PAGES browser, tools = await get_browser_session() cdp_session = await browser.get_or_create_cdp_session() before_y_result = await cdp_session.cdp_client.send.Runtime.evaluate( params={'expression': 'window.scrollY'}, session_id=cdp_session.session_id ) before_y = before_y_result.get('result', {}).get('value', 0) # 执行滚动 result = await tools.scroll(down=down, pages=pages, index=index, browser_session=browser) # 等待渲染（懒加载页面需要更长时间） await asyncio.sleep(2) after_y_result = await cdp_session.cdp_client.send.Runtime.evaluate( params={'expression': 'window.scrollY'}, session_id=cdp_session.session_id ) after_y = after_y_result.get('result', {}).get('value', 0) # 如果第一次检测没动，再等一轮（应对懒加载触发后的延迟滚动） if before_y == after_y and index is None: await asyncio.sleep(2) retry_result = await cdp_session.cdp_client.send.Runtime.evaluate( params={'expression': 'window.scrollY'}, session_id=cdp_session.session_id ) after_y = retry_result.get('result', {}).get('value', 0) if before_y == after_y and index is None: direction = "下" if down else "上" return ToolResult( title="滚动无效", output=f"页面已到达{direction}边界，无法继续滚动", error="No movement detected" ) delta = abs(after_y - before_y) direction = "下" if down else "上" return action_result_to_tool_result(result, f"已向{direction}滚动 {delta}px") except Exception as e: # --- 核心修复 2: 必须补全 output 参数，否则框架会报错 --- return ToolResult( title="滚动失败", output="", # 补全这个缺失的必填参数 error=str(e) ) @tool() async def browser_find_text(text: str) -> ToolResult: """ 查找页面中的文本并滚动到该位置 Find text on the page and scroll to it 在页面中搜索指定的文本，找到后自动滚动到该位置。 Args: text: 要查找的文本内容 Returns: ToolResult: 包含查找结果的工具返回对象 Example: find_text("Privacy Policy") """ try: browser, tools = await get_browser_session() result = await tools.find_text( text=text, browser_session=browser ) return action_result_to_tool_result(result, f"查找文本: {text}") except Exception as e: return ToolResult( title="查找失败", output="", error=f"Failed to find text: {str(e)}", long_term_memory=f"查找文本 '{text}' 失败" ) @tool() async def browser_get_visual_selector_map() -> ToolResult: """ 获取当前页面的视觉快照和交互元素索引映射。 Get visual snapshot and selector map of interactive elements. 该工具会同时执行两个操作： 1. 捕捉当前页面的截图，并用 browser-use 内置方法在截图上标注元素索引号。 2. 生成页面所有可交互元素的索引字典（含 href、type 等属性信息）。 Returns: ToolResult: 包含高亮截图（在 images 中）和元素列表的工具返回对象。 """ try: browser, _ = await get_browser_session() # 1. 构造同时包含 DOM 和 截图 的请求 from browser_use.browser.events import BrowserStateRequestEvent from browser_use.browser.python_highlights import create_highlighted_screenshot_async event = browser.event_bus.dispatch( BrowserStateRequestEvent( include_dom=True, include_screenshot=True, include_recent_events=False ) ) # 2. 等待浏览器返回完整状态 browser_state = await event.event_result(raise_if_none=True, raise_if_any=True) # 3. 提取 Selector Map selector_map = browser_state.dom_state.selector_map if browser_state.dom_state else {} # 4. 提取截图并生成带索引标注的高亮截图（通过 CDP 获取精确 DPI 和滚动偏移） screenshot_b64 = browser_state.screenshot or "" highlighted_b64 = "" if screenshot_b64 and selector_map: try: cdp_session = await browser.get_or_create_cdp_session() highlighted_b64 = await create_highlighted_screenshot_async( screenshot_b64, selector_map, cdp_session=cdp_session, filter_highlight_ids=False ) except Exception: highlighted_b64 = screenshot_b64 # fallback to raw screenshot else: highlighted_b64 = screenshot_b64 # 5. 构建供 Agent 阅读的完整元素列表，包含丰富的属性信息 elements_info = [] for index, node in selector_map.items(): tag = node.tag_name attrs = node.attributes or {} desc = attrs.get('aria-label') or attrs.get('placeholder') or attrs.get('title') or node.get_all_children_text(max_depth=1) or "" # 收集有用的属性片段 extra_parts = [] if attrs.get('href'): extra_parts.append(f"href={attrs['href'][:60]}") if attrs.get('type'): extra_parts.append(f"type={attrs['type']}") if attrs.get('role'): extra_parts.append(f"role={attrs['role']}") if attrs.get('name'): extra_parts.append(f"name={attrs['name']}") extra = f" ({', '.join(extra_parts)})" if extra_parts else "" elements_info.append(f"Index {index}: <{tag}> \"{desc[:50]}\"{extra}") output = f"页面截图已捕获（含元素索引标注）\n找到 {len(selector_map)} 个交互元素\n\n" output += "元素列表:\n" + "\n".join(elements_info) # 6. 将高亮截图存入 images 字段，metadata 保留结构化数据 images = [] if highlighted_b64: images.append({"type": "base64", "media_type": "image/png", "data": highlighted_b64}) return ToolResult( title="视觉元素观察", output=output, long_term_memory=f"在页面观察到 {len(selector_map)} 个元素并保存了截图", images=images, metadata={ "selector_map": {k: str(v) for k, v in list(selector_map.items())[:100]}, "url": browser_state.url, "title": browser_state.title } ) except Exception as e: return ToolResult( title="视觉观察失败", output="", error=f"Failed to get visual selector map: {str(e)}", long_term_memory="获取视觉元素映射失败" ) @tool() async def browser_screenshot() -> ToolResult: """ 请求在下次观察中包含页面截图 Request a screenshot to be included in the next observation 用于视觉检查页面状态，帮助理解页面布局和内容。 Returns: ToolResult: 包含截图请求结果的工具返回对象 Example: screenshot() Note: 截图会在下次页面观察时自动包含在结果中。 """ try: browser, tools = await get_browser_session() result = await tools.screenshot(browser_session=browser) return action_result_to_tool_result(result, "截图请求") except Exception as e: return ToolResult( title="截图失败", output="", error=f"Failed to capture screenshot: {str(e)}", long_term_memory="截图失败" ) # ============================================================ # 标签页管理工具 (Tab Management Tools) # ============================================================ @tool() async def browser_switch_tab(tab_id: str) -> ToolResult: """ 切换到指定标签页 Switch to a different browser tab Args: tab_id: 4字符标签ID（target_id 的最后4位） Returns: ToolResult: 切换结果 Example: switch_tab(tab_id="a3f2") """ try: browser, tools = await get_browser_session() normalized_tab_id = tab_id[-4:] if tab_id else tab_id result = await tools.switch( tab_id=normalized_tab_id, browser_session=browser ) return action_result_to_tool_result(result, f"切换到标签页 {normalized_tab_id}") except Exception as e: return ToolResult( title="切换标签页失败", output="", error=f"Failed to switch tab: {str(e)}", long_term_memory=f"切换到标签页 {tab_id} 失败" ) @tool() async def browser_close_tab(tab_id: str) -> ToolResult: """ 关闭指定标签页 Close a browser tab Args: tab_id: 4字符标签ID Returns: ToolResult: 关闭结果 Example: close_tab(tab_id="a3f2") """ try: browser, tools = await get_browser_session() normalized_tab_id = tab_id[-4:] if tab_id else tab_id result = await tools.close( tab_id=normalized_tab_id, browser_session=browser ) return action_result_to_tool_result(result, f"关闭标签页 {normalized_tab_id}") except Exception as e: return ToolResult( title="关闭标签页失败", output="", error=f"Failed to close tab: {str(e)}", long_term_memory=f"关闭标签页 {tab_id} 失败" ) # ============================================================ # 下拉框工具 (Dropdown Tools) # ============================================================ @tool() async def browser_get_dropdown_options(index: int) -> ToolResult: """ 获取下拉框的所有选项 Get options from a dropdown element Args: index: 下拉框的元素索引 Returns: ToolResult: 包含所有选项的结果 Example: get_dropdown_options(index=8) """ try: browser, tools = await get_browser_session() result = await tools.dropdown_options( index=index, browser_session=browser ) return action_result_to_tool_result(result, f"获取下拉框选项: {index}") except Exception as e: return ToolResult( title="获取下拉框选项失败", output="", error=f"Failed to get dropdown options: {str(e)}", long_term_memory=f"获取下拉框 {index} 选项失败" ) @tool() async def browser_select_dropdown_option(index: int, text: str) -> ToolResult: """ 选择下拉框选项 Select an option from a dropdown Args: index: 下拉框的元素索引 text: 要选择的选项文本（精确匹配） Returns: ToolResult: 选择结果 Example: select_dropdown_option(index=8, text="Option 2") """ try: browser, tools = await get_browser_session() result = await tools.select_dropdown( index=index, text=text, browser_session=browser ) return action_result_to_tool_result(result, f"选择下拉框选项: {text}") except Exception as e: return ToolResult( title="选择下拉框选项失败", output="", error=f"Failed to select dropdown option: {str(e)}", long_term_memory=f"选择选项 '{text}' 失败" ) # ============================================================ # 内容提取工具 (Content Extraction Tools) # ============================================================ def scrub_search_redirect_url(url: str) -> str: """ 自动检测并解析 Bing/Google 等搜索引擎的重定向链接，提取真实目标 URL。 """ if not url or not isinstance(url, str): return url try: parsed = urlparse(url) # 1. 处理 Bing 重定向 (特征：u 参数带 Base64) # 示例：...&u=a1aHR0cHM6Ly96aHVhbmxhbi56aGlodS5jb20vcC8zODYxMjgwOQ&... if "bing.com" in parsed.netloc: u_param = parse_qs(parsed.query).get('u', [None])[0] if u_param: # 移除开头的 'a1', 'a0' 等标识符 b64_str = u_param[2:] # 补齐 Base64 填充符 padding = '=' * (4 - len(b64_str) % 4) decoded = base64.b64decode(b64_str + padding).decode('utf-8', errors='ignore') if decoded.startswith('http'): return decoded # 2. 处理 Google 重定向 (特征：url 参数) if "google.com" in parsed.netloc: url_param = parse_qs(parsed.query).get('url', [None])[0] if url_param: return unquote(url_param) # 3. 兜底：处理常见的跳转参数 for param in ['target', 'dest', 'destination', 'link']: found = parse_qs(parsed.query).get(param, [None])[0] if found and found.startswith('http'): return unquote(found) except Exception: pass # 解析失败则返回原链接 return url async def extraction_adapter(input_data): # 提取字符串 if isinstance(input_data, list): prompt = input_data[-1].content if hasattr(input_data[-1], 'content') else str(input_data[-1]) else: prompt = str(input_data) response = await openrouter_llm_call( messages=[{"role": "user", "content": prompt}] ) content = response["content"] # --- 核心改进：URL 自动修复 --- # 使用正则表达式匹配内容中的所有 URL，并尝试进行洗涤 urls = re.findall(r'https?://[^\s<>"\']+', content) for original_url in urls: clean_url = scrub_search_redirect_url(original_url) if clean_url != original_url: content = content.replace(original_url, clean_url) from argparse import Namespace return Namespace(completion=content) @tool() async def browser_extract_content(query: str, extract_links: bool = False, start_from_char: int = 0) -> ToolResult: """ 使用 LLM 从页面提取结构化数据 Extract content from the current page using LLM Args: query: 提取查询（告诉 LLM 要提取什么内容） extract_links: 是否提取链接（默认 False，节省 token） start_from_char: 从哪个字符开始提取（用于分页提取大内容） Returns: ToolResult: 提取的内容 Example: extract_content(query="提取页面上所有产品的名称和价格", extract_links=True) Note: 需要配置 page_extraction_llm，否则会失败 支持分页提取，最大100k字符 """ try: browser, tools = await get_browser_session() # 注意：extract 需要 page_extraction_llm 参数 # 这里我们假设用户会在初始化时配置 LLM # 如果没有配置，会抛出异常 result = await tools.extract( query=query, extract_links=extract_links, start_from_char=start_from_char, browser_session=browser, page_extraction_llm=RunnableLambda(extraction_adapter), # 需要用户配置 file_system=_file_system ) return action_result_to_tool_result(result, f"提取内容: {query}") except Exception as e: return ToolResult( title="内容提取失败", output="", error=f"Failed to extract content: {str(e)}", long_term_memory=f"提取内容失败: {query}" ) async def _detect_and_download_pdf_via_cdp(browser) -> Optional[str]: """ 检测当前页面是否为 PDF，如果是则通过 CDP（浏览器内 fetch）下载到本地。 优势：自动携带浏览器的 cookies/session，可访问需要登录的 PDF。 返回本地文件路径，非 PDF 页面返回 None。 """ try: current_url = await browser.get_current_page_url() if not current_url: return None parsed = urlparse(current_url) is_pdf = parsed.path.lower().endswith('.pdf') # URL 不明显是 PDF 时，通过 CDP 检查 content-type if not is_pdf: try: cdp = await browser.get_or_create_cdp_session() ct_result = await cdp.cdp_client.send.Runtime.evaluate( params={'expression': 'document.contentType'}, session_id=cdp.session_id ) content_type = ct_result.get('result', {}).get('value', '') is_pdf = 'pdf' in content_type.lower() except Exception: pass if not is_pdf: return None # 通过浏览器内 fetch API 下载 PDF（自动携带 cookies） cdp = await browser.get_or_create_cdp_session() js_code = """ (async () => { try { const resp = await fetch(window.location.href); if (!resp.ok) return JSON.stringify({error: 'HTTP ' + resp.status}); const blob = await resp.blob(); return new Promise((resolve, reject) => { const reader = new FileReader(); reader.onloadend = () => resolve(JSON.stringify({data: reader.result})); reader.onerror = () => resolve(JSON.stringify({error: 'FileReader failed'})); reader.readAsDataURL(blob); }); } catch(e) { return JSON.stringify({error: e.message}); } })() """ result = await cdp.cdp_client.send.Runtime.evaluate( params={ 'expression': js_code, 'awaitPromise': True, 'returnByValue': True, 'timeout': 60000 }, session_id=cdp.session_id ) value = result.get('result', {}).get('value', '') if not value: print("⚠️ CDP fetch PDF: 无返回值") return None data = json.loads(value) if 'error' in data: print(f"⚠️ CDP fetch PDF 失败: {data['error']}") return None # 从 data URL 中提取 base64 并解码 data_url = data['data'] # data:application/pdf;base64,JVBERi0... base64_data = data_url.split(',', 1)[1] pdf_bytes = base64.b64decode(base64_data) # 保存到本地 save_dir = Path.cwd() / ".cache/.browser_use_files" save_dir.mkdir(parents=True, exist_ok=True) filename = Path(parsed.path).name if parsed.path else "" if not filename or not filename.lower().endswith('.pdf'): import time filename = f"downloaded_{int(time.time())}.pdf" save_path = str(save_dir / filename) with open(save_path, 'wb') as f: f.write(pdf_bytes) print(f"📄 PDF 已通过 CDP 下载到: {save_path} ({len(pdf_bytes)} bytes)") return save_path except Exception as e: print(f"⚠️ PDF 检测/下载异常: {e}") return None @tool() async def browser_read_long_content( goal: Union[str, dict], source: str = "page", context: str = "", **kwargs ) -> ToolResult: """ 智能读取长内容。支持自动检测并读取网页上的 PDF 文件。 当 source="page" 且当前页面是 PDF 时，会通过 CDP 下载 PDF 并用 pypdf 解析， 而非使用 DOM 提取（DOM 无法读取浏览器内置 PDF Viewer 的内容）。 通过 CDP 下载可自动携带浏览器的 cookies/session，支持需要登录的 PDF。 """ try: browser, tools = await get_browser_session() # 1. 提取目标文本 (针对 GoalTree 字典结构) final_goal_text = "" if isinstance(goal, dict): final_goal_text = goal.get("mission") or goal.get("goal") or str(goal) else: final_goal_text = str(goal) # 2. 清洗业务背景 (过滤框架注入的 dict 类型 context) business_context = context if isinstance(context, str) else "" # 3. PDF 自动检测：当 source="page" 时检查是否为 PDF 页面 available_files = [] if source.lower() == "page": pdf_path = await _detect_and_download_pdf_via_cdp(browser) if pdf_path: source = pdf_path available_files.append(pdf_path) # 4. 验证并实例化 action_params = ReadContentAction( goal=final_goal_text, source=source, context=business_context ) # 5. 解包参数调用底层方法 result = await tools.read_long_content( **action_params.model_dump(), browser_session=browser, page_extraction_llm=RunnableLambda(extraction_adapter), available_file_paths=available_files ) return action_result_to_tool_result(result, f"深度读取: {source}") except Exception as e: return ToolResult( title="深度读取失败", output="", error=f"Read long content failed: {str(e)}", long_term_memory="参数解析或校验失败，请检查输入" ) @tool() async def browser_get_page_html() -> ToolResult: """ 获取当前页面的完整 HTML Get the full HTML of the current page 返回当前页面的完整 HTML 源代码。 Returns: ToolResult: 包含页面 HTML 的工具返回对象 Example: get_page_html() Note: - 返回的是完整的 HTML 源代码 - 输出会被限制在 10000 字符以内（完整内容保存在 metadata 中） """ try: browser, tools = await get_browser_session() # 使用 CDP 获取页面 HTML cdp = await browser.get_or_create_cdp_session() # 获取页面内容 result = await cdp.cdp_client.send.Runtime.evaluate( params={'expression': 'document.documentElement.outerHTML'}, session_id=cdp.session_id ) html = result.get('result', {}).get('value', '') # 获取 URL 和标题 url = await browser.get_current_page_url() title_result = await cdp.cdp_client.send.Runtime.evaluate( params={'expression': 'document.title'}, session_id=cdp.session_id ) title = title_result.get('result', {}).get('value', '') # 限制输出大小 output_html = html if len(html) > 10000: output_html = html[:10000] + "... (truncated)" return ToolResult( title=f"获取 HTML: {url}", output=f"页面: {title}\nURL: {url}\n\nHTML:\n{output_html}", long_term_memory=f"获取 HTML: {url}", metadata={"url": url, "title": title, "html": html} ) except Exception as e: return ToolResult( title="获取 HTML 失败", output="", error=f"Failed to get page HTML: {str(e)}", long_term_memory="获取 HTML 失败" ) @tool() async def browser_get_selector_map() -> ToolResult: """ 获取当前页面的元素索引映射 Get the selector map of interactive elements on the current page 返回页面所有可交互元素的索引字典，用于后续的元素操作。 Returns: ToolResult: 包含元素映射的工具返回对象 Example: get_selector_map() Note: 返回的索引可以用于 click_element, input_text 等操作 """ try: browser, tools = await get_browser_session() # 关键修复：先触发 BrowserStateRequestEvent 来更新 DOM 状态 # 这会触发 DOM watchdog 重新构建 DOM 树并更新 selector_map from browser_use.browser.events import BrowserStateRequestEvent # 触发事件并等待结果 event = browser.event_bus.dispatch( BrowserStateRequestEvent( include_dom=True, include_screenshot=False, # 不需要截图，节省时间 include_recent_events=False ) ) # 等待 DOM 更新完成 browser_state = await event.event_result(raise_if_none=True, raise_if_any=True) # 从更新后的状态中获取 selector_map selector_map = browser_state.dom_state.selector_map if browser_state.dom_state else {} # 构建输出信息 elements_info = [] for index, node in list(selector_map.items())[:20]: # 只显示前20个 tag = node.tag_name attrs = node.attributes or {} text = attrs.get('aria-label') or attrs.get('placeholder') or attrs.get('value', '') elements_info.append(f"索引 {index}: <{tag}> {text[:50]}") output = f"找到 {len(selector_map)} 个交互元素\n\n" output += "\n".join(elements_info) if len(selector_map) > 20: output += f"\n... 还有 {len(selector_map) - 20} 个元素" return ToolResult( title="获取元素映射", output=output, long_term_memory=f"获取到 {len(selector_map)} 个交互元素", metadata={"selector_map": {k: str(v) for k, v in list(selector_map.items())[:100]}} ) except Exception as e: return ToolResult( title="获取元素映射失败", output="", error=f"Failed to get selector map: {str(e)}", long_term_memory="获取元素映射失败" ) # ============================================================ # JavaScript 执行工具 (JavaScript Tools) # ============================================================ @tool() async def browser_evaluate(code: str) -> ToolResult: """ 在页面中执行 JavaScript 代码 Execute JavaScript code in the page context 允许在当前页面中执行任意 JavaScript 代码，用于复杂的页面操作或数据提取。 Args: code: 要执行的 JavaScript 代码字符串 Returns: ToolResult: 包含执行结果的工具返回对象 Example: evaluate("document.title") evaluate("document.querySelectorAll('a').length") Note: - 代码在页面上下文中执行，可以访问 DOM 和全局变量 - 返回值会被自动序列化为字符串 - 执行结果限制在 20k 字符以内 """ try: browser, tools = await get_browser_session() result = await tools.evaluate( code=code, browser_session=browser ) return action_result_to_tool_result(result, "执行 JavaScript") except Exception as e: return ToolResult( title="JavaScript 执行失败", output="", error=f"Failed to execute JavaScript: {str(e)}", long_term_memory="JavaScript 执行失败" ) @tool() async def browser_ensure_login_with_cookies(cookie_type: str, url: str = "https://www.xiaohongshu.com") -> ToolResult: """ 检查登录状态并在需要时注入 cookies """ try: browser, tools = await get_browser_session() if url: await tools.navigate(url=url, browser_session=browser) await tools.wait(seconds=2, browser_session=browser) check_login_js = """ (function() { const loginBtn = document.querySelector('[class*="login"]') || document.querySelector('[href*="login"]') || Array.from(document.querySelectorAll('button, a')).find(el => (el.textContent || '').includes('登录')); const userInfo = document.querySelector('[class*="user"]') || document.querySelector('[class*="avatar"]'); return { needLogin: !!loginBtn && !userInfo, hasLoginBtn: !!loginBtn, hasUserInfo: !!userInfo }; })() """ result = await tools.evaluate(code=check_login_js, browser_session=browser) status_output = result.extracted_content if isinstance(status_output, str) and status_output.startswith("Result: "): status_output = status_output[8:] login_info: Dict[str, Any] = {} if isinstance(status_output, str): try: login_info = json.loads(status_output) except Exception: login_info = {} elif isinstance(status_output, dict): login_info = status_output if not login_info.get("needLogin"): output = json.dumps({"need_login": False}, ensure_ascii=False) return ToolResult( title="已登录", output=output, long_term_memory=output ) row = _fetch_cookie_row(cookie_type) cookie_value = _extract_cookie_value(row) if not cookie_value: output = json.dumps({"need_login": True, "cookies_count": 0}, ensure_ascii=False) return ToolResult( title="未找到 cookies", output=output, error="未找到 cookies", long_term_memory=output ) domain, base_url = _cookie_domain_for_type(cookie_type, url) cookies = _normalize_cookies(cookie_value, domain, base_url) if not cookies: output = json.dumps({"need_login": True, "cookies_count": 0}, ensure_ascii=False) return ToolResult( title="cookies 解析失败", output=output, error="cookies 解析失败", long_term_memory=output ) await browser._cdp_set_cookies(cookies) if url: await tools.navigate(url=url, browser_session=browser) await tools.wait(seconds=2, browser_session=browser) output = json.dumps({"need_login": True, "cookies_count": len(cookies)}, ensure_ascii=False) return ToolResult( title="已注入 cookies", output=output, long_term_memory=output ) except Exception as e: return ToolResult( title="登录检查失败", output="", error=str(e), long_term_memory="登录检查失败" ) # ============================================================ # 等待用户操作工具 (Wait for User Action) # ============================================================ @tool() async def browser_wait_for_user_action(message: str = "Please complete the action in browser", timeout: int = 300) -> ToolResult: """ 等待用户在浏览器中完成操作（如登录） Wait for user to complete an action in the browser (e.g., login) 暂停自动化流程，等待用户手动完成某些操作（如登录、验证码等）。 Args: message: 提示用户需要完成的操作 timeout: 最大等待时间（秒），默认 300 秒（5 分钟） Returns: ToolResult: 包含等待结果的工具返回对象 Example: wait_for_user_action("Please login to Xiaohongshu", timeout=180) wait_for_user_action("Please complete the CAPTCHA", timeout=60) Note: - 用户需要在浏览器窗口中手动完成操作 - 完成后按回车键继续 - 超时后会自动继续执行 """ try: import asyncio print(f"\n{'='*60}") print(f"⏸️ WAITING FOR USER ACTION") print(f"{'='*60}") print(f"📝 {message}") print(f"⏱️ Timeout: {timeout} seconds") print(f"\n👉 Please complete the action in the browser window") print(f"👉 Press ENTER when done, or wait for timeout") print(f"{'='*60}\n") # Wait for user input or timeout try: loop = asyncio.get_event_loop() # Wait for either user input or timeout await asyncio.wait_for( loop.run_in_executor(None, input), timeout=timeout ) return ToolResult( title="用户操作完成", output=f"User completed: {message}", long_term_memory=f"用户完成操作: {message}" ) except asyncio.TimeoutError: return ToolResult( title="用户操作超时", output=f"Timeout waiting for: {message}", long_term_memory=f"等待用户操作超时: {message}" ) except Exception as e: return ToolResult( title="等待用户操作失败", output="", error=f"Failed to wait for user action: {str(e)}", long_term_memory="等待用户操作失败" ) # ============================================================ # 任务完成工具 (Task Completion) # ============================================================ @tool() async def browser_done(text: str, success: bool = True, files_to_display: Optional[List[str]] = None) -> ToolResult: """ 标记任务完成并返回最终消息 Mark the task as complete and return final message to user Args: text: 给用户的最终消息 success: 任务是否成功完成 files_to_display: 可选的要显示的文件路径列表 Returns: ToolResult: 完成结果 Example: done("任务已完成，提取了10个产品信息", success=True) """ try: browser, tools = await get_browser_session() result = await tools.done( text=text, success=success, files_to_display=files_to_display, file_system=_file_system ) return action_result_to_tool_result(result, "任务完成") except Exception as e: return ToolResult( title="标记任务完成失败", output="", error=f"Failed to complete task: {str(e)}", long_term_memory="标记任务完成失败" ) # ============================================================ # Cookie 持久化工具 # ============================================================ _COOKIES_DIR = Path(__file__).parent.parent.parent.parent.parent / ".cache/.cookies" @tool() async def browser_export_cookies(name: str = "", account: str = "") -> ToolResult: """ 导出当前浏览器的所有 Cookie 到本地 .cookies/ 目录。 文件命名格式：{域名}_{账号名}.json，如 bilibili.com_zhangsan.json 登录成功后调用此工具，下次可通过 browser_load_cookies 恢复登录态。 Args: name: 自定义文件名（可选，提供则忽略自动命名） account: 账号名称（可选，用于区分同一网站的不同账号） """ try: browser, _ = await get_browser_session() # 获取所有 Cookie（CDP 格式） all_cookies = await browser._cdp_get_cookies() if not all_cookies: return ToolResult(title="Cookie 导出", output="当前浏览器没有 Cookie", long_term_memory="无 Cookie 可导出") # 获取当前域名，用于过滤和命名 from urllib.parse import urlparse current_url = await browser.get_current_page_url() or '' domain = urlparse(current_url).netloc.replace("www.", "") or "default" if not name: name = f"{domain}_{account}" if account else domain # 只保留当前域名的 cookie（过滤第三方） cookies = [c for c in all_cookies if domain in c.get("domain", "").lstrip(".")] # 保存 _COOKIES_DIR.mkdir(parents=True, exist_ok=True) cookie_file = _COOKIES_DIR / f"{name}.json" cookie_file.write_text(json.dumps(cookies, ensure_ascii=False, indent=2), encoding="utf-8") return ToolResult( title="Cookie 已导出", output=f"已保存 {len(cookies)} 条 Cookie 到 .cookies/{name}.json（从 {len(all_cookies)} 条中过滤当前域名）", long_term_memory=f"导出 {len(cookies)} 条 Cookie 到 .cookies/{name}.json" ) except Exception as e: return ToolResult(title="Cookie 导出失败", output="", error=str(e), long_term_memory="导出 Cookie 失败") @tool() async def browser_load_cookies(url: str, name: str = "") -> ToolResult: """ 根据目标 URL 自动查找本地 Cookie 文件，注入浏览器并导航到目标页面恢复登录态。 重要：此工具会自动完成导航，调用前不需要先调用 browser_navigate_to_url。 Args: url: 目标 URL（必须提供，同时用于自动匹配 Cookie 文件） name: Cookie 文件名（可选，不传则根据 URL 域名自动查找） """ try: browser, tools = await get_browser_session() if not url.startswith("http"): url = f"https://{url}" # 根据域名自动查找 Cookie 文件 if not name: from urllib.parse import urlparse domain = urlparse(url).netloc.replace("www.", "") if _COOKIES_DIR.exists(): matches = list(_COOKIES_DIR.glob(f"{domain}*.json")) if matches: cookie_file = matches[0] # 取第一个匹配的 else: available = [f.stem for f in _COOKIES_DIR.glob("*.json")] return ToolResult(title="未找到 Cookie", output=f"没有匹配 {domain} 的文件，可用: {available}", error=f"无 {domain} 的 Cookie 文件") else: return ToolResult(title="未找到 Cookie", output=".cookies 目录不存在", error="Cookie 目录不存在") else: cookie_file = _COOKIES_DIR / f"{name}.json" if not cookie_file.exists(): available = [f.stem for f in _COOKIES_DIR.glob("*.json")] if _COOKIES_DIR.exists() else [] return ToolResult(title="文件不存在", output=f"可用: {available}", error=f"未找到 .cookies/{name}.json") cookies = json.loads(cookie_file.read_text(encoding="utf-8")) # 直接注入（export 和 load 使用相同的 CDP 格式，无需标准化） await browser._cdp_set_cookies(cookies) # 导航到目标页面（带上刚注入的 Cookie） if url: if not url.startswith("http"): url = f"https://{url}" await tools.navigate(url=url, browser_session=browser) await tools.wait(seconds=3, browser_session=browser) return ToolResult( title="Cookie 注入并导航完成", output=f"从 {cookie_file.name} 注入 {len(cookies)} 条 Cookie，已导航到 {url}", long_term_memory=f"已从 {cookie_file.name} 注入 Cookie 并导航到 {url}，登录态已恢复" ) except Exception as e: return ToolResult(title="Cookie 加载失败", output="", error=str(e), long_term_memory="加载 Cookie 失败") # ============================================================ # 导出所有工具函数（供外部使用） # ============================================================ __all__ = [ # 会话管理 'init_browser_session', 'get_browser_session', 'cleanup_browser_session', 'kill_browser_session', # 导航类工具 'browser_navigate_to_url', 'browser_search_web', 'browser_go_back', 'browser_wait', # 元素交互工具 'browser_click_element', 'browser_input_text', 'browser_send_keys', 'browser_upload_file', # 滚动和视图工具 'browser_scroll_page', 'browser_find_text', 'browser_screenshot', # 标签页管理工具 'browser_switch_tab', 'browser_close_tab', # 下拉框工具 'browser_get_dropdown_options', 'browser_select_dropdown_option', # 内容提取工具 'browser_extract_content', 'browser_get_page_html', 'browser_read_long_content', 'browser_download_direct_url', 'browser_get_selector_map', 'browser_get_visual_selector_map', # JavaScript 执行工具 'browser_evaluate', 'browser_ensure_login_with_cookies', # 等待用户操作 'browser_wait_for_user_action', # 任务完成 'browser_done', # Cookie 持久化 'browser_export_cookies', 'browser_load_cookies', ]