Add optional stateless MCP server for the real browser

Expose a conservative browser-cli tool surface to MCP hosts without
duplicating browser-control logic: every tool is a thin adapter over the
existing Python SDK, and each call builds a fresh BrowserCLI so the real
browser stays the only owner of tab and navigation state.

The MCP process resolves BROWSER_CLI_PROFILE, BROWSER_CLI_REMOTE, and
BROWSER_CLI_KEY explicitly instead of passing None down, so a pinned server
targets one browser rather than fanning listing calls out across every
connected browser. Explicit tool arguments still win.

Tool names keep a browser_ prefix for hosts that expose raw MCP names, while
BROWSER_CLI_MCP_TOOL_PREFIX lets hosts that already namespace by server drop
it and avoid names like browser_cli_testing_browser_tabs_list.

JavaScript evaluation, raw commands, storage writes, and session import stay
unexposed, and Streamable HTTP refuses non-loopback binds because the endpoint
has no authentication of its own; remote browsers go through browser-cli's
authenticated remote transport instead.

MCP stays an optional extra, so normal installs are unaffected.

browser_cli/mcp/server.py

@@ -0,0 +1,228 @@
+"""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 os
+from collections.abc import Callable
+from typing import Any, Literal
+from urllib.parse import urlsplit
+
+from browser_cli import BrowserCLI
+from browser_cli.mcp.naming import resolve_tool_prefix, tool_name
+from browser_cli.mcp.serialization import structured
+
+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.
+"""
+
+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,
+  )
+
+  @mcp.tool(name=tool_name("tabs_list", prefix))
+  def tabs_list(
+    browser: str | None = None,
+    remote: str | None = None,
+    key: str | None = None,
+  ) -> list[dict[str, Any]]:
+    """List current tabs. Optionally target a browser alias or authenticated remote."""
+    return structured(_client(client_factory, browser, remote, key).tabs.list())
+
+  @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,
+    browser: str | None = None,
+    remote: str | None = None,
+    key: str | None = None,
+  ) -> dict[str, Any]:
+    """Open a URL in a new real-browser tab and return its current metadata."""
+    tab = _client(client_factory, browser, remote, key).tabs.open(
+      url, wait=wait, timeout=timeout, background=background, focus=focus
+    )
+    return structured(tab)
+
+  @mcp.tool(name=tool_name("tabs_close", prefix))
+  def tabs_close(
+    tab_id: int,
+    browser: str | None = None,
+    remote: str | None = None,
+    key: str | None = None,
+  ) -> dict[str, int]:
+    """Close one tab by its current ID. This changes the user's real browser."""
+    closed = _client(client_factory, browser, remote, key).tabs.close(tab_id)
+    return {"closed": closed}
+
+  @mcp.tool(name=tool_name("navigate", prefix))
+  def navigate(
+    tab_id: int,
+    url: str,
+    browser: str | None = None,
+    remote: str | None = None,
+    key: str | None = None,
+  ) -> dict[str, Any]:
+    """Navigate an existing tab to a URL and return a fresh tab snapshot."""
+    client = _client(client_factory, browser, remote, key)
+    client.nav.to(tab_id, url)
+    return structured(client.tabs.status(tab_id))
+
+  @mcp.tool(name=tool_name("page_info", prefix))
+  def page_info(
+    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."""
+    return structured(_client(client_factory, browser, remote, key).page.info())
+
+  @mcp.tool(name=tool_name("extract_text", prefix))
+  def extract_text(
+    browser: str | None = None,
+    remote: str | None = None,
+    key: str | None = None,
+  ) -> str:
+    """Extract plain text from the active page."""
+    return _client(client_factory, browser, remote, key).extract.text()
+
+  @mcp.tool(name=tool_name("extract_markdown", prefix))
+  def extract_markdown(
+    selector: str | None = None,
+    browser: str | None = None,
+    remote: str | None = None,
+    key: str | None = None,
+  ) -> str:
+    """Extract clean Markdown from the active page or an optional CSS selector."""
+    return _client(client_factory, browser, remote, key).extract.markdown(selector)
+
+  @mcp.tool(name=tool_name("dom_query", prefix))
+  def dom_query(
+    selector: str,
+    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."""
+    return structured(_client(client_factory, browser, remote, key).dom.query(selector))
+
+  @mcp.tool(name=tool_name("dom_click", prefix))
+  def dom_click(
+    selector: str,
+    browser: str | None = None,
+    remote: str | None = None,
+    key: str | None = None,
+  ) -> dict[str, Any]:
+    """Click the first matching element, then return current active-page info."""
+    client = _client(client_factory, browser, remote, key)
+    client.dom.click(selector)
+    return structured(client.page.info())
+
+  @mcp.tool(name=tool_name("dom_type", prefix))
+  def dom_type(
+    selector: str,
+    text: str,
+    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_factory, browser, remote, key).dom.type(selector, text)
+    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()