Browser Use

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

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

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

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

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

要求

GET /v2/sandboxcapabilities 中,browser.status 应为 "ready",同时还应包含 filesexec

GET /v1/capabilities 的返回结果中,browser.status 应为 "ready",并且 filesexec 也应可用。

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

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

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

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

如果需要读取完整返回结构(successmessage 和状态码)或二进制响应体,请改用 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"
        }
      ]
    }
  ]
}

一个节点带 rolenamerefchildren

ref 是该元素在 Chromium 中的 backend node id,clickfillupload 都通过它定位。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 写入值,然后派发 inputchange 事件。因此,受框架控制的输入框也会收到真实变更,而不只是修改 DOM 属性。

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

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

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

AIO_AGENT_BROWSER_BIN 指向 agent-browser CLI 后,按 ref 操作的工具(snapshotclickfillupload)会改由该 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

响应状态为 200content-typeimage/jpeg

format 默认为 pngquality 只对 jpeg 生效。full_page=true 会截取整个可滚动页面,而不只是视口;PNG 响应还包含 x-image-widthx-image-height

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

处理失败

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

两种情况都要检查返回结构中的 successmessage;第二种还要检查状态码。因此,示例使用 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 返回 503messagebrowser: 开头。

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

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

调用形状上的错误是普通的:既没有 selector 也没有 ref400ref 不是 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 默认发 SIGTERMsignal 可以换成别的信号名。在长期运行的沙箱里,留着不用的进程是真实开销,它占的端口也是。

还可以检查并清理会话。

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

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

相关页面