Referenz des MCP-Servers¶

async def browser_open(
    self,
    session_id: Annotated[
        Optional[str], Field(description="Custom ID; a random 12-character hex ID if omitted.")
    ] = None,
    headless: Annotated[bool, Field(description="Hide the browser window.")] = True,
    real_chrome: Annotated[bool, Field(description="Use locally installed Chrome.")] = False,
    timezone_id: Annotated[
        str | None, Field(description="Browser timezone; uses the system timezone if omitted.")
    ] = None,
    locale: Annotated[
        str | None,
        Field(description="Browser language, Accept-Language, and formatting locale; system default if omitted."),
    ] = None,
    useragent: Annotated[
        Optional[str],
        Field(description="User-Agent override; otherwise generated in headless mode, native in headful mode."),
    ] = None,
    proxy: Annotated[
        Optional[str | Dict[str, str]],
        Field(description="Proxy URL, or dictionary with server and optional username/password."),
    ] = None,
    cdp_url: Annotated[
        Optional[str],
        Field(
            description="Connect to an existing Chromium browser over CDP; browser launch settings do not apply."
        ),
    ] = None,
    new_context: Annotated[
        bool,
        Field(
            description="With cdp_url, create and close a separate context. False reuses and preserves the default context, ignoring useragent, locale, timezone_id, proxy and additional_args."
        ),
    ] = True,
    executable_path: Annotated[
        Optional[str],
        Field(description="Absolute Chromium-compatible executable path; overrides the server default."),
    ] = None,
    cookies: Annotated[
        Sequence[SetCookieParam] | None, Field(description="Initial cookies as Playwright cookie dictionaries.")
    ] = None,
    hide_canvas: Annotated[bool, Field(description="Add noise to canvas operations.")] = False,
    block_webrtc: Annotated[bool, Field(description="Disable non-proxied WebRTC UDP to reduce IP leaks.")] = False,
    allow_webgl: Annotated[bool, Field(description="Enable WebGL; disabling it can trigger bot detection.")] = True,
    additional_args: Annotated[
        Optional[Dict], Field(description="Browser context options that override Scrapling settings.")
    ] = None,
) -> SessionCreatedModel:
    """Open a reusable local or remote browser session. Pass per-request options to `browser_fetch`."""
    session_id = self._new_session_id(session_id)
    options: Dict[str, Any] = dict(
        proxy=proxy,
        locale=locale,
        cookies=cookies,
        block_ads=True,
        record_requests=True,
        useragent=useragent,
        timezone_id=timezone_id,
        additional_args=additional_args,
    )
    session = (
        AsyncRemoteSession(cdp_url, new_context=new_context, **options)
        if cdp_url is not None
        else AsyncStealthySession(
            headless=headless,
            real_chrome=real_chrome,
            executable_path=self._resolve_executable_path(executable_path),
            hide_canvas=hide_canvas,
            block_webrtc=block_webrtc,
            allow_webgl=allow_webgl,
            **options,
        )
    )
    await session.start()
    return self._register_session(session_id, session, "stealthy")

async def browser_save_state(
    self,
    session_id: Annotated[str, Field(description="Browser session ID.")],
    path: Annotated[
        NonEmptyString,
        Field(
            description="File on the MCP server host; relative to its working directory. Overwrites the file; parent must exist."
        ),
    ],
) -> str:
    """Save browser cookies, local storage and IndexedDB as Playwright-compatible JSON. Excludes sessionStorage."""
    await self._get_session(session_id, ["stealthy"]).session.save_state(path)
    return "State saved."

async def browser_load_state(
    self,
    session_id: Annotated[str, Field(description="Browser session ID.")],
    path: Annotated[
        NonEmptyString, Field(description="File on the MCP server host; relative to its working directory.")
    ],
) -> str:
    """Replace browser cookies, local storage and IndexedDB from a Playwright-compatible JSON file.

    Leaves sessionStorage and open tabs unchanged. Fetch or reload pages to use the restored login."""
    await self._get_session(session_id, ["stealthy"]).session.load_state(path)
    return "State loaded."

async def open_request_session(
    self,
    session_id: Annotated[
        Optional[str], Field(description="Custom ID; a random 12-character hex ID if omitted.")
    ] = None,
    impersonate: Annotated[
        _MCPImpersonateType,
        Field(
            description='Browser/version to impersonate; "chrome" uses the latest supported Chrome profile. Lists pick randomly per attempt; None disables impersonation.'
        ),
    ] = "chrome",
    proxy: Annotated[
        Optional[str], Field(description="Proxy URL for all session requests, optionally including credentials.")
    ] = None,
) -> SessionCreatedModel:
    """Open a reusable HTTP session for `session_make_request`, keeping cookies, connections, and request settings."""
    session_id = self._new_session_id(session_id)
    session = FetcherSession(impersonate=impersonate, proxy=proxy)
    await session.__aenter__()
    return self._register_session(session_id, session, "static")

async def close_session(
    self,
    session_id: Annotated[str, Field(description="Session ID to close.")],
) -> SessionClosedModel:
    """Close a session and free its resources."""
    entry = self._sessions.pop(session_id, None)
    if entry is None:
        raise ValueError(f"Session '{session_id}' not found. Use list_sessions to see active sessions.")

    if entry.session_type == "static":
        await entry.session.__aexit__(None, None, None)
    else:
        await entry.session.close()
    return SessionClosedModel(
        session_id=session_id,
        message=f"Session '{session_id}' closed successfully.",
    )

async def list_sessions(self) -> List[SessionInfo]:
    """List registered sessions, their status, and effective settings."""
    return [
        SessionInfo(
            session_id=sid,
            session_type=entry.session_type,
            created_at=entry.created_at,
            is_alive=entry.session._is_alive,
            settings=_session_settings(entry.session),
        )
        for sid, entry in self._sessions.items()
    ]

async def browser_network_requests(
    self,
    session_id: Annotated[str, Field(description="Browser session ID.")],
    url_pattern: Annotated[Optional[str], Field(description="Filter URLs by regular expression.")] = None,
    include_static: Annotated[bool, Field(description="Include successful non-API requests too.")] = False,
    after_id: Annotated[
        int, Field(ge=0, description="Return requests with a higher ID; use next_cursor for more.")
    ] = 0,
    limit: Annotated[int, Field(ge=1, le=200, description="Maximum requests to return.")] = 50,
) -> NetworkRequestsModel:
    """List completed requests across page changes, with IDs and pagination. Recording starts when the browser session opens and closing does not wait for running captures."""
    network = self._get_session(session_id, ["stealthy"]).session.network
    records = network.search(
        url_pattern=url_pattern, include_static=include_static, after_id=after_id, limit=limit + 1
    )
    shown = records[:limit]
    return NetworkRequestsModel(
        requests=[NetworkRequestInfo.model_validate(record, from_attributes=True) for record in shown],
        next_cursor=shown[-1].meta["network_id"] if shown else max(after_id, network.last_id),
        has_more=len(records) > limit,
        dropped_count=network.dropped_count,
    )

async def browser_network_request(
    self,
    session_id: Annotated[str, Field(description="Browser session ID.")],
    request_id: Annotated[int, Field(ge=1, description="ID from `browser_network_requests`.")],
    part: Annotated[
        NetworkPart, Field(description="Request summary, headers, or text body; incomplete headers are marked.")
    ] = "summary",
) -> NetworkRequestModel:
    """Read a complete saved request part without resending it, with structured data and availability notes."""
    network = self._get_session(session_id, ["stealthy"]).session.network
    record = network.get(request_id)
    if record is None:
        raise ValueError(
            f"Request {request_id} was not found or is no longer retained. Use browser_network_requests."
        )
    return _request_details(record, part)

async def browser_extract(
    self,
    session_id: Annotated[str, Field(description="Browser session ID")],
    extraction_type: Annotated[
        SessionExtractionType, Field(description="Output format; defaults to an AI ARIA snapshot.")
    ] = "snapshot",
    target: Annotated[
        Optional[NonEmptyString],
        Field(description="Playwright selector or aria-ref=<ref> matching one element; omitted reads the page."),
    ] = None,
    depth: Annotated[
        Optional[int],
        Field(description="Snapshot depth before search; unlimited if omitted. Ignored for other formats."),
    ] = None,
    boxes: Annotated[
        bool, Field(description="Include viewport bounding boxes in snapshots; ignored for other formats.")
    ] = True,
    search: Annotated[
        Optional[NonEmptyString],
        Field(description="Snapshot-only Python regex; case-sensitive unless (?i). No matches returns empty text."),
    ] = None,
    main_content_only: Annotated[
        bool, Field(description="Clean body/target content for AI; ignored for snapshots.")
    ] = True,
) -> str:
    """Read the current page or one element as snapshot, HTML, Markdown, or text without navigating.
    Return plain text; snapshot search keeps refs, ancestor paths and three lines of nearby context."""
    pattern = None
    if search is not None:
        if extraction_type != "snapshot":
            raise ValueError("search requires snapshot output.")
        try:
            pattern = re.compile(search)
        except re.error as error:
            raise ValueError(f"Invalid search regex: {error}") from error
    if extraction_type == "snapshot":
        snapshot = await self._browser_snapshot(session_id, css_selector=target, depth=depth, boxes=boxes)
        return _search_snapshot(snapshot, pattern) if pattern is not None else snapshot
    with self._browser_page(session_id) as (_, page):
        content = (
            await page.locator(target).evaluate("(element) => element.outerHTML")
            if target is not None
            else await ResponseFactory._get_async_page_content(page)
        )
        return _CONTROL_CHARS_PATTERN.sub(
            "",
            "".join(
                Convertor._extract_content(
                    Selector(content=content, url=page.url),
                    extraction_type=extraction_type,
                    main_content_only=main_content_only,
                )
            ),
        )

async def browser_actions(
    self,
    session_id: Annotated[str, Field(description="Browser session ID")],
    actions: Annotated[
        List[BrowserAction],
        Field(
            min_length=1,
            description="Ordered actions; fields need target; move/click need target or viewport (x, y); mouse_down/mouse_up act at the current pointer.",
        ),
    ],
    slowly: Annotated[
        bool,
        Field(
            description="Fresh random delays: 50-150 ms per character, 100-300 ms between all actions, including waits."
        ),
    ] = False,
) -> str:
    """Chain all kinds of browser actions. Do actions without multiple calls or plan ahead to execute a sequence of actions.
    Run mouse, field, key, dialog, and wait actions on the current page in order.
    Load waits observe the current document; wait for a result element for delayed navigation.
    Stops on the first error, identifying its action; partial effects remain. No retries.
    Element mouse targets auto-wait and scroll into view; coordinates and wheel do not wait for navigation or scrolling.
    mouse_down holds across calls for observation and further moves; mouse_up releases, including after errors.
    For drag and drop, move over the target twice before releasing. Cancelled/timed-out clicks attempt release."""
    _validate_actions(actions)
    with self._browser_page(session_id) as (_, page):
        await _run_actions(page, actions, slowly)
    return "Actions completed."

async def browser_evaluate(
    self,
    session_id: Annotated[str, Field(description="Browser session ID")],
    expression: Annotated[NonEmptyString, Field(description="JavaScript expression or function to invoke.")],
    arg: Annotated[
        Optional[Dict[str, Any]],
        Field(description="JSON object passed to the function; use properties for scalar or array values."),
    ] = None,
    isolated_context: Annotated[
        bool,
        Field(
            description="False accesses the page's own JS variables; True uses a separate context with the same DOM."
        ),
    ] = True,
) -> str:
    """Run JavaScript on the current page, await promises, and return JSON text without a snapshot.
    Return JSON-compatible values; null/undefined become null. Script effects are not undone on failure.
    Cancellation stops waiting, but may not stop the script."""
    with self._browser_page(session_id) as (_, page):
        return dumps(
            await page.evaluate(expression, arg=arg, isolated_context=isolated_context),
            ensure_ascii=False,
            allow_nan=False,
            separators=(",", ":"),
        )

async def browser_screenshot(
    self,
    session_id: Annotated[str, Field(description="Browser session ID")],
    image_type: Annotated[ScreenshotType, Field(description="Image format.")] = "png",
    full_page: Annotated[
        bool, Field(description="Capture the full scrollable page; incompatible with selector/ref.")
    ] = False,
    quality: Annotated[
        Optional[Annotated[int, Field(ge=0, le=100)]], Field(description="JPEG quality, 0-100; invalid for PNG.")
    ] = None,
    timeout: Annotated[
        NonNegativeFiniteFloat, Field(description="Capture limit in milliseconds; 0 disables it.")
    ] = 30000,
    selector: Annotated[
        Optional[NonEmptyString],
        Field(description="Playwright selector matching one element; omit when using ref."),
    ] = None,
    ref: Annotated[
        Optional[NonEmptyString],
        Field(description="Current snapshot element reference; omit both targets for page capture."),
    ] = None,
) -> List[ImageContent | TextContent]:
    """Capture the current page or one element without navigating; return the image and current URL.
    Element capture waits and scrolls into view. Use browser_actions for other waits before capture."""
    if quality is not None and image_type != "jpeg":
        raise ValueError("'quality' is only valid when 'image_type' is 'jpeg'.")
    if selector is not None and ref is not None:
        raise ValueError("Provide either 'selector' or 'ref', not both.")
    if full_page and (selector is not None or ref is not None):
        raise ValueError("'full_page' cannot be combined with 'selector' or 'ref'.")

    with self._browser_page(session_id) as (_, page):
        options: Dict[str, Any] = {"type": image_type, "quality": quality, "timeout": timeout}
        if selector is not None or ref is not None:
            started = monotonic()
            element = await page.locator(selector or f"aria-ref={ref}").element_handle(timeout=timeout)
            try:
                options["timeout"] = max(1, timeout - (monotonic() - started) * 1000) if timeout else 0
                data = await element.screenshot(**options)
            except BaseException as exc:
                try:
                    with CancelScope(shield=True):
                        await element.dispose()
                except Exception as cleanup_error:
                    raise exc from cleanup_error
                raise
            with CancelScope(shield=True):
                await element.dispose()
        else:
            data = await page.screenshot(full_page=full_page, **options)
        return [Image(data=data, format=image_type).to_image_content(), TextContent(type="text", text=page.url)]

async def browser_observe(
    self,
    session_id: Annotated[str, Field(description="Browser session ID")],
    duration_ms: Annotated[int, Field(ge=100, le=5000, description="Observation window in milliseconds.")] = 1500,
    max_frames: Annotated[int, Field(ge=2, le=20, description="Maximum number of sampled frames.")] = 12,
) -> List[ImageContent | TextContent]:
    """Inspect browser page motion through a short sequence of timestamped viewport images. The page keeps running during and after capture.
    Use this tool when you get any challenge that require timed actions like games, appearing/disappearing CAPTCHAs, or interacting with any moving elements.
    """
    with self._browser_page(session_id) as (_, page):
        frames: Dict[int, Dict[str, Any]] = {}
        started = monotonic()

        def capture(frame: Dict[str, Any]) -> None:
            elapsed = (monotonic() - started) * 1000
            if elapsed < duration_ms:
                frames[int(elapsed * max_frames / duration_ms)] = frame

        try:
            await page.screencast.start(on_frame=capture, quality=60, size={"width": 800, "height": 800})
            await sleep(max(0, duration_ms / 1000 - (monotonic() - started)))
        except BaseException as error:
            try:
                await _stop_screencast(page)
            except Exception as cleanup_error:
                raise error from cleanup_error
            raise
        await _stop_screencast(page)
        if not frames:
            raise RuntimeError("No frames were captured. Try browser_screenshot or a longer observation window.")
        return [
            content
            for frame in frames.values()
            for content in (
                TextContent(
                    type="text",
                    text=dumps(
                        {
                            "timestamp_ms": frame["timestamp"],
                            "viewport_width": frame["viewportWidth"],
                            "viewport_height": frame["viewportHeight"],
                        },
                        separators=(",", ":"),
                    ),
                ),
                Image(data=frame["data"], format="jpeg").to_image_content(),
            )
        ]

@staticmethod
async def make_request(
    url: Annotated[str, Field(description="URL to fetch.")],
    method: Annotated[SUPPORTED_HTTP_METHODS, Field(description="HTTP method.")] = "GET",
    impersonate: Annotated[
        _MCPImpersonateType,
        Field(
            description='Browser/version to impersonate; "chrome" uses the latest supported Chrome profile. Lists pick randomly per attempt; None disables impersonation.'
        ),
    ] = "chrome",
    extraction_type: Annotated[extraction_types, Field(description="Content output format.")] = "markdown",
    css_selector: Annotated[
        Optional[str], Field(description="Select matching elements after `main_content_only` filtering.")
    ] = None,
    main_content_only: Annotated[
        bool, Field(description="Sanitize <body> content before selection; False uses the full document.")
    ] = True,
    params: Annotated[Optional[Dict], Field(description="Query parameters.")] = None,
    data: Annotated[Optional[Dict[str, str] | str], Field(description="Form or raw body; ignored for GET.")] = None,
    json: Annotated[Optional[Dict | List], Field(description="JSON body; ignored for GET.")] = None,
    headers: Annotated[Optional[Mapping[str, Optional[str]]], Field(description="Request headers.")] = None,
    cookies: Annotated[Optional[Dict[str, str]], Field(description="Request cookies.")] = None,
    timeout: Annotated[Optional[int | float], Field(description="Timeout in seconds.")] = 30,
    follow_redirects: Annotated[
        FollowRedirects,
        Field(
            description='"safe" blocks redirects to private/internal IPs; True allows all; False disables redirects.'
        ),
    ] = "safe",
    max_redirects: Annotated[int, Field(description="Redirect limit; -1 means unlimited.")] = 30,
    retries: Annotated[
        Optional[int], Field(description="Maximum attempts, including the first; None or values below 1 send once.")
    ] = 3,
    retry_delay: Annotated[Optional[int], Field(description="Seconds between retries.")] = 1,
    proxy: Annotated[Optional[str], Field(description="Proxy URL, optionally including credentials.")] = None,
    proxy_auth: Annotated[
        Optional[Dict[str, str]], Field(description='Proxy basic auth: {"username": ..., "password": ...}.')
    ] = None,
    auth: Annotated[
        Optional[Dict[str, str]], Field(description='Basic auth: {"username": ..., "password": ...}.')
    ] = None,
    verify: Annotated[Optional[bool], Field(description="Verify TLS certificates.")] = True,
    http3: Annotated[
        Optional[bool], Field(description="Use HTTP/3; may conflict with browser impersonation.")
    ] = False,
    stealthy_headers: Annotated[
        Optional[bool],
        Field(description="Fill missing browser headers and use a Google referer if none was supplied."),
    ] = True,
) -> ResponseModel:
    """Make an HTTP request without JavaScript. Suitable for low-to-mid protection."""
    normalized_proxy_auth = _normalize_credentials(proxy_auth)
    normalized_auth = _normalize_credentials(auth)

    request_kwargs: Dict[str, Any] = dict(
        auth=normalized_auth,
        proxy=proxy,
        http3=http3,
        verify=verify,
        params=params,
        headers=headers,
        cookies=cookies,
        timeout=timeout,
        retries=retries,
        proxy_auth=normalized_proxy_auth,
        retry_delay=retry_delay,
        impersonate=impersonate,
        max_redirects=max_redirects,
        follow_redirects=follow_redirects,
        stealthy_headers=stealthy_headers,
    )
    if method != "GET":
        request_kwargs.update(data=data, json=json)

    async with FetcherSession() as session:
        page = await getattr(session, method.lower())(url, **request_kwargs)
        return _translate_response(page, extraction_type, css_selector, main_content_only)

async def browser_fetch_once(
    self,
    url: Annotated[str, Field(description="URL to fetch.")],
    extraction_type: Annotated[extraction_types, Field(description="Content output format.")] = "markdown",
    css_selector: Annotated[
        Optional[str], Field(description="Select matching elements after `main_content_only` filtering.")
    ] = None,
    main_content_only: Annotated[
        bool, Field(description="Sanitize <body> content before selection; False uses the full document.")
    ] = True,
    headless: Annotated[bool, Field(description="Hide the browser window.")] = True,
    google_search: Annotated[
        bool, Field(description="Set a Google referer, overriding any supplied referer.")
    ] = True,
    real_chrome: Annotated[bool, Field(description="Use locally installed Chrome.")] = False,
    wait: Annotated[
        int | float, Field(description="Extra milliseconds after the page is ready, before returning.")
    ] = 0,
    proxy: Annotated[
        Optional[str | Dict[str, str]],
        Field(description="Proxy URL, or dictionary with server and optional username/password."),
    ] = None,
    timezone_id: Annotated[
        str | None, Field(description="Browser timezone; uses the system timezone if omitted.")
    ] = None,
    locale: Annotated[
        str | None,
        Field(description="Browser language, Accept-Language, and formatting locale; system default if omitted."),
    ] = None,
    extra_headers: Annotated[Optional[Dict[str, str]], Field(description="Additional request headers.")] = None,
    useragent: Annotated[
        Optional[str],
        Field(description="User-Agent override; otherwise generated in headless mode, native in headful mode."),
    ] = None,
    hide_canvas: Annotated[bool, Field(description="Add noise to canvas operations.")] = False,
    executable_path: Annotated[
        Optional[str],
        Field(description="Absolute Chromium-compatible executable path; overrides the server default."),
    ] = None,
    timeout: Annotated[
        int | float, Field(description="Navigation and page-operation timeout in milliseconds.")
    ] = 30000,
    disable_resources: Annotated[
        bool,
        Field(
            description="Block font, image, media, beacon, object, imageset, texttrack, websocket, csp_report, and stylesheet requests."
        ),
    ] = False,
    wait_selector: Annotated[
        Optional[str], Field(description="Wait for the first CSS match; continue if the wait fails.")
    ] = None,
    cookies: Annotated[
        Sequence[SetCookieParam] | None, Field(description="Initial cookies as Playwright cookie dictionaries.")
    ] = None,
    network_idle: Annotated[
        bool, Field(description="Try to wait for 500 ms without network activity; continue if the wait fails.")
    ] = False,
    wait_selector_state: Annotated[
        SelectorWaitStates, Field(description="Target state of `wait_selector`.")
    ] = "attached",
    block_webrtc: Annotated[bool, Field(description="Disable non-proxied WebRTC UDP to reduce IP leaks.")] = False,
    allow_webgl: Annotated[bool, Field(description="Enable WebGL; disabling it can trigger bot detection.")] = True,
    solve_cloudflare: Annotated[
        bool, Field(description="Attempt to solve Cloudflare Turnstile/interstitial challenges before returning.")
    ] = False,
    additional_args: Annotated[
        Optional[Dict], Field(description="Browser context options that override Scrapling settings.")
    ] = None,
    pierce_shadow: Annotated[bool, Field(description="Include open Shadow DOM content.")] = False,
) -> ResponseModel:
    """Fetch a page with a stealth browser and JavaScript rendering."""
    async with AsyncStealthySession(
        wait=wait,
        proxy=proxy,
        locale=locale,
        timeout=timeout,
        cookies=cookies,
        headless=headless,
        block_ads=True,
        max_pages=1,
        useragent=useragent,
        timezone_id=timezone_id,
        real_chrome=real_chrome,
        hide_canvas=hide_canvas,
        allow_webgl=allow_webgl,
        network_idle=network_idle,
        pierce_shadow=pierce_shadow,
        block_webrtc=block_webrtc,
        wait_selector=wait_selector,
        google_search=google_search,
        extra_headers=extra_headers,
        executable_path=self._resolve_executable_path(executable_path),
        additional_args=additional_args,
        solve_cloudflare=solve_cloudflare,
        disable_resources=disable_resources,
        wait_selector_state=wait_selector_state,
    ) as session:
        page = await session.fetch(url)

    return _translate_response(page, extraction_type, css_selector, main_content_only)