浏览器自动化示例

本指南展示如何使用 AIO Sandbox 进行浏览器自动化、Web 抓取和 UI 测试。

概述

AIO Sandbox 提供多种与浏览器交互的方式:

  • VNC 访问:通过远程桌面进行可视化浏览器交互
  • Chrome DevTools 协议(CDP):程序化浏览器控制
  • 浏览器 MCP 服务器:高级浏览器自动化工具

VNC 浏览器访问

基本 VNC 交互

通过 VNC 可视化访问浏览器:

# 打开 VNC 界面
open http://localhost:8080/vnc/index.html?autoconnect=true

VNC 界面提供:

  • 带浏览器的完整桌面环境
  • 可视化交互功能
  • 截图和屏幕录制
  • 鼠标和键盘输入

自动化脚本中的 VNC

在需要视觉验证的场景中使用 VNC:

import requests
import time

# 通过 VNC API 截图
response = requests.get("http://localhost:8080/vnc/screenshot")
with open("screenshot.png", "wb") as f:
    f.write(response.content)

# 发送键盘输入
requests.post("http://localhost:8080/vnc/keyboard",
              json={"keys": "Hello World"})

Chrome DevTools 协议(CDP)

CDP 连接

使用 CDP 连接到浏览器:

const CDP = require('chrome-remote-interface');

async function example() {
    const client = await CDP();
    const {Network, Page, Runtime} = client;

    // 启用必要的域
    await Network.enable();
    await Page.enable();

    // 导航到页面
    await Page.navigate({url: 'https://example.com'});
    await Page.loadEventFired();

    // 执行 JavaScript
    const result = await Runtime.evaluate({
        expression: 'document.title'
    });

    console.log('页面标题:', result.result.value);

    await client.close();
}

example().catch(console.error);

Python CDP 示例

使用 Python 和 pychrome:

import pychrome
import time

# 连接到浏览器
browser = pychrome.Browser(url="http://localhost:8080")
tab = browser.new_tab()

# 启用页面域
tab.Page.enable()

# 导航到页面
tab.Page.navigate(url="https://httpbin.org/get")
tab.wait(1)

# 获取页面内容
result = tab.Runtime.evaluate(expression="document.body.innerText")
print("页面内容:", result['result']['value'])

# 清理
browser.close_tab(tab)

浏览器 MCP 服务器 API 参考

@agent-infra/mcp-server-browser 包提供以下工具:

导航和基本操作

  • browser_navigate - 导航到 URL
  • browser_go_back - 返回上一页
  • browser_go_forward - 前进到下一页
  • browser_close - 关闭浏览器

元素交互

  • browser_get_clickable_elements - 获取可点击/可悬停/可选择的元素
  • browser_click - 点击元素(使用元素索引)
  • browser_hover - 悬停在元素上
  • browser_select - 选择元素
  • browser_form_input_fill - 填充输入字段

内容检索

  • browser_get_text - 获取页面文本内容
  • browser_get_markdown - 获取页面的 Markdown 格式
  • browser_read_links - 获取所有页面链接
  • browser_screenshot - 截图

高级功能

  • browser_scroll - 滚动页面
  • browser_evaluate - 执行 JavaScript
  • browser_new_tab - 打开新标签页
  • browser_switch_tab - 在标签页之间切换
  • browser_tab_list - 列出所有标签页
  • browser_get_download_list - 获取下载的文件

视觉模式(可选)

  • browser_vision_screen_capture - 视觉模式截图
  • browser_vision_screen_click - 基于视觉的点击

浏览器 MCP 服务器集成

Python 与 MCP 客户端

使用浏览器 MCP 服务器进行高级自动化:

import asyncio
from mcp import Client
import httpx

async def browser_automation():
    async with httpx.AsyncClient() as http_client:
        # 连接到 MCP 服务器
        async with Client("http://localhost:8080/mcp") as client:

            # 导航到页面
            result = await client.call_tool("browser_navigate", {
                "url": "https://example.com"
            })

            # 截图
            screenshot = await client.call_tool("browser_screenshot", {
                "path": "/tmp/screenshot.png"
            })

            # 获取页面文本内容
            content = await client.call_tool("browser_get_text")

            # 获取可点击元素
            elements = await client.call_tool("browser_get_clickable_elements")

            print("页面内容:", content.content[0].text)
            print("可点击元素:", elements.content)

asyncio.run(browser_automation())

JavaScript MCP 集成

const { McpClient } = require('@modelcontextprotocol/sdk');

async function browserAutomation() {
    const client = new McpClient("http://localhost:8080/mcp");

    try {
        await client.connect();

        // 导航到页面
        await client.callTool("browser_navigate", {
            url: "https://news.ycombinator.com"
        });

        // 等待页面加载
        await new Promise(resolve => setTimeout(resolve, 2000));

        // 获取可点击元素(如文章标题)
        const elements = await client.callTool("browser_get_clickable_elements");

        // 获取页面 Markdown 内容
        const markdown = await client.callTool("browser_get_markdown");

        console.log("可点击元素:", elements.content);
        console.log("页面内容:", markdown.content);

    } finally {
        await client.disconnect();
    }
}

browserAutomation().catch(console.error);

Web 抓取示例

电商价格监控

import asyncio
import json
from datetime import datetime

async def monitor_prices():
    # 要监控的产品 URL
    products = [
        {"name": "笔记本电脑", "url": "https://example-store.com/laptop"},
        {"name": "手机", "url": "https://example-store.com/phone"}
    ]

    results = []

    for product in products:
        # 使用浏览器导航并提取价格
        navigation_result = await client.call_tool("browser_navigate", {
            "url": product["url"]
        })

        # 获取可点击元素并提取价格
        elements = await client.call_tool("browser_get_clickable_elements")
        price_text = await client.call_tool("browser_get_text")

        results.append({
            "product": product["name"],
            "price": price_text.content[0].text,
            "timestamp": datetime.now().isoformat(),
            "url": product["url"]
        })

    # 通过文件 API 保存结果到文件
    await client.call_tool("file_write", {
        "path": "/tmp/price_data.json",
        "content": json.dumps(results, indent=2)
    })

    return results

# 运行价格监控
prices = asyncio.run(monitor_prices())
print("价格监控完成:", prices)

社交媒体内容提取

async def extract_social_posts():
    # 导航到社交媒体页面
    await client.call_tool("browser_navigate", {
        "url": "https://example-social.com/trending"
    })

    # 等待动态内容加载
    await asyncio.sleep(3)

    # 获取页面内容和可点击元素
    content = await client.call_tool("browser_get_text")
    elements = await client.call_tool("browser_get_clickable_elements")

    # 处理提取的数据
    return {
        "page_content": content.content[0].text,
        "clickable_elements": elements.content
    }

浏览器测试示例

表单自动化测试

async def test_contact_form():
    # 导航到表单页面
    await client.call_tool("browser_navigate", {
        "url": "https://example.com/contact"
    })

    # 首先获取表单元素
    elements = await client.call_tool("browser_get_clickable_elements")

    # 使用 form_input_fill 填充表单字段
    await client.call_tool("browser_form_input_fill", {
        "selector": "input[name='name']",
        "value": "测试用户"
    })

    await client.call_tool("browser_form_input_fill", {
        "selector": "input[name='email']",
        "value": "test@example.com"
    })

    await client.call_tool("browser_form_input_fill", {
        "selector": "textarea[name='message']",
        "value": "这是一条测试消息"
    })

    # 点击提交按钮提交表单
    await client.call_tool("browser_click", {
        "index": 0  # 使用元素中提交按钮的索引
    })

    # 等待响应
    await asyncio.sleep(2)

    # 通过获取页面文本验证成功消息
    page_text = await client.call_tool("browser_get_text")

    assert "谢谢" in page_text.content[0].text
    print("表单提交测试通过!")

性能测试

async def measure_page_performance():
    start_time = time.time()

    # 导航到页面
    await client.call_tool("browser_navigate", {
        "url": "https://example-app.com"
    })

    # 等待页面加载(使用 sleep 或评估 JS)
    await asyncio.sleep(3)

    load_time = time.time() - start_time

    # 使用 JavaScript 评估获取性能指标
    metrics = await client.call_tool("browser_evaluate", {
        "expression": "JSON.stringify({loadTime: performance.timing.loadEventEnd - performance.timing.navigationStart, domContentLoaded: performance.timing.domContentLoadedEventEnd - performance.timing.navigationStart})"
    })

    return {
        "total_load_time": load_time,
        "performance_metrics": metrics.result
    }

Playwright 集成

对于高级浏览器自动化,集成 Playwright:

from playwright.async_api import async_playwright
import asyncio

async def playwright_example():
    async with async_playwright() as p:
        # 连接到 AIO Sandbox 中的现有浏览器
        browser = await p.chromium.connect_over_cdp(
            "ws://localhost:8080/cdp"
        )

        page = await browser.new_page()

        # 导航和交互
        await page.goto("https://example.com")
        await page.fill("input[name='search']", "测试查询")
        await page.click("button[type='submit']")

        # 等待结果
        await page.wait_for_selector(".search-results")

        # 提取数据
        results = await page.query_selector_all(".result-item")
        for result in results:
            title = await result.inner_text()
            print(f"结果:{title}")

        await browser.close()

asyncio.run(playwright_example())

Selenium 集成

使用 Selenium WebDriver 与 AIO Sandbox:

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options

def selenium_example():
    # 配置 Chrome 选项以进行远程连接
    chrome_options = Options()
    chrome_options.add_experimental_option("debuggerAddress", "localhost:9222")

    # 连接到浏览器
    driver = webdriver.Chrome(options=chrome_options)

    try:
        # 导航和交互
        driver.get("https://example.com")

        # 查找和交互元素
        search_box = driver.find_element(By.NAME, "search")
        search_box.send_keys("测试查询")

        submit_button = driver.find_element(By.CSS_SELECTOR, "button[type='submit']")
        submit_button.click()

        # 等待并提取结果
        driver.implicitly_wait(10)
        results = driver.find_elements(By.CSS_SELECTOR, ".result-item")

        for result in results:
            print(f"结果:{result.text}")

    finally:
        driver.quit()

selenium_example()

最佳实践

资源管理

async def managed_browser_session():
    try:
        # 导航到起始页面
        await client.call_tool("browser_navigate", {
            "url": "https://example.com"
        })

        # 执行浏览器操作
        yield client

    finally:
        # 完成后关闭浏览器
        await client.call_tool("browser_close")

错误处理

async def robust_browser_automation():
    max_retries = 3
    retry_count = 0

    while retry_count < max_retries:
        try:
            # 尝试浏览器操作
            result = await client.call_tool("browser_navigate", {
                "url": "https://example.com"
            })

            if result.success:
                break

        except Exception as e:
            retry_count += 1
            if retry_count >= max_retries:
                raise Exception(f"{max_retries} 次重试后浏览器操作失败:{e}")

            # 重试前等待
            await asyncio.sleep(2 ** retry_count)

性能优化

async def optimized_scraping():
    # 使用多个标签页进行并发抓取
    urls = ["url1", "url2", "url3"]
    results = []

    for i, url in enumerate(urls):
        if i > 0:
            # 为其他 URL 打开新标签页
            await client.call_tool("browser_new_tab")
            await client.call_tool("browser_switch_tab", {"index": i})

        # 导航到 URL
        await client.call_tool("browser_navigate", {"url": url})

        # 提取内容
        content = await client.call_tool("browser_get_text")
        results.append(content)

    return results

调试和监控

基于截图的调试

async def debug_with_screenshots():
    # 操作前截图
    await client.call_tool("browser_screenshot", {
        "path": "/tmp/before_action.png"
    })

    # 执行操作
    await client.call_tool("browser_click", {
        "selector": ".button"
    })

    # 操作后截图
    await client.call_tool("browser_screenshot", {
        "path": "/tmp/after_action.png"
    })

    # 根据需要比较或分析截图

控制台日志监控

async def monitor_console_logs():
    # 导航到页面
    await client.call_tool("browser_navigate", {
        "url": "https://example.com"
    })

    # 使用 JavaScript 评估检查控制台错误
    console_check = await client.call_tool("browser_evaluate", {
        "expression": """
        (() => {
            const logs = [];
            const originalError = console.error;
            console.error = (...args) => {
                logs.push({level: 'error', message: args.join(' ')});
                originalError.apply(console, args);
            };
            return JSON.stringify(logs);
        })()
        """
    })

    print("控制台监控结果:", console_check.result)

与其他 AIO Sandbox 组件集成

浏览器 + 文件操作

async def browser_to_file_workflow():
    # 使用浏览器抓取数据
    await client.call_tool("browser_navigate", {
        "url": "https://api-docs.example.com"
    })

    # 提取 API 文档
    docs = await client.call_tool("browser_get_text")

    # 保存到文件
    await client.call_tool("file_write", {
        "path": "/tmp/api_docs.txt",
        "content": docs.content[0].text
    })

    # 使用 Shell 命令处理
    await client.call_tool("shell_exec", {
        "command": "grep -E 'POST|GET|PUT|DELETE' /tmp/api_docs.txt > /tmp/endpoints.txt"
    })

浏览器 + 代码执行

async def browser_driven_analysis():
    # 抓取数据
    await client.call_tool("browser_navigate", {
        "url": "https://data-source.com"
    })

    # 使用 JavaScript 评估提取 JSON 数据
    data = await client.call_tool("browser_evaluate", {
        "expression": "document.querySelector('pre.json-data').textContent"
    })

    # 使用 Python 执行处理
    analysis_code = f"""
import json
import pandas as pd

# 解析抓取的数据
data = json.loads('''{data.result}''')

# 执行分析
df = pd.DataFrame(data)
summary = df.describe()

print("数据分析摘要:")
print(summary.to_string())
"""

    result = await client.call_tool("jupyter_execute", {
        "code": analysis_code
    })

    print("分析结果:", result.content)

这份全面的指南涵盖了使用 AIO Sandbox 进行浏览器自动化的主要方法。选择最适合您用例的方法:

  • VNC 用于可视化交互和调试
  • CDP 用于底层浏览器控制
  • MCP 浏览器服务器 用于高级自动化
  • Playwright/Selenium 用于熟悉的框架

对于更高级的场景,将浏览器自动化与其他 AIO Sandbox 组件(如文件操作、Shell 命令和代码执行)结合使用。