使用 SDK 工具集进行浏览器和计算机使用

from anthropic import Anthropic
from example_browser_provider import ExampleBrowser  # made-up provider

client = Anthropic()
with ExampleBrowser() 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 its 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 BetaMessageParam, BetaToolResultBlockParam

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

with MyBrowser(backend, url_policy=url_policy) as browser:
    messages: list[BetaMessageParam] = [{"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})

import re
from urllib.parse import urlsplit

from anthropic import Anthropic
from anthropic.tools import ToolError
from anthropic.tools.browser import (
    BetaAbstractBrowserToolset20260801,
    BetaBrowserNavigateResult,
    BetaBrowserState,
    BetaScreenshotResult,
    BetaToolsetCallContext,
    BetaURLContext,
)
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: BetaToolsetCallContext) -> BetaBrowserState:
        return BetaBrowserState(
            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: BetaToolsetCallContext, input: BetaBrowserNavigateInput
    ) -> BetaBrowserNavigateResult:
        # input.url 为 "back"、"forward"、"reload"，或是您的 URL 策略
        # 已允许的 URL（保持 Claude 所写的原样）。与 is_allowed 一样，backend.goto 会为
        # 不以协议开头的 URL 添加 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: BetaToolsetCallContext, input: BetaBrowserScreenshotInput
    ) -> BetaScreenshotResult:
        data = self.backend.png_base64(input.tab_id)
        return BetaScreenshotResult(data=data, media_type="image/png")

    def left_click(
        self, context: BetaToolsetCallContext, 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()


ALLOWED_HOSTS = ("example.com", "iana.org")
SCHEME_PREFIX = re.compile(r"[a-z][a-z0-9+.-]*:", re.IGNORECASE)


def is_allowed(url: str) -> bool:
    # 仅为示例，并非生产环境策略。请参阅 "Write a URL policy"。
    if url.lower() == "about:blank":
        return True
    with_scheme = url if SCHEME_PREFIX.match(url) else f"https://{url}"
    # 在 Web URL 中，浏览器会将 "\" 视为 "/"。
    try:
        parts = urlsplit(with_scheme.replace("\\", "/"))
    except ValueError:
        return False
    host = parts.hostname or ""
    listed = any(host == name or host.endswith(f".{name}") for name in ALLOWED_HOSTS)
    return parts.scheme in ("http", "https") and listed


def url_policy(context: BetaURLContext, url: str) -> None:
    if not is_allowed(url):
        raise ToolError(f"blocked: {url} is not on an allowed host")


client = Anthropic()
with MyBrowser(backend, url_policy=url_policy) 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.",
            }
        ],
        stream=True,
        run_tools_eagerly=True,  # so a call can start before the response ends
    )
    for stream in runner:
        print(stream.get_final_message())

import re
from urllib.parse import urlsplit

from anthropic.tools import ToolError
from anthropic.tools.browser import BetaURLContext

ALLOWED_HOSTS = ("example.com", "iana.org")
SCHEME_PREFIX = re.compile(r"[a-z][a-z0-9+.-]*:", re.IGNORECASE)


def is_allowed(url: str) -> bool:
    # 仅为示例，并非生产环境策略。
    if url.lower() == "about:blank":
        return True
    with_scheme = url if SCHEME_PREFIX.match(url) else f"https://{url}"
    # 在 Web URL 中，浏览器会将 "\" 视为 "/"。
    try:
        parts = urlsplit(with_scheme.replace("\\", "/"))
    except ValueError:
        return False
    host = parts.hostname or ""
    listed = any(host == name or host.endswith(f".{name}") for name in ALLOWED_HOSTS)
    return parts.scheme in ("http", "https") and listed


def url_policy(context: BetaURLContext, url: str) -> None:
    if not is_allowed(url):
        raise ToolError(f"blocked: {url} is not on an allowed host")


browser = MyBrowser(backend, url_policy=url_policy)

class MyBrowser(BetaAbstractBrowserToolset20260801):
    ...

    # 通过 context.route("**/*", self._guard) 注册到 Playwright 浏览器上下文
    def _guard(self, route):
        # is_allowed 是“编写 URL 策略”一节中的函数
        if not is_allowed(route.request.url):
            return route.abort("blockedbyclient")
        route.continue_()

from anthropic.tools.browser import BetaLocalFilePolicy

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

import json
from collections.abc import Callable
from urllib.parse import urlsplit

from anthropic.tools.browser import BetaConfirmContext

GATED = {"javascript_exec", "file_upload"}


def shown(value: object) -> str:
    """A value as JSON, with every character outside printable ASCII escaped."""
    return json.dumps(value, ensure_ascii=True, indent=2)


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

    def confirm(context: BetaConfirmContext) -> bool:
        name = context.member
        if name not in GATED:
            return True
        detail = shown(context.input.to_dict())
        page = context.tab_url
        where = shown(page) if page else "a tab with no URL"
        question = f"Allow {name} on {where}?\n{detail}"
        if page is None or urlsplit(page).scheme not in ("http", "https"):
            return ask_user(question)  # no web origin: ask every time
        # 一次批准仅适用于此页面上的这一确切输入。
        key = (name, page, detail)
        if key not in granted and ask_user(question):
            granted.add(key)
        return key in granted

    return confirm


# 同时实现了 javascript_exec 和 file_upload 的 MyBrowser
# file_upload 还需要“限制上传和下载”中的 file_policy
browser = MyBrowser(
    backend,
    configs={"javascript_exec": {"enabled": True}, "file_upload": {"enabled": True}},
    confirm=make_confirm(),
    url_policy=url_policy,
)

from anthropic import Anthropic
from anthropic.tools.computer import (
    BetaAbstractComputerToolset20260801,
    BetaComputerCursorPositionResult,
    BetaScreenshotResult,
    BetaToolsetCallContext,
)
from anthropic.types.beta import (
    BetaComputerCursorPositionInput,
    BetaComputerKeyInput,
    BetaComputerLeftClickInput,
    BetaComputerScreenshotInput,
    BetaComputerTypeInput,
)


class MyDesktop(BetaAbstractComputerToolset20260801):
    def __init__(self, display, **options):
        super().__init__(**options)
        self.display = display

    def screenshot(
        self, context: BetaToolsetCallContext, input: BetaComputerScreenshotInput
    ) -> BetaScreenshotResult:
        # Claude 的坐标是此图像中的像素位置。
        data = self.display.png_base64()
        return BetaScreenshotResult(data=data, media_type="image/png")

    def cursor_position(
        self, context: BetaToolsetCallContext, input: BetaComputerCursorPositionInput
    ) -> BetaComputerCursorPositionResult:
        x, y = self.display.cursor()
        return BetaComputerCursorPositionResult(x=x, y=y)

    def left_click(
        self, context: BetaToolsetCallContext, input: BetaComputerLeftClickInput
    ) -> None:
        # input.coordinate 为 [x, y]，或为 None 表示在光标处点击。input.text 是
        # 要按住的修饰键，例如 "shift"。无需返回内容：Claude 读取 "Clicked."
        self.display.click(input.coordinate, input.text)

    def type(
        self, context: BetaToolsetCallContext, input: BetaComputerTypeInput
    ) -> None:
        self.display.type(input.text)

    def key(self, context: BetaToolsetCallContext, input: BetaComputerKeyInput) -> None:
        self.display.press(input.text, input.repeat or 1)

    def close(self) -> None:
        super().close()  # first, so no call is still using the desktop when it closes
        self.display.close()


client = Anthropic()
with MyDesktop(
    display,
    # type 和 key 操作必需；请参阅“安全运行计算机工具集”
    confirm=lambda context: ask_user(f'Allow the "{context.member}" tool?'),
) as desktop:
    runner = client.beta.messages.tool_runner(
        model="claude-opus-5-5",
        max_tokens=1024,
        tools=[desktop],
        messages=[
            {"role": "user", "content": "Open the calculator and compute 17 * 23."}
        ],
    )
    for message in runner:
        print(message)

from anthropic.tools import ToolError

SCREENSHOT_SIZE = (1280, 720)  # the size of every full screenshot the driver returns
DISPLAY_SIZE = (2560, 1440)


def to_display(coordinate: list[int]) -> tuple[int, int]:
    x, y = coordinate
    width, height = SCREENSHOT_SIZE
    if not (0 <= x < width and 0 <= y < height):
        # 拒绝该坐标点。若将其移入范围内，点击位置将并非 Claude 所选。
        raise ToolError(f"[{x}, {y}] is outside the {width}x{height} screenshot")
    return x * DISPLAY_SIZE[0] // width, y * DISPLAY_SIZE[1] // height


def to_screenshot(x: int, y: int) -> tuple[int, int]:
    width, height = DISPLAY_SIZE
    return x * SCREENSHOT_SIZE[0] // width, y * SCREENSHOT_SIZE[1] // height


to_display([640, 360])  # (1280, 720)

from anthropic.tools.computer import BetaComputerConfirmContext

KEYBOARD = {"type", "key", "hold_key"}
CLICKS = {
    "left_click",
    "right_click",
    "middle_click",
    "double_click",
    "triple_click",
    "left_click_drag",
    "left_mouse_down",
    "left_mouse_up",
}
ASK_FIRST = KEYBOARD


def confirm(context: BetaComputerConfirmContext) -> bool:
    if context.member not in ASK_FIRST:
        return True
    # 此处所示为“Gate consequential members”中的函数。
    detail = shown(context.input.to_dict())
    return ask_user(f'Allow the "{context.member}" tool with this input?\n{detail}')


desktop = MyDesktop(display, confirm=confirm)

# MyBrowser、backend 和 url_policy 来自高级示例。
# confirm 是“安全运行计算机工具集”中的函数。
with (
    MyBrowser(backend, url_policy=url_policy) as browser,
    MyDesktop(display, confirm=confirm) as desktop,
):
    runner = client.beta.messages.tool_runner(
        model="claude-opus-5-5",
        max_tokens=1024,
        tools=[browser, desktop],
        messages=[
            {
                "role": "user",
                "content": "Copy the heading on example.com into the open text editor.",
            }
        ],
        stream=True,
        run_tools_eagerly=True,  # so a call can start before the response ends
    )
    for stream in runner:
        print(stream.get_final_message())