快速上手：LangChain + AgentRun 浏览器沙箱极简集成指南

from agentrun.sandbox import Sandbox, TemplateTypefrom playwright.sync_api import sync_playwright# 创建 Browser Sandboxsandbox = Sandbox.create(    template_type=TemplateType.BROWSER,    template_name="your-template-name",    sandbox_idle_timeout_seconds=300)# 获取 CDP URL（用于 Playwright 连接）cdp_url = sandbox.get_cdp_url()# 使用 Playwright 连接并操作with sync_playwright() as p:    browser = p.chromium.connect_over_cdp(cdp_url)    page = browser.contexts[0].pages[0]    page.goto("https://www.example.com")    page.screenshot(path="screenshot.png")    browser.close()# 销毁 Sandboxsandbox.delete()

mkdir -p langchain-democd langchain-demo

# LangChain 核心库langchain>=0.1.0langchain-openai>=0.0.5langchain-community>=0.0.20# AgentRun SDKagentrun-sdk[playwright,server]>=0.0.8# 浏览器自动化playwright>=1.40.0# 环境变量管理python-dotenv>=1.0.0

pip install -r requirements.txt

# 阿里云百炼平台的 API Key，用于调用大模型能力# 请前往 https://bailian.console.aliyun.com/?tab=app#/api-key 创建和查看DASHSCOPE_API_KEY=sk-your-bailian-api-key# 阿里云账号的访问密钥 ID 和访问密钥 Secret，用于 AgentRun SDK 鉴权ALIBABA_CLOUD_ACCESS_KEY_ID=your-akALIBABA_CLOUD_ACCESS_KEY_SECRET=your-skALIBABA_CLOUD_ACCOUNT_ID=your-main-account-idALIBABA_CLOUD_REGION=cn-hangzhou# browser sandbox 模板的名称，可以在 https://functionai.console.aliyun.com/cn-hangzhou/agent/runtime/sandbox 控制台创建BROWSER_TEMPLATE_NAME=sandbox-your-template-name# agentrun 的控制面和数据面的 API 端点请求地址，默认cn-hangzhouAGENTRUN_CONTROL_ENDPOINT=agentrun.cn-hangzhou.aliyuncs.comAGENTRUN_DATA_ENDPOINT=https://${your-main-account-id}.agentrun-data.cn-hangzhou.aliyuncs.com

"""Sandbox 生命周期管理模块负责 AgentRun Browser Sandbox 的创建、管理和销毁。提供统一的接口供 LangChain Agent 使用。"""import osfrom typing import Optional, Dict, Anyfrom dotenv import load_dotenv# 加载环境变量load_dotenv()class SandboxManager:    """Sandbox 生命周期管理器"""    def __init__(self):        self._sandbox: Optional[Any] = None        self._sandbox_id: Optional[str] = None        self._cdp_url: Optional[str] = None        self._vnc_url: Optional[str] = None    def create(        self,        template_name: Optional[str] = None,        idle_timeout: int = 3000    ) -> Dict[str, Any]:        """        创建或获取一个浏览器 sandbox 实例        Args:            template_name: Sandbox 模板名称，如果为 None 则从环境变量读取            idle_timeout: 空闲超时时间（秒），默认 3000 秒        Returns:            dict: 包含 sandbox_id, cdp_url, vnc_url 的字典        Raises:            RuntimeError: 创建失败时抛出异常        """        try:            from agentrun.sandbox import Sandbox, TemplateType            # 如果已有 sandbox，直接返回            if self._sandbox is not None:                return self.get_info()            # 从环境变量获取模板名称            if template_name is None:                template_name = os.getenv(                    "BROWSER_TEMPLATE_NAME",                    "sandbox-browser-demo"                )            # 创建 sandbox            self._sandbox = Sandbox.create(                template_type=TemplateType.BROWSER,                template_name=template_name,                sandbox_idle_timeout_seconds=idle_timeout            )            self._sandbox_id = self._sandbox.sandbox_id            self._cdp_url = self._get_cdp_url()            self._vnc_url = self._get_vnc_url()            return self.get_info()        except ImportError as e:            print(e)            raise RuntimeError(                "agentrun-sdk 未安装，请运行: pip install agentrun-sdk[playwright,server]"            )        except Exception as e:            raise RuntimeError(f"创建 Sandbox 失败: {str(e)}")    def get_info(self) -> Dict[str, Any]:        """        获取当前 sandbox 的信息        Returns:            dict: 包含 sandbox_id, cdp_url, vnc_url 的字典        Raises:            RuntimeError: 如果没有活动的 sandbox        """        if self._sandbox is None:            raise RuntimeError("没有活动的 sandbox，请先创建")        return {            "sandbox_id": self._sandbox_id,            "cdp_url": self._cdp_url,            "vnc_url": self._vnc_url,        }    def get_cdp_url(self) -> Optional[str]:        """获取 CDP URL"""        return self._sandbox.get_cdp_url()    def get_vnc_url(self) -> Optional[str]:        """获取 VNC URL"""        return self._sandbox.get_vnc_url()    def get_sandbox_id(self) -> Optional[str]:        """获取 Sandbox ID"""        return self._sandbox_id    def destroy(self) -> str:        """        销毁当前的 sandbox 实例        Returns:            str: 操作结果描述        """        if self._sandbox is None:            return "没有活动的 sandbox"        try:            sandbox_id = self._sandbox_id            # 尝试销毁 sandbox            if hasattr(self._sandbox, 'delete'):                self._sandbox.delete()            elif hasattr(self._sandbox, 'stop'):                self._sandbox.stop()            elif hasattr(self._sandbox, 'destroy'):                self._sandbox.destroy()            # 清理状态            self._sandbox = None            self._sandbox_id = None            self._cdp_url = None            self._vnc_url = None            return f"Sandbox 已销毁: {sandbox_id}"        except Exception as e:            # 即使销毁失败，也清理本地状态            self._sandbox = None            self._sandbox_id = None            self._cdp_url = None            self._vnc_url = None            return f"销毁 Sandbox 时出错: {str(e)}"    def is_active(self) -> bool:        """检查 sandbox 是否活跃"""        return self._sandbox is not None    def __enter__(self):        """上下文管理器入口"""        return self    def __exit__(self, exc_type, exc_val, exc_tb):        """上下文管理器退出，自动销毁"""        self.destroy()        return False# 全局单例（可选，用于简单场景）_global_manager: Optional[SandboxManager] = Nonedef get_global_manager() -> SandboxManager:    """获取全局 SandboxManager 单例"""    global _global_manager    if _global_manager is None:        _global_manager = SandboxManager()    return _global_managerdef reset_global_manager():    """重置全局 SandboxManager"""    global _global_manager    if _global_manager:        _global_manager.destroy()    _global_manager = None

"""LangChain Agent 和 Tools 注册模块负责创建 LangChain Agent，注册 Sandbox 相关的 tools，并集成 VNC 可视化。本模块使用 sandbox_manager.py 中封装的 SandboxManager 来管理 sandbox 生命周期。"""import osfrom dotenv import load_dotenvfrom langchain.tools import toolfrom langchain_openai import ChatOpenAIfrom langchain.agents import create_agentfrom pydantic import BaseModel, Field# 导入 sandbox 管理器from sandbox_manager import SandboxManager# 加载环境变量load_dotenv()# 全局 sandbox 管理器实例（单例模式）_sandbox_manager: SandboxManager | None = Nonedef get_sandbox_manager() -> SandboxManager:    """获取 sandbox 管理器实例（单例模式）"""    global _sandbox_manager    if _sandbox_manager is None:        _sandbox_manager = SandboxManager()    return _sandbox_manager# ============ LangChain Tools 定义 ============@tooldef create_browser_sandbox(    template_name: str = None,    idle_timeout: int = 3000) -> str:    """创建或获取一个浏览器 sandbox 实例。    当需要访问网页、执行浏览器操作时，首先需要创建 sandbox。    创建成功后，会返回 sandbox 信息，包括 VNC URL 用于可视化。    Args:        template_name: Sandbox 模板名称，如果不提供则从环境变量 BROWSER_TEMPLATE_NAME 读取        idle_timeout: 空闲超时时间（秒），默认 3000 秒    Returns:        Sandbox 信息字符串，包括 ID、CDP URL、VNC URL    """    try:        manager = get_sandbox_manager()        # 如果 template_name 为空字符串，转换为 None 以便从环境变量读取        if template_name == "":            template_name = None        info = manager.create(template_name=template_name, idle_timeout=idle_timeout)        result = f"""✅ Sandbox 创建成功！📋 Sandbox 信息:- ID: {info['sandbox_id']}- CDP URL: {info['cdp_url']}"""        vnc_url = info.get('vnc_url')        if vnc_url:            result += f"- VNC URL: {vnc_url}\n\n"            result += "提示: VNC 查看器应该已自动打开，您可以在浏览器中实时查看浏览器操作。"        else:            result += "\n警告: 未获取到 VNC URL，可能无法使用可视化功能。"        return result    except Exception as e:        return f" 创建 Sandbox 失败: {str(e)}"@tooldef get_sandbox_info() -> str:    """获取当前 sandbox 的详细信息，包括 ID、CDP URL、VNC URL 等。    当需要查看当前 sandbox 状态或获取 VNC 连接信息时使用此工具。    Returns:        Sandbox 信息字符串    """    try:        manager = get_sandbox_manager()        info = manager.get_info()        result = f"""📋 当前 Sandbox 信息:- Sandbox ID: {info['sandbox_id']}- CDP URL: {info['cdp_url']}"""        if info.get('vnc_url'):            result += f"- VNC URL: {info['vnc_url']}\n\n"            result += "您可以使用 VNC URL 在浏览器中实时查看操作过程。\n"            result += "   推荐使用 vnc.html 文件或 noVNC 客户端。"        return result    except RuntimeError as e:        return f" {str(e)}"    except Exception as e:        return f" 获取 Sandbox 信息失败: {str(e)}"class NavigateInput(BaseModel):    """浏览器导航输入参数"""    url: str = Field(description="要访问的网页 URL，必须以 http:// 或 https:// 开头")    wait_until: str = Field(        default="load",        description="等待页面加载的状态: load, domcontentloaded, networkidle"    )    timeout: int = Field(        default=30000,        description="超时时间（毫秒），默认 30000"    )@tool(args_schema=NavigateInput)def navigate_to_url(url: str, wait_until: str = "load", timeout: int = 30000) -> str:    """使用 sandbox 中的浏览器导航到指定 URL。    当用户需要访问网页时使用此工具。导航后可以在 VNC 中实时查看页面。    Args:        url: 要访问的网页 URL        wait_until: 等待页面加载的状态（load/domcontentloaded/networkidle）        timeout: 超时时间（毫秒）    Returns:        导航结果描述    """    try:        manager = get_sandbox_manager()        if not manager.is_active():            return " 错误: 请先创建 sandbox"        # 验证 URL        if not url.startswith(("http://", "https://")):            return f" 错误: 无效的 URL 格式: {url}"        cdp_url = manager.get_cdp_url()        if not cdp_url:            return " 错误: 无法获取 CDP URL"        # 使用 Playwright 连接浏览器并导航        try:            from playwright.sync_api import sync_playwright            with sync_playwright() as p:                browser = p.chromium.connect_over_cdp(cdp_url)                pages = browser.contexts[0].pages if browser.contexts else []                if pages:                    page = pages[0]                else:                    page = browser.new_page()                page.goto(url, wait_until=wait_until, timeout=timeout)                title = page.title()                return f"已成功导航到: {url}\n📄 页面标题: {title}\n💡 您可以在 VNC 中查看页面内容。"        except ImportError:            return f"导航指令已发送: {url}\n💡 提示: 安装 playwright 以启用实际导航功能 (pip install playwright)"        except Exception as e:            return f" 导航失败: {str(e)}"    except Exception as e:        return f" 操作失败: {str(e)}"@tool("browser_screenshot", description="在浏览器 sandbox 中截取当前页面截图")def take_screenshot(filename: str = "screenshot.png") -> str:    """截取浏览器当前页面的截图。    Args:        filename: 截图文件名，默认 "screenshot.png"    Returns:        操作结果    """    try:        manager = get_sandbox_manager()        if not manager.is_active():            return " 错误: 请先创建 sandbox"        cdp_url = manager.get_cdp_url()        if not cdp_url:            return " 错误: 无法获取 CDP URL"        try:            from playwright.sync_api import sync_playwright            with sync_playwright() as p:                browser = p.chromium.connect_over_cdp(cdp_url)                pages = browser.contexts[0].pages if browser.contexts else []                if pages:                    page = pages[0]                else:                    return " 错误: 没有打开的页面"                page.screenshot(path=filename)                return f"截图已保存: {filename}"        except ImportError:            return " 错误: 需要安装 playwright (pip install playwright)"        except Exception as e:            return f" 截图失败: {str(e)}"    except Exception as e:        return f" 操作失败: {str(e)}"@tool("destroy_sandbox", description="销毁当前的 sandbox 实例，释放资源。注意：仅在程序退出或明确需要释放资源时使用，不要在一轮对话后销毁。")def destroy_sandbox() -> str:    """销毁当前的 sandbox 实例。    重要提示：此工具应该仅在以下情况使用：    - 程序即将退出    - 明确需要释放资源    - 用户明确要求销毁    不要在一轮对话完成后就销毁 sandbox，因为 sandbox 可以在多轮对话中复用。    Returns:        操作结果    """    try:        manager = get_sandbox_manager()        result = manager.destroy()        return result    except Exception as e:        return f" 销毁失败: {str(e)}"# ============ Agent 创建 ============def create_browser_agent(system_prompt: str = None):    """    创建带有 sandbox 工具的 LangChain Agent    Args:        system_prompt: 自定义系统提示词，如果为 None 则使用默认提示词    Returns:        LangChain Agent 实例    """    # 配置 DashScope API    api_key = os.getenv("DASHSCOPE_API_KEY")    if not api_key:        raise ValueError("请设置环境变量 DASHSCOPE_API_KEY")    base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1"    model_name = os.getenv("QWEN_MODEL", "qwen-plus")    # 创建 LLM    model = ChatOpenAI(        model=model_name,        api_key=api_key,        base_url=base_url,        temperature=0.7,    )    # 创建工具列表    tools = [        create_browser_sandbox,        get_sandbox_info,        navigate_to_url,        take_screenshot,        destroy_sandbox,    ]    # 默认系统提示词    if system_prompt is None:        system_prompt = """你是一个浏览器自动化助手，可以使用 sandbox 来访问和操作网页。当用户需要访问网页时，请按以下步骤操作：1. 首先创建或获取 sandbox（如果还没有）2. 使用 navigate_to_url 导航到目标网页3. 执行用户请求的操作4. 如果需要，可以截取截图重要提示：- 创建 sandbox 后，会返回 VNC URL，用户可以使用它实时查看浏览器操作- 所有操作都会在 VNC 中实时显示，方便调试和监控- sandbox 可以在多轮对话中复用，不要在一轮对话完成后就销毁- 只有在用户明确要求销毁时才使用 destroy_sandbox 工具- 不要主动建议用户销毁 sandbox，除非用户明确要求- 请始终用中文回复，确保操作准确、高效。"""    # 创建 Agent    agent = create_agent(        model=model,        tools=tools,        system_prompt=system_prompt,    )    return agentdef get_available_tools():    """获取所有可用的工具列表"""    return [        create_browser_sandbox,        get_sandbox_info,        navigate_to_url,        take_screenshot,        destroy_sandbox,    ]

"""LangChain + AgentRun Browser Sandbox 集成示例主入口文件，演示如何使用 LangChain Agent 与 AgentRun Browser Sandbox 集成。"""import osimport sysimport signalimport webbrowserimport urllib.parseimport threadingimport http.serverimport socketserverfrom pathlib import Pathfrom dotenv import load_dotenvfrom langchain_agent import create_browser_agent, get_sandbox_manager# 加载环境变量load_dotenv()# 全局 HTTP 服务器实例_http_server = None_http_port = 8080# 全局清理标志，用于防止重复清理_cleanup_done = Falsedef start_http_server():    """启动一个简单的 HTTP 服务器来提供 vnc.html"""    global _http_server    if _http_server is not None:        return _http_port    try:        current_dir = Path(__file__).parent.absolute()        class VNCRequestHandler(http.server.SimpleHTTPRequestHandler):            def __init__(self, *args, **kwargs):                super().__init__(*args, directory=str(current_dir), **kwargs)            def log_message(self, format, *args):                # 静默日志，避免输出过多信息                pass        # 尝试启动服务器        for port in range(_http_port, _http_port + 10):            try:                server = socketserver.TCPServer(("", port), VNCRequestHandler)                server.allow_reuse_address = True                # 在后台线程中运行服务器                def run_server():                    server.serve_forever()                thread = threading.Thread(target=run_server, daemon=True)                thread.start()                _http_server = server                return port            except OSError:                continue        return None    except Exception as e:        print(f"启动 HTTP 服务器失败: {str(e)}")        return Nonedef open_vnc_viewer(vnc_url: str):    """    自动打开 VNC 查看器并设置 VNC URL    Args:        vnc_url: VNC WebSocket URL    """    if not vnc_url:        return    try:        # 获取当前文件所在目录        current_dir = Path(__file__).parent.absolute()        vnc_html_path = current_dir / "vnc.html"        # 检查文件是否存在        if not vnc_html_path.exists():            print(f"警告: vnc.html 文件不存在: {vnc_html_path}")            print_vnc_info(vnc_url)            return        # 启动 HTTP 服务器        port = start_http_server()        if port:            # 编码 VNC URL 作为 URL 参数            encoded_url = urllib.parse.quote(vnc_url, safe='')            # 构建 HTTP URL            http_url = f"http://localhost:{port}/vnc.html?url={encoded_url}"            # 打开浏览器            print(f"\n正在打开 VNC 查看器...")            print(f"HTTP 服务器运行在: http://localhost:{port}")            print(f"VNC URL: {vnc_url[:80]}...")            print(f"完整 URL: {http_url[:100]}...")            webbrowser.open(http_url)            print(f"VNC 查看器已打开")            print(f"VNC URL 已通过 URL 参数自动设置，页面加载后会自动连接")        else:            # 如果 HTTP 服务器启动失败，尝试使用 file:// 协议            print(f"HTTP 服务器启动失败，尝试使用文件协议...")            encoded_url = urllib.parse.quote(vnc_url, safe='')            file_url = f"file://{vnc_html_path}?url={encoded_url}"            webbrowser.open(file_url)            print(f"VNC 查看器已打开（使用文件协议）")            print(f"提示: 如果无法自动连接，请手动复制 VNC URL 到输入框")    except Exception as e:        print(f"自动打开 VNC 查看器失败: {str(e)}")        print_vnc_info(vnc_url)def print_vnc_info(vnc_url: str):    """打印 VNC 连接信息"""    if not vnc_url:        return    print("\n" + "=" * 60)    print("VNC 可视化连接信息")    print("=" * 60)    print(f"\nVNC URL: {vnc_url}")    print("\n使用方式:")    print("   1. 使用 noVNC 客户端连接")    print("   2. 或在浏览器中访问 VNC 查看器页面")    print("   3. 实时查看浏览器操作过程")    print("\n" + "=" * 60 + "\n")def cleanup_sandbox():    """    清理 sandbox 资源    这个函数可以被信号处理器、异常处理器和正常退出流程调用    """    global _cleanup_done    # 防止重复清理    if _cleanup_done:        return    _cleanup_done = True    try:        manager = get_sandbox_manager()        if manager.is_active():            print("\n" + "=" * 60)            print("正在清理 sandbox...")            print("=" * 60)            result = manager.destroy()            print(f"清理结果: {result}\n")        else:            print("\n没有活动的 sandbox 需要清理\n")    except Exception as e:        print(f"\n清理 sandbox 时出错: {str(e)}\n")def signal_handler(signum, frame):    """    信号处理器，处理 Ctrl+C (SIGINT) 和其他信号    Args:        signum: 信号编号        frame: 当前堆栈帧    """    print("\n\n收到中断信号，正在清理资源...")    cleanup_sandbox()    print("清理完成")    sys.exit(0)def main():    """主函数"""    global _cleanup_done    # 重置清理标志    _cleanup_done = False    # 注册信号处理器，处理 Ctrl+C (SIGINT)    signal.signal(signal.SIGINT, signal_handler)    # 在 Windows 上，SIGBREAK 也可以处理    if hasattr(signal, 'SIGBREAK'):        signal.signal(signal.SIGBREAK, signal_handler)    print("=" * 60)    print("LangChain + AgentRun Browser Sandbox 集成示例")    print("=" * 60)    print()    try:        # 创建 Agent        print("正在初始化 LangChain Agent...")        agent = create_browser_agent()        print("Agent 初始化完成\n")        # 示例查询        queries = [            "创建一个浏览器 sandbox",            "获取当前 sandbox 的信息，包括 VNC URL",            "导航到 https://www.aliyun.com",            "截取当前页面截图",        ]        # 执行查询        for i, query in enumerate(queries, 1):            print(f"\n{'=' * 60}")            print(f"查询 {i}: {query}")            print(f"{'=' * 60}\n")            try:                result = agent.invoke({                    "messages": [{"role": "user", "content": query}]                })                # 提取最后一条消息的内容                output = result.get("messages", [])[-1].content if isinstance(result.get("messages"), list) else result.get("output", str(result))                print(f"\n结果:\n{output}\n")                # 如果是创建 sandbox，自动打开 VNC 查看器                if i == 1:                    try:                        # 等待一下确保 sandbox 完全创建                        import time                        time.sleep(1)                        manager = get_sandbox_manager()                        if manager.is_active():                            info = manager.get_info()                            vnc_url = info.get('vnc_url')                            if vnc_url:                                print(f"\n检测到 VNC URL: {vnc_url[:80]}...")                                open_vnc_viewer(vnc_url)                                print_vnc_info(vnc_url)                            else:                                print("\n警告: 未获取到 VNC URL，请检查 sandbox 创建是否成功")                    except Exception as e:                        print(f"打开 VNC 查看器时出错: {str(e)}")                        import traceback                        traceback.print_exc()                # 如果是获取信息，显示 VNC 信息                elif i == 2:                    try:                        manager = get_sandbox_manager()                        if manager.is_active():                            info = manager.get_info()                            if info.get('vnc_url'):                                print_vnc_info(info['vnc_url'])                    except:                        pass            except Exception as e:                print(f"查询失败: {str(e)}\n")                import traceback                traceback.print_exc()        # 交互式查询        print("\n" + "=" * 60)        print("进入交互模式（输入 'quit' 或 'exit' 退出，Ctrl+C 或 Ctrl+D 中断）")        print("=" * 60 + "\n")        while True:            try:                user_input = input("请输入您的查询: ").strip()            except EOFError:                # 处理 Ctrl+D (EOF)                print("\n\n检测到输入结束 (Ctrl+D)，正在清理资源...")                cleanup_sandbox()                print("清理完成")                break            except KeyboardInterrupt:                # 处理 Ctrl+C (在 input 调用期间)                print("\n\n检测到中断信号 (Ctrl+C)，正在清理资源...")                cleanup_sandbox()                print("清理完成")                break            if not user_input:                continue            if user_input.lower() in ['quit', 'exit', '退出']:                print("\nBye")                # 退出前清理 sandbox                cleanup_sandbox()                break            try:                result = agent.invoke({                    "messages": [{"role": "user", "content": user_input}]                })                output = result.get("messages", [])[-1].content if isinstance(result.get("messages"), list) else result.get("output", str(result))                print(f"\n结果:\n{output}\n")                # 检查是否需要打开或显示 VNC 信息                user_input_lower = user_input.lower()                if "创建" in user_input_lower and "sandbox" in user_input_lower:                    # 如果是创建 sandbox，自动打开 VNC 查看器                    try:                        # 等待一下确保 sandbox 完全创建                        import time                        time.sleep(1)                        manager = get_sandbox_manager()                        if manager.is_active():                            info = manager.get_info()                            vnc_url = info.get('vnc_url')                            if vnc_url:                                print(f"\n检测到 VNC URL: {vnc_url[:80]}...")                                open_vnc_viewer(vnc_url)                                print_vnc_info(vnc_url)                            else:                                print("\n警告: 未获取到 VNC URL，请检查 sandbox 创建是否成功")                    except Exception as e:                        print(f"打开 VNC 查看器时出错: {str(e)}")                        import traceback                        traceback.print_exc()                elif "sandbox" in user_input_lower or "vnc" in user_input_lower:                    # 其他情况只显示信息                    try:                        manager = get_sandbox_manager()                        if manager.is_active():                            info = manager.get_info()                            if info.get('vnc_url'):                                print_vnc_info(info['vnc_url'])                    except:                        pass            except Exception as e:                print(f"查询失败: {str(e)}\n")                import traceback                traceback.print_exc()        # 清理资源（仅在程序正常退出时）        cleanup_sandbox()    except KeyboardInterrupt:        # 处理顶层 KeyboardInterrupt (Ctrl+C)        print("\n\n检测到中断信号 (Ctrl+C)，正在清理资源...")        cleanup_sandbox()        print("清理完成")        sys.exit(0)    except EOFError:        # 处理顶层 EOFError (Ctrl+D)        print("\n\n检测到输入结束 (Ctrl+D)，正在清理资源...")        cleanup_sandbox()        print("清理完成")        sys.exit(0)    except ValueError as e:        print(f"配置错误: {str(e)}")        print("\n提示: 请确保已设置以下环境变量:")        print("   - DASHSCOPE_API_KEY: DashScope API Key")        print("   - ALIBABA_CLOUD_ACCOUNT_ID: 阿里云账号 ID")        print("   - ALIBABA_CLOUD_ACCESS_KEY_ID: 访问密钥 ID")        print("   - ALIBABA_CLOUD_ACCESS_KEY_SECRET: 访问密钥 Secret")        print("   - ALIBABA_CLOUD_REGION: 区域（默认: cn-hangzhou）")    except Exception as e:        print(f"发生错误: {str(e)}")        import traceback        traceback.print_exc()        # 发生错误时也尝试清理        cleanup_sandbox()if __name__ == "__main__":    main()

# 通过 Agent 调用result = agent.invoke({    "messages": [{"role": "user", "content": "获取 sandbox 信息"}]})# 或直接通过管理器获取manager = get_sandbox_manager()info = manager.get_info()vnc_url = info['vnc_url']

import webbrowserimport urllib.parsefrom pathlib import Pathdef open_vnc_viewer(vnc_url: str):    """自动打开 VNC 查看器"""    current_dir = Path(__file__).parent.absolute()    vnc_html_path = current_dir / "vnc.html"    if vnc_html_path.exists():        # 通过 URL 参数传递 VNC URL        encoded_url = urllib.parse.quote(vnc_url, safe='')        file_url = f"file://{vnc_html_path}?url={encoded_url}"        webbrowser.open(file_url)

// 从 URL 参数获取 VNC URLconst urlParams = new URLSearchParams(window.location.search);const vncUrl = urlParams.get('url');// 加载 noVNC 库async function loadNoVNC() {    const module = await import('https://cdn.jsdelivr.net/gh/novnc/noVNC@v1.4.0/core/rfb.js');    return module.default;}// 连接 VNCasync function connectVNC(url) {    const RFB = await loadNoVNC();    rfb = new RFB(vncScreen, url, {        shared: true,        credentials: { password: '' }    });    rfb.addEventListener('connect', () => {        console.log('VNC 连接成功');    });}

python main.py

用户请求 → Agent → Tool → SandboxManager → AgentRun SDK → 云端 Sandbox                                    ↓用户响应 ← Agent ← Tool ← SandboxManager ← 操作结果

# 第 1 轮对话用户: "创建一个 sandbox 并访问淘宝首页"→ Agent 调用: create_browser_sandbox → navigate_to_url→ Manager: 创建 Sandbox → Playwright 导航→ 结果: "Sandbox 已创建，已访问淘宝首页"# 第 2 轮对话（复用 Sandbox）用户: "截取当前页面"→ Agent 调用: take_screenshot→ Manager: 使用现有 Sandbox → Playwright 截图→ 结果: "截图已保存"# 第 3 轮对话（复用 Sandbox）用户: "访问京东首页"→ Agent 调用: navigate_to_url→ Manager: 使用现有 Sandbox → Playwright 导航→ 结果: "已访问京东首页"

@tooldef extract_table_data(url: str) -> str:    """从网页中提取表格数据"""    from playwright.sync_api import sync_playwright    manager = get_sandbox_manager()    cdp_url = manager.get_info()['cdp_url']    with sync_playwright() as p:        browser = p.chromium.connect_over_cdp(cdp_url)        page = browser.contexts[0].pages[0]        page.goto(url)        tables = page.query_selector_all("table")        return f"找到 {len(tables)} 个表格"

custom_prompt = """你是一个专业的网页数据提取助手。在执行任务前，请先创建 sandbox，然后使用浏览器工具完成任务。"""agent = create_browser_agent(system_prompt=custom_prompt)

<!DOCTYPE html><html><head>    <title>Browser Sandbox VNC 查看器</title>    <style>        body { margin: 0; padding: 0; background: #000; }        #vnc-container { width: 100vw; height: 100vh; }    </style></head><body>    <div id="vnc-container"></div>    <script type="module">        const params = new URLSearchParams(window.location.search);        const vncUrl = params.get('url');        if (!vncUrl) {            alert('请提供 VNC URL 参数');        } else {            const module = await import('https://cdn.jsdelivr.net/gh/novnc/noVNC@v1.4.0/core/rfb.js');            const RFB = module.default;            const rfb = new RFB(                document.getElementById('vnc-container'),                vncUrl,                { shared: true, credentials: { password: '' } }            );            rfb.scaleViewport = true;        }    </script></body></html>

import webbrowserimport urllib.parsevnc_url = sandbox.vnc_urlencoded_url = urllib.parse.quote(vnc_url, safe='')viewer_url = f"file:///path/to/vnc-viewer.html?url={encoded_url}"webbrowser.open(viewer_url)

import React, { useEffect, useRef } from 'react';interface VNCViewerProps {  vncUrl: string;  onConnect?: () => void;  onDisconnect?: () => void;}export const VNCViewer: React.FC<VNCViewerProps> = ({   vncUrl,   onConnect,   onDisconnect }) => {  const containerRef = useRef<HTMLDivElement>(null);  useEffect(() => {    let rfb: any;    const initVNC = async () => {      if (!containerRef.current || !vncUrl) return;      const { default: RFB } = await import('@novnc/novnc/core/rfb');      rfb = new RFB(containerRef.current, vncUrl, {        shared: true,        credentials: { password: '' }      });      rfb.scaleViewport = true;      rfb.addEventListener('connect', () => onConnect?.());      rfb.addEventListener('disconnect', () => onDisconnect?.());    };    initVNC();    return () => {      if (rfb) rfb.disconnect();    };  }, [vncUrl, onConnect, onDisconnect]);  return (    <div       ref={containerRef}       style={{ width: '100%', height: '600px', background: '#000' }}     />  );};

import React, { useState, useEffect } from 'react';import { VNCViewer } from './VNCViewer';function App() {  const [vncUrl, setVncUrl] = useState<string>('');  useEffect(() => {    fetch('/api/sandbox/create', { method: 'POST' })      .then(res => res.json())      .then(data => setVncUrl(data.vnc_url));  }, []);  return (    <div>      <h1>Browser Sandbox 实时监控</h1>      {vncUrl ? (        <VNCViewer           vncUrl={vncUrl}          onConnect={() => console.log('已连接')}          onDisconnect={() => console.log('已断开')}        />      ) : (        <p>正在初始化...</p>      )}    </div>  );}

from playwright.sync_api import sync_playwrightfrom agentrun.sandbox import Sandbox, TemplateType# 创建 Sandboxsandbox = Sandbox.create(    template_type=TemplateType.BROWSER,    template_name="your-template-name",    sandbox_idle_timeout_seconds=3000)# 使用 Playwright 连接with sync_playwright() as p:    browser = p.chromium.connect_over_cdp(sandbox.cdp_url)    page = browser.contexts[0].pages[0]    # 执行操作    page.goto("https://www.example.com")    page.screenshot(path="screenshot.png")    content = page.content()    browser.close()# 清理sandbox.delete()

const puppeteer = require('puppeteer-core');// CDP URL 从 Sandbox 获取const cdpUrl = 'wss://your-account.funagent-data-pre.cn-hangzhou.aliyuncs.com/sandboxes/xxx/ws/automation';(async () => {  const browser = await puppeteer.connect({    browserWSEndpoint: cdpUrl,    defaultViewport: null  });  const page = (await browser.pages())[0];  await page.goto('https://www.example.com');  await page.screenshot({ path: 'screenshot.png' });  await browser.close();})();