通过 SDK 工具集使用浏览器使用工具

from anthropic import Anthropic
from anthropic.tools.browser import (
    BetaAbstractBrowserToolset20260801,
    BetaBrowserNavigateResult,
    BetaBrowserScreenshotResult,
    BrowserState,
    ToolsetCallContext,
)
from anthropic.types.beta import (
    BetaBrowserLeftClickInput,
    BetaBrowserNavigateInput,
    BetaBrowserScreenshotInput,
    BetaBrowserStateTabEntryParam,
)


class MyBrowser(BetaAbstractBrowserToolset20260801):
    def __init__(self, backend, **options):
        super().__init__(**options)
        self.backend = backend

    def _browser_state(self, context: ToolsetCallContext) -> BrowserState:
        return BrowserState(
            tabs=[
                BetaBrowserStateTabEntryParam(
                    tab_id=tab.id,
                    title=tab.title,
                    url=tab.url,
                    active=tab.id == self.backend.active,
                )
                for tab in self.backend.tabs()
            ],
            state_changes=self.backend.drain_changes(),
        )

    def navigate(
        self, context: ToolsetCallContext, input: BetaBrowserNavigateInput
    ) -> BetaBrowserNavigateResult:
        # input.url 是已通过 URL 策略检查的 URL，或为 "back"、"forward"、
        # 或 "reload"。当 Claude 省略协议时，SDK 会自动添加 https://。
        page = self.backend.goto(input.url, input.tab_id)
        return BetaBrowserNavigateResult(
            url=page.url, status=page.status, title=page.title
        )

    def screenshot(
        self, context: ToolsetCallContext, input: BetaBrowserScreenshotInput
    ) -> BetaBrowserScreenshotResult:
        data = self.backend.png_base64(input.tab_id)
        return BetaBrowserScreenshotResult(data=data, media_type="image/png")

    def left_click(
        self, context: ToolsetCallContext, input: BetaBrowserLeftClickInput
    ) -> None:
        # 无需返回内容：Claude 读取到的是 "Clicked."
        self.backend.click(input.target, input.tab_id)

    def close(self) -> None:
        super().close()  # first, so no call is still using the browser when it closes
        if not self.backend.closed:
            self.backend.close()


client = Anthropic()
with MyBrowser(backend, allowed_domains=["example.com", "iana.org"]) as browser:
    runner = client.beta.messages.tool_runner(
        model="claude-opus-5-5",
        max_tokens=1024,
        tools=[browser],
        messages=[
            {
                "role": "user",
                "content": "Open example.com and tell me the page heading.",
            }
        ],
    )
    for message in runner:
        print(message)

# 同时实现了 read_console 的 MyBrowser
browser = MyBrowser(
    backend, configs={"read_console": {"enabled": True}, "navigate": {"enabled": False}}
)

import time


class TracedBrowser(MyBrowser):
    def execute(self, context, name, input):
        started = time.monotonic()
        result = super().execute(context, name, input)
        elapsed_ms = (time.monotonic() - started) * 1000
        call_id = context.tool_use.id if context.tool_use else "-"
        log.info("%s %s %.0fms", call_id, name, elapsed_ms)
        return redact(result) if name == "get_page_text" else result

from anthropic.types.beta import BetaToolResultBlockParam

NOT_EXECUTED = "Not executed: an earlier action in this turn failed."
MAX_TURNS = 10

with MyBrowser(backend, allowed_domains=["example.com"]) as browser:
    messages = [{"role": "user", "content": "Open example.com"}]
    for _ in range(MAX_TURNS):
        response = client.beta.messages.create(
            model="claude-opus-5-5",
            max_tokens=1024,
            tools=[browser],
            messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})
        calls = [
            block
            for block in response.content
            if block.type == "tool_use" and block.toolset_name == browser.toolset_name
        ]
        if not calls:
            break
        results: list[BetaToolResultBlockParam] = []
        failed = False
        for call in calls:
            if failed:
                # 某次调用失败后，本轮中其余的调用只会得到应答，而不会实际执行。
                results.append(
                    {
                        "type": "tool_result",
                        "tool_use_id": call.id,
                        "toolset_name": call.toolset_name,
                        "content": NOT_EXECUTED,
                        "is_error": True,
                    }
                )
                continue
            result = browser.tool_result(call)
            failed = bool(result.get("is_error"))
            results.append(result)
        messages.append({"role": "user", "content": results})

browser = MyBrowser(backend, allowed_domains=["example.com", "iana.org"])

# 列出浏览器可以访问但 Claude 不得访问的主机和网络：
# IPv4 和 IPv6 中的环回、链路本地和私有地址范围、0.0.0.0/8、您的
# 云平台元数据地址和名称（例如 metadata.google.internal）、
# localhost，以及您的内部主机名。
browser = MyBrowser(backend, blocked_domains=internal_networks)

from urllib.parse import urlsplit

from anthropic.tools.browser import ToolError, URLContext, default_url_policy

listed = default_url_policy(allowed_domains=["example.com"])


def url_policy(context: URLContext, url: str) -> None:
    listed(context, url)  # the default rules first
    if context.phase == "request" and urlsplit(url).scheme != "https":
        raise ToolError("Only https navigation is allowed.")


browser = MyBrowser(backend, url_policy=url_policy)

from anthropic.tools.browser import URLContext


class MyBrowser(BetaAbstractBrowserToolset20260801):
    ...

    # 通过 context.route("**/*", self._guard) 注册到 Playwright 浏览器上下文
    def _guard(self, route):
        url_context = URLContext(
            member="navigate", phase="request", tab_id=self.backend.active
        )
        if not self.check_url(url_context, route.request.url).allowed:
            return route.abort("blockedbyclient")
        route.continue_()

from anthropic.tools.browser import LocalFilePolicy

# 同时实现了 file_upload 的 MyBrowser
browser = MyBrowser(
    backend,
    configs={"file_upload": {"enabled": True}},
    confirm=make_confirm(),  # required for file_upload; see Gate consequential members
    file_policy=LocalFilePolicy(
        upload_roots=["/task/uploads"],
        download_dir="/task/downloads",
        expose_download_paths=False,
    ),
)

import json
from collections.abc import Callable

from anthropic.tools.browser import ConfirmContext

GATED = {"javascript_exec", "file_upload"}


def shown(context: ConfirmContext) -> str:
    """The call's input as JSON, with every character outside printable
    ASCII escaped."""
    return json.dumps(context.input.to_dict(), ensure_ascii=True, indent=2)


def make_confirm() -> Callable[[ConfirmContext], bool]:
    granted: set[tuple[str, str, str]] = set()

    def confirm(context: ConfirmContext) -> bool:
        name = context.member
        if name not in GATED:
            return True
        detail = shown(context)
        page = context.tab_url
        origin = context.origin
        if page is None or origin is None or origin.startswith("chrome-error:"):
            # 无来源或为错误页面：每次都询问。
            return ask_user(
                f"Allow {name} on {page or 'a page with no origin'}?\n{detail}"
            )
        # 一次批准仅适用于此页面上的这一确切输入。
        key = (name, page, detail)
        if key not in granted and ask_user(f"Allow {name} on {page}?\n{detail}"):
            granted.add(key)
        return key in granted

    return confirm


# 同时实现了 javascript_exec 和 file_upload 的 MyBrowser
browser = MyBrowser(
    backend,
    configs={"javascript_exec": {"enabled": True}, "file_upload": {"enabled": True}},
    confirm=make_confirm(),
)