• 简体中文

      • Browser Use

      这个示例从零搭一个小网页：agent 写入页面、启动静态服务器，然后通过 HTTP 反复观察页面、填写姓名、点击链接、读取结果，最后截图。

      页面由文件 API 和 command plane 在沙箱中创建，因此整个流程可离线运行，每次结果一致。

      工具包括 navigate、snapshot、fill、click、evaluate 和 screenshot，路由如下：

      • /v2/browser/*
      • /v1/browser/*

      创建示例页面还需要一次文件写入和一次命令执行。两套接口的请求体完全相同。

      要求

      GET /v2/sandbox 的 capabilities 中，browser.status 应为 "ready"，同时还应包含 files 和 exec。

      GET /v1/capabilities 的返回结果中，browser.status 应为 "ready"，并且 files 和 exec 也应可用。

      browser plane 是 CDP 客户端，因此主机上必须有 Chromium 监听调试端口，默认地址为 127.0.0.1:9222。详见 浏览器 API。

      页面由 python3 启动的静态服务器提供，也可以替换为其他静态服务器。选择一个未被占用的端口，下面使用 18780。

      各段代码接续同一个脚本，因此会沿用 BASE_URL、第一段创建的客户端和端口。

      大多数调用使用 sb.post 或 sb.get，这两个方法会自动取出返回结构中的 data。

      如果需要读取完整返回结构（success、message 和状态码）或二进制响应体，请改用 sb.http。

      准备可驱动的页面

      先通过文件 API 写入两个 HTML 文件，再通过 command plane 启动静态服务器。服务器在后台启动后立即返回，后续可通过响应中的 id 操作该进程：

      Python
      TypeScript
      import time

BASE_URL = "http://127.0.0.1:18091"
PORT = 18780
DIR = "/tmp/webapp"
sb = Aio(BASE_URL)

INDEX_HTML = """<!doctype html>
<title>Greeter</title>
<input id="name" placeholder="Your name">
<button id="greet">Greet</button>
<p id="result"></p>
<a href="/thanks.html">Thanks</a>
<script>
document.getElementById('greet').onclick = () => {
  const name = document.getElementById('name').value;
  setTimeout(() => {
    document.getElementById('result').textContent = 'hello, ' + name;
  }, 300);
};
</script>
"""

THANKS_HTML = """<!doctype html>
<title>Thanks</title>
<p>Thanks!</p>
<a href="/index.html">Back</a>
"""

for name, html in [("index.html", INDEX_HTML), ("thanks.html", THANKS_HTML)]:
    written = sb.post("/v2/fs/write", path=f"{DIR}/{name}", content=html)
    print(written["file"], written["bytes_written"])
# → /tmp/webapp/index.html 398
# → /tmp/webapp/thanks.html 84

serve = sb.post(
    "/v2/commands",
    command=f"python3 -m http.server {PORT} --bind 127.0.0.1 --directory {DIR}",
    mode="async",
)
command = serve["command_id"]
print(serve["status"], command)
# → running aff4cddb-8933-494d-9baf-1f5b3401cfe1


def port_state() -> str:
    probe = (f"python3 -c \"import socket; print('free' if "
             f"socket.socket().connect_ex(('127.0.0.1', {PORT})) else 'in use')\"")
    out = sb.post("/v2/commands", command=probe)
    return out["output"].strip()


while port_state() == "free":
    time.sleep(0.2)
print(port_state())
# → in use
      Python
      TypeScript
      import time

from agent_sandbox import Sandbox

BASE_URL = "http://127.0.0.1:18091"
PORT = 18780
DIR = "/tmp/webapp"
client = Sandbox(base_url=BASE_URL)
# The 1.x SDK has no snapshot, fill, or ref click, and no file delete; those
# steps go through the shared helper, with the same bodies the v2 tools take.
sb = Aio(BASE_URL)

INDEX_HTML = """<!doctype html>
<title>Greeter</title>
<input id="name" placeholder="Your name">
<button id="greet">Greet</button>
<p id="result"></p>
<a href="/thanks.html">Thanks</a>
<script>
document.getElementById('greet').onclick = () => {
  const name = document.getElementById('name').value;
  setTimeout(() => {
    document.getElementById('result').textContent = 'hello, ' + name;
  }, 300);
};
</script>
"""

THANKS_HTML = """<!doctype html>
<title>Thanks</title>
<p>Thanks!</p>
<a href="/index.html">Back</a>
"""

for name, html in [("index.html", INDEX_HTML), ("thanks.html", THANKS_HTML)]:
    written = client.file.write_file(file=f"{DIR}/{name}", content=html).data
    print(written.file, written.bytes_written)
# → /tmp/webapp/index.html 398
# → /tmp/webapp/thanks.html 84

serve = client.bash.exec(
    command=f"python3 -m http.server {PORT} --bind 127.0.0.1 --directory {DIR}",
    async_mode=True,
).data
session = serve.session_id
print(serve.status, session)
# → running fb118692-a65c-40f7-a8d1-53b469aab28c


def port_state() -> str:
    probe = (f"python3 -c \"import socket; print('free' if "
             f"socket.socket().connect_ex(('127.0.0.1', {PORT})) else 'in use')\"")
    return client.bash.exec(command=probe).data.output.strip()


while port_state() == "free":
    time.sleep(0.2)
print(port_state())
# → in use

      command plane 写入的内容，文件 API 也能读取。

      两个 plane 共享同一个文件系统，因此服务器提供的正是上一步写入的文件。

      必须等待端口就绪。后台命令可能在进程绑定端口前就返回；如果过早导航，Chromium 会打开错误页，而不是示例页面。

      流程最后会使用同一个 helper 确认端口已经释放。

      导航与观察

      navigate 会在 wait_until 指定的事件触发后返回，默认等待 load。

      完整返回结构还包含 message: "Navigated"，但 sb.post 会将其剥离，只保留 data。

      evaluate 在页面中执行表达式；snapshot 返回无障碍树。agent 读取的是这棵树，而不是像素或坐标：

      Python
      TypeScript
      nav = sb.post("/v2/browser/navigate", url=f"http://127.0.0.1:{PORT}/index.html")
print(nav["url"])
# → http://127.0.0.1:18780/index.html

title = sb.post("/v2/browser/evaluate", expression="document.title")["value"]
print(title)
# → Greeter

snapshot = sb.post("/v2/browser/snapshot", interactive_only=True)
      Python
      TypeScript
      nav = sb.post("/v1/browser/navigate", url=f"http://127.0.0.1:{PORT}/index.html")
print(nav["url"])
# → http://127.0.0.1:18780/index.html

title = sb.post("/v1/browser/evaluate", expression="document.title")["value"]
print(title)
# → Greeter

snapshot = sb.post("/v1/browser/snapshot", interactive_only=True)

      interactive_only: true 把树剪成可操作的节点；本示例页面会返回三个：

      {
  "role": "RootWebArea",
  "children": [
    {
      "role": "group",
      "children": [
        {
          "role": "textbox",
          "name": "Your name",
          "ref": "e6"
        },
        {
          "role": "button",
          "name": "Greet",
          "ref": "e14"
        },
        {
          "role": "link",
          "name": "Thanks",
          "ref": "e16"
        }
      ]
    }
  ]
}

      一个节点带 role、name、ref 和 children。

      ref 是该元素在 Chromium 中的 backend node id，click、fill 和 upload 都通过它定位。ref 的取值每次加载页面都会变化，因此要从快照中读取，不能写死在脚本里。

      填写、点击并等待结果

      这个循环由两个 helper 组成：一个按 role 和 name 找 ref，另一个轮询页面，直到出现 agent 等待的内容。

      Python
      TypeScript
      def find_ref(node: dict, role: str, name: str) -> str:
    if node.get("role") == role and node.get("name") == name:
        return node["ref"]
    for child in node.get("children", []):
        found = find_ref(child, role, name)
        if found:
            return found
    return ""


box = find_ref(snapshot, "textbox", "Your name")
button = find_ref(snapshot, "button", "Greet")

sb.post("/v2/browser/fill", ref=box, value="Ada")
sb.post("/v2/browser/click", ref=button)


def wait_for_text(selector: str, tries: int = 20) -> str:
    expr = f"document.querySelector({selector!r}).textContent"
    for _ in range(tries):
        value = sb.post("/v2/browser/evaluate", expression=expr)["value"]
        if value:
            return value
        time.sleep(0.25)
    raise TimeoutError(f"{selector} stayed empty")


print(wait_for_text("#result"))
# → hello, Ada
      Python
      TypeScript
      def find_ref(node: dict, role: str, name: str) -> str:
    if node.get("role") == role and node.get("name") == name:
        return node["ref"]
    for child in node.get("children", []):
        found = find_ref(child, role, name)
        if found:
            return found
    return ""


box = find_ref(snapshot, "textbox", "Your name")
button = find_ref(snapshot, "button", "Greet")

sb.post("/v1/browser/fill", ref=box, value="Ada")
sb.post("/v1/browser/click", ref=button)


def wait_for_text(selector: str, tries: int = 20) -> str:
    expr = f"document.querySelector({selector!r}).textContent"
    for _ in range(tries):
        value = sb.post("/v1/browser/evaluate", expression=expr)["value"]
        if value:
            return value
        time.sleep(0.25)
    raise TimeoutError(f"{selector} stayed empty")


print(wait_for_text("#result"))
# → hello, Ada

      fill 使用元素原型上的原生 setter 写入值，然后派发 input 和 change 事件。因此，受框架控制的输入框也会收到真实变更，而不只是修改 DOM 属性。

      按 ref 执行 click 时，鼠标会在元素中心按下并抬起，页面自己的处理函数也会执行。

      两个操作成功时都会返回 message: "Filled" 或 "Clicked"。sb.post 会剥离这部分内容，因此示例使用 wait_for_text 确认 DOM 已经发生变化。

      默认情况下，这些工具都使用 daemon 内置的 CDP 后端。

      将 AIO_AGENT_BROWSER_BIN 指向 agent-browser CLI 后，按 ref 操作的工具（snapshot、click、fill、upload）会改由该 CLI 提供，响应也会带上 hint: "backend=agent-browser"。

      wait_for_text 是本示例统一使用的等待方式。由于没有专门的 wait 路由，agent 会轮询 evaluate，直到页面出现目标内容或达到次数上限。

      示例页面会在 300 毫秒后写入结果，因此第一次轮询得到的是空字符串。

      截图

      截图路由返回的是图片本身，不是 JSON：

      Python
      TypeScript
      shot = sb.http.get("/v2/browser/screenshot", params={"format": "jpeg", "quality": 80})
with open("greeter.jpg", "wb") as out:
    out.write(shot.content)
print(len(shot.content))
# → 7382
      Python
      TypeScript
      shot = b"".join(client.browser.screenshot(format="jpeg", quality=80))
with open("greeter.jpg", "wb") as out:
    out.write(shot)
print(len(shot))
# → 7382

      响应状态为 200，content-type 为 image/jpeg。

      format 默认为 png，quality 只对 jpeg 生效。full_page=true 会截取整个可滚动页面，而不只是视口；PNG 响应还包含 x-image-width 和 x-image-height。

      图片字节流会直接返回给调用方，不会写入沙箱。沙箱生成的文件请通过文件 API 的下载路由取回，见 文件操作。

      处理失败

      下面有两种容易误判的失败：它们在 HTTP 层都可能看起来像成功。

      两种情况都要检查返回结构中的 success 和 message；第二种还要检查状态码。因此，示例使用 sb.http，而不是 sb.post：

      Python
      TypeScript
      dead = sb.http.post("/v2/browser/navigate",
                    json={"url": f"http://127.0.0.1:{PORT + 1}/"}).json()
print(dead["success"], dead["message"])
# → True Navigated
where = sb.post("/v2/browser/evaluate", expression="location.href")["value"]
print(where)
# → chrome-error://chromewebdata/

sb.post("/v2/browser/navigate", url=f"http://127.0.0.1:{PORT}/thanks.html")
stale = sb.http.post("/v2/browser/fill", json={"ref": box, "value": "Ada"})
print(stale.status_code, stale.json()["message"])
# → 503 browser: DOM.resolveNode: No node with given id found
      Python
      TypeScript
      dead = sb.http.post("/v1/browser/navigate",
                    json={"url": f"http://127.0.0.1:{PORT + 1}/"}).json()
print(dead["success"], dead["message"])
# → True Navigated
where = sb.post("/v1/browser/evaluate", expression="location.href")["value"]
print(where)
# → chrome-error://chromewebdata/

sb.post("/v1/browser/navigate", url=f"http://127.0.0.1:{PORT}/thanks.html")
stale = sb.http.post("/v1/browser/fill", json={"ref": box, "value": "Ada"})
print(stale.status_code, stale.json()["message"])
# → 503 browser: DOM.resolveNode: No node with given id found

      连接被拒绝仍然算一次导航：Chromium 会加载自己的错误页，而该页面会触发 load，所以 navigate 仍然返回成功。

      此时应检查页面最终内容，而不能只看状态码。另一种情况是主机完全没有响应；生命周期事件或底层 CDP 调用超时后，navigate 返回 503，message 以 browser: 开头。

      ref 只属于获取它的那份文档。导航后，同一个 ref 可能无法再次解析，也可能解析到当前占用该 id 的其他节点。

      因此，同一个调用在另一次运行中可能返回 200 Filled，但实际填写了其他元素。每次导航后都要重新获取快照，不能依赖旧 ref。

      调用形状上的错误是普通的：既没有 selector 也没有 ref 是 400，ref 不是 e<number> 是 400，选择器匹配不到是 404。

      清理资源

      如果不停止页面服务器，脚本结束后它仍会继续运行。

      示例会先向进程发送信号，再删除页面目录，并通过第一段中的端口探测确认端口已释放。两个响应都需要读取 message，因此使用 sb.http：

      Python
      TypeScript
      killed = sb.http.post(f"/v2/commands/{command}/kill", json={}).json()
removed = sb.http.post("/v2/fs/delete", json={"path": DIR, "recursive": True}).json()
print(killed["message"], "|", removed["message"], "|", port_state())
# → Operation successful | Directory deleted successfully | free
      Python
      TypeScript
      killed = client.bash.kill(session_id=session)
removed = sb.http.post("/v1/file/delete", json={"path": DIR, "recursive": True}).json()
print(killed.message, "|", removed["message"], "|", port_state())
# → Signal SIGTERM sent | Directory deleted successfully | free

      kill 默认发 SIGTERM，signal 可以换成别的信号名。在长期运行的沙箱里，留着不用的进程是真实开销，它占的端口也是。

      还可以检查并清理会话。

      不带会话 id 的执行会创建新会话，因此上面的轮询循环会留下多个会话。

      • GET /v2/commands/sessions：列出会话。
      • DELETE /v2/commands/sessions/{id}：关闭指定会话。
      • GET /v1/bash/sessions：列出会话。
      • POST /v1/bash/sessions/{id}/close：关闭指定会话。

      相关页面