核心用法（3/3）

Extend agents with custom Python functions. Add API calls, file operations, or any custom logic as agent tools.

Examples:

• deterministic clicks
• file handling
• calling APIs
• human-in-the-loop
• browser interactions
• calling LLMs
• get 2fa codes
• send emails
• Playwright integration (see GitHub example)
• ...

Simply add @tools.action(...) to your function.

from browser_use import Tools, Agent, ActionResult

tools = Tools()

@tools.action(description='Ask human for help with a question')
async def ask_human(question: str) -> ActionResult:
    answer = input(f'{question} > ')
    return ActionResult(extracted_content=f'The human responded with: {answer}')

agent = Agent(task='...', llm=llm, tools=tools)

• description (required) - What the tool does, the LLM uses this to decide when to call it.
• allowed_domains - List of domains where tool can run (e.g. ['*.example.com']), defaults to all domains

The Agent fills your function parameters based on their names, type hints, & defaults.

Common Pitfall: Parameter names must match exactly! Use browser_session: BrowserSession (not browser: Browser). The agent injects special parameters by name matching, so using incorrect names will cause your tool to fail silently. See Available Objects below for the correct parameter names.

Add Tools from an MCP Server

The Python library can also import tools from an external MCP server. This extends a Browser Use agent; it is different from running Browser Use as an MCP server.

Install uv and Browser Use, and set BROWSER_USE_API_KEY for the agent's model. Warm the reference time server's package cache before connecting:

uvx mcp-server-time==2026.8.18 --help

import asyncio
from browser_use import Agent, ChatBrowserUse, Tools
from browser_use.mcp.client import MCPClient

async def main():
    tools = Tools()
    client = MCPClient(
        server_name="time",
        command="uvx",
        args=["mcp-server-time==2026.8.18", "--local-timezone=UTC"],
    )
    try:
        await client.register_to_tools(
            tools, tool_filter=["get_current_time"], prefix="time_"
        )
        agent = Agent(
            task="Use time_get_current_time to report the current UTC time.",
            llm=ChatBrowserUse(),
            tools=tools,
        )
        await agent.run()
    finally:
        await client.disconnect()

asyncio.run(main())

tool_filter uses the server's original names. prefix changes the names the agent sees and avoids collisions with built-in actions, which remain available. Keep the connection open until the agent finishes, then disconnect it. Create a fresh client for a new run.

The current client launches local stdio servers, has a roughly 10-second connection budget, and reads one tools/list response. It does not yet discover tools on later catalog pages or connect directly to HTTP/SSE endpoints. Only run trusted servers: their subprocesses execute on your machine, and their tools may have side effects. The time lookup itself needs no model key; running the agent above uses your configured model and may incur model charges.

Available Objects

Your function has access to these objects:

• browser_session: BrowserSession - Current browser session for CDP access
• cdp_client - Direct Chrome DevTools Protocol client
• page_extraction_llm: BaseChatModel - The LLM you pass into agent. This can be used to do a custom llm call here.
• file_system: FileSystem - File system access
• available_file_paths: list[str] - Available files for upload/processing
• has_sensitive_data: bool - Whether action contains sensitive data

Browser Interaction Examples

You can use browser_session to directly interact with page elements using CSS selectors:

from browser_use import Tools, Agent, ActionResult, BrowserSession

tools = Tools()

@tools.action(description='Click the submit button using CSS selector')
async def click_submit_button(browser_session: BrowserSession):
    # Get the current page
    page = await browser_session.must_get_current_page()

    # Get element(s) by CSS selector
    elements = await page.get_elements_by_css_selector('button[type="submit"]')

    if not elements:
        return ActionResult(extracted_content='No submit button found')

    # Click the first matching element
    await elements[0].click()

    return ActionResult(extracted_content='Submit button clicked!')

Available methods on Page:

• get_elements_by_css_selector(selector: str) - Returns list of matching elements
• get_element_by_prompt(prompt: str, llm) - Returns element or None using LLM
• must_get_element_by_prompt(prompt: str, llm) - Returns element or raises error

Available methods on Element:

• click() - Click the element
• type(text: str) - Type text into the element
• get_text() - Get element text content
• See browser_use/actor/element.py for more methods

Pydantic Input

You can use Pydantic for the tool parameters:

import json
from pydantic import BaseModel, Field
from browser_use import Tools

tools = Tools()

class Cars(BaseModel):
    name: str = Field(description='The name of the car, e.g. "Toyota Camry"')
    price: int = Field(description='The price of the car as int in USD, e.g. 25000')

@tools.action(description='Save cars to file')
def save_cars(cars: list[Cars]) -> str:
    with open('cars.json', 'w') as f:
        json.dump(cars, f)
    return f'Saved {len(cars)} cars to file'

task = "find cars and save them to file"

Domain Restrictions

Limit tools to specific domains:

@tools.action(
    description='Fill out banking forms',
    allowed_domains=['https://mybank.com']
)
def fill_bank_form(account_number: str) -> str:
    # Only works on mybank.com
    return f'Filled form for account {account_number}'

Advanced Example

For a comprehensive example of custom tools with Playwright integration, see: Playwright Integration Example

This shows how to create custom actions that use Playwright's precise browser automation alongside Browser-Use.

Common Pitfalls

The agent injects special parameters by name, not by type. Using incorrect parameter names is the most common cause of tools failing silently.

❌ Wrong: Using browser: Browser

from browser_use import Tools, ActionResult, Browser

@tools.action('My action')
def my_action(browser: Browser) -> ActionResult:  # WRONG!
    # This will NOT receive the browser session
    pass

✅ Correct: Using browser_session: BrowserSession

from browser_use import Tools, ActionResult, BrowserSession

@tools.action('My action')
async def my_action(browser_session: BrowserSession) -> ActionResult:  # CORRECT!
    page = await browser_session.must_get_current_page()
    # Now you have access to the browser
    return ActionResult(extracted_content='Done')

Key Points

1. Use browser_session: BrowserSession - not browser: Browser
2. Use async functions - recommended for consistency with browser operations
3. Return ActionResult - not plain strings (though strings work, ActionResult provides more control)
4. Parameter names must match exactly - see Available Objects for the full list of injectable parameters

Attach your own CDP observer to a live V4 browser and save request metadata.

You can attach a CDP client to a live V4 browser and record its network events. This is a live observer, not a download of everything that happened during a run. V4 run events describe agent activity; they are not the browser's HTTP request log.

Find the live browser

For an agent run, watch its ordered run events for browser.ready or browser.reattached. The event's data.browser_session_id identifies the browser. You can also list the session's browsers with GET /api/v4/browsers?agentSessionId=SESSION_ID. A session can use more than one browser over time, so do not assume the first browser lasts for every follow-up.

Use an API key with browser read access in the same project to retrieve the active browser's cdpUrl. BROWSER_SESSION_ID is a browser ID, not a run ID or agent session ID.

export BROWSER_SESSION_ID="your-active-browser-id"
# BROWSER_USE_API_KEY must already be set. Do not print the resulting CDP URL.
set -o pipefail
BROWSER_CDP_URL="$(curl --fail --silent --show-error \
  "https://api.browser-use.com/api/v4/browsers/$BROWSER_SESSION_ID" \
  -H "X-Browser-Use-API-Key: $BROWSER_USE_API_KEY" \
  | jq --exit-status --raw-output '.cdpUrl // empty')" || exit 1
export BROWSER_CDP_URL

For a standalone V4 browser, you can use cdpUrl from the create response instead. A stopped browser has no live CDP URL. Check the HTTP status and browser state if discovery fails; do not create a replacement browser and mistake it for the agent's browser.

Treat the CDP URL as a credential. Anyone who can connect can control the browser and read its session. Keep it out of logs, tickets, source control, and shared shell history.

Record metadata from existing pages

Install Playwright with pip install playwright. No local browser download is needed because this example connects to an existing browser.

Save this as capture_network.py, then run python capture_network.py. It records for 30 seconds and creates network.jsonl with owner-only permissions. It refuses to overwrite an existing file. Start it before the activity you want to observe.

import asyncio
import json
import os
from urllib.parse import urlsplit

from playwright.async_api import async_playwright

def http_origin(value):
    try:
        url = urlsplit(value)
        if url.scheme not in {"http", "https"} or not url.hostname:
            return None
        host = f"[{url.hostname}]" if ":" in url.hostname else url.hostname
        port = url.port
    except ValueError:
        return None  # A malformed URL must not interrupt the capture.
    default_port = {"http": 80, "https": 443}[url.scheme]
    suffix = f":{port}" if port not in (None, default_port) else ""
    return f"{url.scheme}://{host}{suffix}"

async def main():
    async with async_playwright() as p:
        cdp_url = os.environ["BROWSER_CDP_URL"]
        try:
            browser = await p.chromium.connect_over_cdp(cdp_url)
        except Exception:
            # Transport errors can include the credential-bearing CDP URL.
            raise RuntimeError("Check the live CDP URL and browser state") from None
        context = browser.contexts[0]
        sessions = []
        fd = os.open("network.jsonl", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)

        with os.fdopen(fd, "w") as output:
            def write(row):
                output.write(json.dumps(row) + "\n")
                output.flush()

            async def observe(page, page_number):
                cdp = await context.new_cdp_session(page)
                sessions.append(cdp)

                def record(event, params):
                    row = {"event": event, "page": page_number,
                           "requestId": params["requestId"]}
                    if event == "request":
                        request = params["request"]
                        row.update(method=request["method"],
                                   origin=http_origin(request["url"]),
                                   resourceType=params.get("type"))
                    elif event == "response":
                        response = params["response"]
                        row.update(status=response["status"],
                                   mimeType=response["mimeType"])
                    elif event == "finished":
                        row["bytes"] = params["encodedDataLength"]
                    else:
                        row["error"] = params.get("errorText")
                    write(row)

                events = {
                    "requestWillBeSent": "request", "responseReceived": "response",
                    "loadingFinished": "finished", "loadingFailed": "failed",
                }
                for name, event in events.items():
                    cdp.on(f"Network.{name}",
                           lambda params, event=event: record(event, params))
                await cdp.send("Network.enable")

            try:
                for number, page in enumerate(context.pages, start=1):
                    await observe(page, number)
                print(f"Observing {len(sessions)} existing page(s) for 30 seconds")
                await asyncio.sleep(30)
            finally:
                for cdp in sessions:
                    try:
                        await cdp.detach()
                    except Exception:
                        pass  # The page or browser may already have closed.

asyncio.run(main())

Each request is correlated by (page, requestId). A response row gives the HTTP status; a finished row gives transferred bytes. Keep the event sequence: redirects can produce more than one request event with the same request ID. Empty output means no matching events were observed, not that the run made no requests.

The example does not navigate, intercept requests, change the cache, or stop the browser. It drops URL paths, queries, credentials, headers, cookies, and request and response bodies. HTTP origins retain non-default ports; malformed and non-HTTP URLs have a null origin. Origins can still be sensitive; review the file before sharing it. Keep a bounded capture duration and a retention policy.

Coverage and lifecycle

• Capture starts only after Network.enable for each attached page. Earlier traffic is not reconstructed, and attaching after browser.ready can miss startup requests.
• This small example attaches only to pages present at startup. Popups, new tabs, workers, service workers, and some cross-process frames need separate target handling. It is not complete browser-wide capture.
• Reattach if V4 provisions a new browser. A logger attached to the old browser cannot observe the replacement.
• JSONL events are not HAR. A HAR exporter needs its own request, redirect, timing, and body handling. Network.getResponseBody is a separate, optional CDP call; bodies may be unavailable and can contain secrets or customer data.
• Detaching an observer does not stop a browser or end its billing. Leave an agent-owned browser under the agent's lifecycle. For a standalone browser you created, stop it through the browser API when finished.

See the CDP Network reference and Playwright CDP sessions for event fields and target-specific behavior.

Save a Browser Use task as a rerunnable script for repeated live data extraction and self-healing runs.

Teach a Cloud agent a repeated browser task once. It can save the working code in its workspace. Later runs reuse that script, fetch fresh data from the live site, and repair the code if the site changed.

This pattern works well for repeated data extraction, checks, and reports. The example below collects the top five Hacker News stories. The first run writes and tests the script. Every later run sends a new request to the live page.

<img alt="A first agent run saves and tests script.py and a README in a workspace; later runs reuse the script and repair it only when necessary" />

<img alt="A first agent run saves and tests script.py and a README in a workspace; later runs reuse the script and repair it only when necessary" />

The file stays. The running process does not. Each later run starts an agent again, but it can reuse the code instead of rebuilding the browser workflow.

Save the working script

Create one workspace, then ask the first run to do the job and save a tested script:

from browser_use_sdk.v4 import BrowserUse

client = BrowserUse()
workspace = client.workspaces.create(name="live-hn-data")

first = client.runs.create(
    """
    Open https://news.ycombinator.com/ and return the top five story titles and
    URLs as JSON.

    Save a script at scripts/extract_top_stories.py that collects the same
    fields. The script must send a fresh request to the live page every time it
    runs. Do not hard-code story titles, save the page HTML, or read an old
    result. Print JSON with source_url set to https://news.ycombinator.com/ and
    include fetched_at, source_sha256, and stories.

    Run the script twice and save the outputs as proof/run-1.json and
    proof/run-2.json. Confirm both outputs have a fresh fetched_at value and
    five stories. Add scripts/README.md with the command to run it again.
    """,
    model="grok-4.5",
    workspace_id=workspace.id,
)
client.runs.wait_for_completion(first.id)

saved = client.workspaces.files(workspace.id, prefix="scripts/")
print([file.path for file in saved.files])
print(f"Keep this workspace ID: {workspace.id}")

import { BrowserUse } from "browser-use-sdk/v4";

const client = new BrowserUse();
const workspace = await client.workspaces.create({ name: "live-hn-data" });

const first = await client.runs.create({
  task: `
    Open https://news.ycombinator.com/ and return the top five story titles and
    URLs as JSON.

    Save a script at scripts/extract_top_stories.py that collects the same
    fields. The script must send a fresh request to the live page every time it
    runs. Do not hard-code story titles, save the page HTML, or read an old
    result. Print JSON with source_url set to https://news.ycombinator.com/ and
    include fetched_at, source_sha256, and stories.

    Run the script twice and save the outputs as proof/run-1.json and
    proof/run-2.json. Confirm both outputs have a fresh fetched_at value and
    five stories. Add scripts/README.md with the command to run it again.
  `,
  model: "grok-4.5",
  workspaceId: workspace.id,
});
await client.runs.waitForCompletion(first.id);

const saved = await client.workspaces.files(workspace.id, {
  prefix: "scripts/",
});
console.log(saved.files.map((file) => file.path));
console.log(`Keep this workspace ID: ${workspace.id}`);

Rerun it later

Copy the workspace ID from the first process. Pass that ID without a session ID to start a new conversation with the saved files:

from browser_use_sdk.v4 import BrowserUse

client = BrowserUse()
workspace_id = "your-workspace-id"

later = client.runs.create(
    """
    Run python scripts/extract_top_stories.py and return its JSON. Check that
    fetched_at is current, source_url is https://news.ycombinator.com/, and it
    has five stories. If the script fails or the result is malformed, inspect
    the live page and repair the saved script before returning the new result.
    """,
    model="grok-4.5",
    workspace_id=workspace_id,
)
result = client.runs.wait_for_completion(later.id)
print(result.result)

import { BrowserUse } from "browser-use-sdk/v4";

const client = new BrowserUse();
const workspaceId = "your-workspace-id";

const later = await client.runs.create({
  task: `
    Run python scripts/extract_top_stories.py and return its JSON. Check that
    fetched_at is current, source_url is https://news.ycombinator.com/, and it
    has five stories. If the script fails or the result is malformed, inspect
    the live page and repair the saved script before returning the new result.
  `,
  model: "grok-4.5",
  workspaceId,
});
const result = await client.runs.waitForCompletion(later.id);
console.log(result.result);

Live data, not a cached answer

The saved script must request the source page on every execution. fetched_at shows when the request ran, and source_sha256 identifies the response body. The data can stay the same between two runs when the page has not changed. The script still fetched it again.

Do not put old output, page HTML, or fixed values inside the script. Save output files only as optional history or proof.

Can rerunnable tasks be cheaper at scale?

Reusing code can remove browser steps and model work that would otherwise be repeated on every extraction. Every later run still starts a model and uses tokens, so the savings depend on the task, site, and model. Compare the cost and duration of the first run with later runs in your own project before making a savings claim.

Workspace or session?

• Pass workspace_id / workspaceId for a new conversation with the same files. The new run gets a new browser.
• Pass session_id / sessionId to continue the old conversation and its workspace. Cloud can also reuse that session's browser while it is still alive.

A saved script can break when a site changes. Ask the agent to check the live result and self-heal the script instead of trusting stale output.

Hook into agent lifecycle events — before/after actions, navigation, extraction, and error handling callbacks.

Browser-Use provides lifecycle hooks that allow you to execute custom code at specific points during the agent's execution. Hook functions can be used to read and modify agent state while running, implement custom logic, change configuration, integrate the Agent with external applications.

Available Hooks

Currently, Browser-Use provides the following hooks:

await agent.run(on_step_start=..., on_step_end=...)

Each hook should be an async callable function that accepts the agent instance as its only parameter.

Basic Example

import asyncio
from pathlib import Path

from browser_use import Agent, ChatOpenAI
from browser_use.browser.events import ScreenshotEvent

async def my_step_hook(agent: Agent):
    # inside a hook you can access all the state and methods under the Agent object:
    #   agent.settings, agent.state, agent.task
    #   agent.tools, agent.llm, agent.browser_session
    #   agent.pause(), agent.resume(), agent.add_new_task(...), etc.

    # You also have direct access to the browser state
    state = await agent.browser_session.get_browser_state_summary()

    current_url = state.url
    visit_log = agent.history.urls()
    previous_url = visit_log[-2] if len(visit_log) >= 2 else None
    print(f'Agent was last on URL: {previous_url} and is now on {current_url}')
    cdp_session = await agent.browser_session.get_or_create_cdp_session()

    # Example: Get page HTML content
    doc = await cdp_session.cdp_client.send.DOM.getDocument(session_id=cdp_session.session_id)
    html_result = await cdp_session.cdp_client.send.DOM.getOuterHTML(
        params={'nodeId': doc['root']['nodeId']}, session_id=cdp_session.session_id
    )
    page_html = html_result['outerHTML']

    # Example: Take a screenshot using the event system
    screenshot_event = agent.browser_session.event_bus.dispatch(ScreenshotEvent(full_page=False))
    await screenshot_event
    result = await screenshot_event.event_result(raise_if_any=True, raise_if_none=True)

    # Example: pause agent execution and resume it based on some custom code
    if '/finished' in current_url:
        agent.pause()
        Path('result.txt').write_text(page_html)
        input('Saved "finished" page content to result.txt, press [Enter] to resume...')
        agent.resume()

async def main():
    agent = Agent(
        task='Search for the latest news about AI',
        llm=ChatOpenAI(model='gpt-5-mini'),
    )

    await agent.run(
        on_step_start=my_step_hook,
        # on_step_end=...
        max_steps=10,
    )

if __name__ == '__main__':
    asyncio.run(main())

Data Available in Hooks

When working with agent hooks, you have access to the entire Agent instance. Here are some useful data points you can access:

• agent.task provides the main task, agent.add_new_task(...) allows you to queue up a new one
• agent.tools gives access to the Tools() object and Registry() containing the available actions
• agent.tools.registry.execute_action('click', {'index': 123}, browser_session=agent.browser_session)
• agent.sensitive_data contains the sensitive data dict, which can be updated in-place to add/remove/modify items
• agent.settings contains all the configuration options passed to the Agent(...) at init time
• agent.llm provides direct access to the main LLM object (e.g. ChatOpenAI)
• agent.state provides access to internal state, including agent thoughts, outputs, actions, and more
• agent.history gives access to historical data from the agent's execution:
• agent.history.model_thoughts(): Reasoning from Browser Use's model.
• agent.history.model_outputs(): Raw outputs from the Browser Use's model.
• agent.history.model_actions(): Actions taken by the agent
• agent.history.extracted_content(): Content extracted from web pages
• agent.history.urls(): URLs visited by the agent
• agent.browser_session provides direct access to the BrowserSession and CDP interface
• agent.browser_session.agent_focus_target_id: Get the current target ID the agent is focused on
• agent.browser_session.get_or_create_cdp_session(): Get the current CDP session for browser interaction
• agent.browser_session.get_tabs(): Get all tabs currently open
• agent.browser_session.get_current_page_url(): Get the URL of the current active tab
• agent.browser_session.get_current_page_title(): Get the title of the current active tab

Tips for Using Hooks

• Avoid blocking operations: Since hooks run in the same execution thread as the agent, keep them efficient and avoid blocking operations.
• Use custom tools instead: Hooks are fairly advanced. Most use cases can be implemented with custom tools instead.
• Increase step_timeout: If your hook is doing something that takes a long time, you can increase the step_timeout parameter in the Agent(...) constructor.

V2 agent models and file handling.

Models

Pass llm explicitly to select a model:

result = await client.run("...", llm="browser-use-2.0")

const result = await client.run("...", { llm: "browser-use-2.0" });

Files

Upload images, PDFs, documents, and text files (10 MB max) to sessions, and download output files from completed tasks.

Upload a file

Get a presigned URL, then upload via POST.

import httpx
from browser_use_sdk import AsyncBrowserUse

client = AsyncBrowserUse()
session = await client.sessions.create()

upload = await client.files.session_url(
    session.id,
    file_name="input.pdf",
    content_type="application/pdf",
    size_bytes=1024,
)

with open("input.pdf", "rb") as f:
    async with httpx.AsyncClient() as http:
        await http.post(upload.url, content=f.read(), headers={"Content-Type": "application/pdf"})

result = await client.run("Summarize the uploaded PDF", session_id=session.id)

import { BrowserUse } from "browser-use-sdk";
import { readFileSync } from "fs";

const client = new BrowserUse();
const session = await client.sessions.create();

const upload = await client.files.sessionUrl(session.id, {
  fileName: "input.pdf",
  contentType: "application/pdf",
  sizeBytes: 1024,
});

await fetch(upload.url, {
  method: "POST",
  body: readFileSync("input.pdf"),
  headers: { "Content-Type": "application/pdf" },
});

const result = await client.run("Summarize the uploaded PDF", { sessionId: session.id });

Presigned URLs expire after 120 seconds. Max file size: 10 MB.

Download task output files

result = await client.tasks.get(task_id)
for file in result.output_files:
    output = await client.files.task_output(task_id, file.id)
    print(output.download_url)  # download URL

const result = await client.tasks.get(taskId);
for (const file of result.outputFiles) {
  const output = await client.files.taskOutput(taskId, file.id);
  console.log(output.downloadUrl);
}

Streaming steps

Use async for to yield steps as the agent works.

from browser_use_sdk import AsyncBrowserUse

client = AsyncBrowserUse()
run = client.run("Find the most upvoted post on Reddit r/technology today")
async for step in run:
    print(f"Step {step.number}: {step.next_goal}")
    print(f"  URL: {step.url}")

print(run.result.output)  # final result after iteration

import { BrowserUse } from "browser-use-sdk";

const client = new BrowserUse();
const run = client.run("Find the most upvoted post on Reddit r/technology today");
for await (const step of run) {
  console.log(`Step ${step.number}: ${step.nextGoal}`);
  console.log(`  URL: ${step.url}`);
}

console.log(run.result?.output);  // final result after iteration

Key parameters