browser-cli/browser_cli/mcp/server.py

"""Stateless MCP server exposing a conservative browser-cli tool surface.

The MCP process stores no browser client, tab ID, or navigation state. Every
call constructs a fresh :class:`browser_cli.BrowserCLI`; the real browser is
the sole owner of browser state.
"""
from __future__ import annotations

import argparse
import base64
import json
import os
from collections.abc import Callable
from typing import Any, Literal

from browser_cli import BrowserCLI
from browser_cli.mcp.naming import resolve_tool_prefix, tool_name
from browser_cli.mcp.serialization import structured
from browser_cli.mcp.targets import resolve_tab_id
from browser_cli.url_match import find_reusable_tab, needs_navigation, tab_id as read_tab_id, tab_url
from browser_cli.version_manager import project_version

ClientFactory = Callable[..., BrowserCLI]

_SERVER_INSTRUCTIONS = """Control a real, user-visible browser through browser-cli.
The server is stateless: pass browser, remote, and key on each tool call when a
specific target is required. Tool calls affect the user's actual browser. Read
current tabs/page state instead of assuming IDs or content from an earlier call.
Reuse before opening: tabs_open takes over an already-open tab whose URL is
nearly the same (same site and page, or any tab of that site when you ask for
its root) and reports reused=true with that tab's ID. Pass reuse="none" only
when a separate tab is really wanted, and reuse="origin" to take over any tab
of the site. Use tabs_list(search=...) first when you want to pick the tab.
"""

def _client(factory: ClientFactory, browser: str | None, remote: str | None, key: str | None) -> BrowserCLI:
  """Build a fresh client, making MCP process environment defaults explicit.

  Explicit values are important for SDK multi-browser routing: leaving
  ``browser=None`` would fan out list/count calls before lower transport code
  gets a chance to consult ``BROWSER_CLI_PROFILE``.
  """
  return factory(
    browser=browser or os.environ.get("BROWSER_CLI_PROFILE"),
    remote=remote or os.environ.get("BROWSER_CLI_REMOTE"),
    key=key or os.environ.get("BROWSER_CLI_KEY"),
  )

def _screenshot_bytes(data_url: str) -> tuple[bytes, str]:
  """Decode a browser screenshot data URL into bytes and an MCP image format."""
  header, separator, payload = data_url.partition(",")
  if not separator or ";base64" not in header:
    raise ValueError("Browser returned an invalid screenshot data URL")
  media_type = header[5:].split(";", 1)[0].lower()
  image_format = "jpeg" if media_type in {"image/jpeg", "image/jpg"} else "png"
  return base64.b64decode(payload, validate=True), image_format

def create_server(*, client_factory: ClientFactory = BrowserCLI, tool_prefix: str | None = None):
  """Create the MCP server. Supplying *client_factory* keeps tests browser-free."""
  try:
    from mcp.server import MCPServer
    from mcp.server.mcpserver import Image
  except ImportError as exc:  # pragma: no cover - exercised without the optional extra
    raise RuntimeError(
      "MCP support is not installed. Install real-browser-cli with the 'mcp' extra: "
      "uv tool install 'real-browser-cli[mcp]'"
    ) from exc

  prefix = resolve_tool_prefix() if tool_prefix is None else tool_prefix
  mcp = MCPServer(
    "browser-cli",
    description="Control a real running browser through the browser-cli SDK.",
    instructions=_SERVER_INSTRUCTIONS,
    version=project_version(),
  )

  @mcp.tool(name=tool_name("tabs_list", prefix))
  def tabs_list(
    search: str | None = None,
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> list[dict[str, Any]]:
    """List current tabs, optionally filtering by URL or title substring.

    Check this before opening anything so an already-open page gets reused.
    """
    tabs = structured(_client(client_factory, browser, remote, key).tabs.list())
    if not search:
      return tabs
    needle = search.lower()
    return [
      tab for tab in tabs
      if needle in str(tab.get("url") or "").lower()
      or needle in str(tab.get("title") or "").lower()
    ]

  @mcp.tool(name=tool_name("tabs_open", prefix))
  def tabs_open(
    url: str,
    wait: bool = False,
    timeout: float = 30.0,
    background: bool = False,
    focus: bool = False,
    reuse: Literal["similar", "exact", "origin", "none"] = "similar",
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> dict[str, Any]:
    """Open a URL in the real browser, reusing an already-open tab with a nearly identical URL.

    Returns the tab metadata plus ``reused``. ``reuse`` selects the match:
    ``similar`` (default: same site and page, or any tab of that site when the
    requested URL is its root), ``exact``, ``origin`` (any tab of the site), or
    ``none`` to always create a new tab.
    """
    client = _client(client_factory, browser, remote, key)
    existing = find_reusable_tab(structured(client.tabs.list()), url, reuse)
    if existing is not None:
      target = read_tab_id(existing)
      if needs_navigation(tab_url(existing), url):
        client.nav.to(target, url)
      if focus:
        client.tabs.activate(target)
      return {**structured(client.tabs.status(target)), "reused": True}
    tab = client.tabs.open(url, wait=wait, timeout=timeout, background=background, focus=focus)
    return {**structured(tab), "reused": False}

  @mcp.tool(name=tool_name("tabs_close", prefix))
  def tabs_close(
    tab_id: int | None = None,
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> dict[str, int]:
    """Close a tab, defaulting to the active tab. This changes the real browser."""
    client = _client(client_factory, browser, remote, key)
    target = resolve_tab_id(client, tab_id)
    return {"closed": client.tabs.close(target), "tab_id": target}

  @mcp.tool(name=tool_name("navigate", prefix))
  def navigate(
    url: str,
    tab_id: int | None = None,
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> dict[str, Any]:
    """Navigate a tab to a URL, defaulting to the active tab, and return it."""
    client = _client(client_factory, browser, remote, key)
    target = resolve_tab_id(client, tab_id)
    client.nav.to(target, url)
    return structured(client.tabs.status(target))

  @mcp.tool(name=tool_name("page_info", prefix))
  def page_info(
    tab_id: int | None = None,
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> dict[str, Any]:
    """Return title, URL, readiness, language, and metadata for the active page or tab_id."""
    client = _client(client_factory, browser, remote, key)
    if tab_id is not None:
      return structured(client.tabs.status(tab_id))
    return structured(client.page.info())

  @mcp.tool(name=tool_name("extract_text", prefix))
  def extract_text(
    tab_id: int | None = None,
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> str:
    """Extract plain text from the active page or tab_id."""
    return _client(client_factory, browser, remote, key).extract.text(tab_id)

  @mcp.tool(name=tool_name("extract_markdown", prefix))
  def extract_markdown(
    selector: str | None = None,
    tab_id: int | None = None,
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> str:
    """Extract clean Markdown from the active page or tab_id, optionally scoped by CSS selector."""
    return _client(client_factory, browser, remote, key).extract.markdown(selector, tab_id)

  @mcp.tool(name=tool_name("dom_query", prefix))
  def dom_query(
    selector: str,
    tab_id: int | None = None,
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> list[dict[str, Any]]:
    """Return elements matching a CSS selector on the active page or tab_id."""
    client = _client(client_factory, browser, remote, key)
    if tab_id is None:
      return structured(client.dom.query(selector))
    code = f"""
      (() => {{
        const selector = {json.dumps(selector)};
        return [...document.querySelectorAll(selector)].map(el => ({{
          tag: el.tagName,
          text: (el.innerText || el.textContent || '').replace(/\\s+/g, ' ').trim(),
          attrs: Object.fromEntries([...el.attributes].map(attr => [attr.name, attr.value])),
        }}));
      }})()
    """
    return structured(client.dom.eval(code, tab_id))

  @mcp.tool(name=tool_name("dom_click", prefix))
  def dom_click(
    selector: str,
    tab_id: int | None = None,
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> dict[str, Any]:
    """Click the first matching element, then return current page info."""
    client = _client(client_factory, browser, remote, key)
    if tab_id is None:
      client.dom.click(selector)
      return structured(client.page.info())
    code = f"""
      (() => {{
        const selector = {json.dumps(selector)};
        const el = document.querySelector(selector);
        if (!el) throw new Error(`No element found for selector: ${{selector}}`);
        el.click();
        return true;
      }})()
    """
    client.dom.eval(code, tab_id)
    return structured(client.page.info(tab_id))

  @mcp.tool(name=tool_name("dom_type", prefix))
  def dom_type(
    selector: str,
    text: str,
    tab_id: int | None = None,
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> dict[str, bool]:
    """Type text into the first element matching a CSS selector."""
    client = _client(client_factory, browser, remote, key)
    if tab_id is None:
      client.dom.type(selector, text)
      return {"typed": True}
    code = f"""
      (() => {{
        const selector = {json.dumps(selector)};
        const text = {json.dumps(text)};
        const el = document.querySelector(selector);
        if (!el) throw new Error(`No element found for selector: ${{selector}}`);
        el.focus();
        el.value = text;
        el.dispatchEvent(new InputEvent('input', {{ bubbles: true, inputType: 'insertText', data: text }}));
        el.dispatchEvent(new Event('change', {{ bubbles: true }}));
        return true;
      }})()
    """
    client.dom.eval(code, tab_id)
    return {"typed": True}

  @mcp.tool(name=tool_name("screenshot", prefix), structured_output=False)
  def screenshot(
    tab_id: int | None = None,
    format: Literal["png", "jpeg"] = "png",
    quality: int | None = None,
    browser: str | None = None,
    remote: str | None = None,
    key: str | None = None,
  ) -> Any:
    """Capture the visible area of the active or specified tab as an image."""
    data_url = _client(client_factory, browser, remote, key).tabs.screenshot(
      tab_id, format=format, quality=quality
    )
    data, actual_format = _screenshot_bytes(data_url)
    return Image(data=data, format=actual_format)

  return mcp

def _parser() -> argparse.ArgumentParser:
  parser = argparse.ArgumentParser(description="Run the stateless browser-cli MCP server.")
  parser.add_argument("--transport", choices=("stdio", "streamable-http"), default="stdio")
  parser.add_argument("--host", default="127.0.0.1", help="HTTP bind host (streamable-http only).")
  parser.add_argument("--port", type=int, default=8000, help="HTTP bind port (streamable-http only).")
  parser.add_argument("--path", default="/mcp", help="MCP endpoint path (streamable-http only).")
  return parser

def main(argv: list[str] | None = None) -> None:
  """Run over stdio, or stateless Streamable HTTP when explicitly selected."""
  args = _parser().parse_args(argv)
  mcp = create_server()
  if args.transport == "stdio":
    mcp.run()
    return
  if args.host not in {"127.0.0.1", "localhost", "::1"}:
    raise SystemExit(
      "Refusing to expose the unauthenticated MCP server beyond localhost. "
      "Use browser-cli's authenticated remote transport from a local MCP server instead."
    )
  mcp.run(
    transport="streamable-http",
    host=args.host,
    port=args.port,
    streamable_http_path=args.path,
    stateless_http=True,
    json_response=True,
  )

if __name__ == "__main__":
  main()