This repository has no description
13 kB
355 lines
1#!/usr/bin/env python3
2"""Control the user's everyday Firefox from MCP clients without WebDriver.
3
4Two roles:
5
6host Native messaging host, spawned by Firefox when the Browser MCP
7 extension connects. Relays newline-delimited JSON requests from a
8 unix socket to the extension and routes replies back. Any number of
9 clients can talk to it at once.
10serve stdio MCP server. Connects to the host's socket per tool call, so it
11 keeps working across Firefox restarts without restarting itself.
12
13Neither role touches Marionette or remote debugging, so the browser never
14reports navigator.webdriver = true and captcha providers see a normal browser.
15"""
16
17from __future__ import annotations
18
19import asyncio
20import base64
21import contextlib
22import itertools
23import json
24import os
25import select
26import socket
27import struct
28import sys
29from pathlib import Path
30from typing import Any
31
32from mcp.server.fastmcp import FastMCP, Image
33
34PROTOCOL_LIMIT = 64 * 1024 * 1024 # screenshots can be several MB
35
36
37def socket_path() -> Path:
38 """Socket shared by host and server; both run as the same user."""
39 if override := os.environ.get("BROWSER_MCP_SOCKET"):
40 return Path(override)
41 if runtime := os.environ.get("XDG_RUNTIME_DIR"):
42 return Path(runtime) / "browser-mcp.sock"
43 return Path.home() / ".cache" / "browser-mcp" / "browser-mcp.sock"
44
45
46# ── host ────────────────────────────────────────────────────────────────
47
48
49def read_exact(fd: int, n: int) -> bytes | None:
50 buf = b""
51 while len(buf) < n:
52 chunk = os.read(fd, n - len(buf))
53 if not chunk:
54 return None
55 buf += chunk
56 return buf
57
58
59def read_native(fd: int) -> dict[str, Any] | None:
60 header = read_exact(fd, 4)
61 if header is None:
62 return None
63 (length,) = struct.unpack("=I", header)
64 body = read_exact(fd, length)
65 if body is None:
66 return None
67 msg: dict[str, Any] = json.loads(body)
68 return msg
69
70
71def write_native(msg: dict[str, Any]) -> None:
72 data = json.dumps(msg).encode()
73 sys.stdout.buffer.write(struct.pack("=I", len(data)) + data)
74 sys.stdout.buffer.flush()
75
76
77def bind_socket(path: Path) -> socket.socket:
78 """Bind the relay socket, taking it over from a previous host.
79
80 The latest Firefox to start (e.g. after a restart while the old host
81 lingers) owns the socket.
82 """
83 path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
84 with contextlib.suppress(FileNotFoundError):
85 path.unlink()
86 old_umask = os.umask(0o177)
87 try:
88 listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
89 listener.bind(str(path))
90 finally:
91 os.umask(old_umask)
92 listener.listen()
93 listener.setblocking(False)
94 return listener
95
96
97class Client:
98 def __init__(self, conn: socket.socket) -> None:
99 self.conn = conn
100 self.buf = b""
101
102
103def run_host() -> None:
104 path = socket_path()
105 listener = bind_socket(path)
106 inode = path.stat().st_ino
107 stdin_fd = sys.stdin.buffer.fileno()
108 clients: dict[int, Client] = {}
109 # extension request id -> (client fd, client's own request id)
110 pending: dict[int, tuple[int, Any]] = {}
111 ids = itertools.count(1)
112
113 def drop(fd: int) -> None:
114 client = clients.pop(fd)
115 client.conn.close()
116 for rid in [r for r, (cfd, _) in pending.items() if cfd == fd]:
117 del pending[rid]
118
119 def reply(fd: int, msg: dict[str, Any]) -> None:
120 client = clients.get(fd)
121 if client is None:
122 return
123 client.conn.setblocking(True)
124 try:
125 client.conn.sendall(json.dumps(msg).encode() + b"\n")
126 except OSError:
127 drop(fd)
128 else:
129 client.conn.setblocking(False)
130
131 try:
132 while True:
133 readable, _, _ = select.select(
134 [stdin_fd, listener.fileno(), *clients], [], []
135 )
136 for fd in readable:
137 if fd == stdin_fd:
138 msg = read_native(stdin_fd)
139 if msg is None:
140 return # Firefox closed the port
141 target = pending.pop(msg.get("id", -1), None)
142 if target is not None:
143 cfd, cid = target
144 reply(cfd, {**msg, "id": cid})
145 elif fd == listener.fileno():
146 conn, _ = listener.accept()
147 conn.setblocking(False)
148 clients[conn.fileno()] = Client(conn)
149 else:
150 client = clients[fd]
151 try:
152 chunk = client.conn.recv(65536)
153 except BlockingIOError:
154 continue
155 except OSError:
156 chunk = b""
157 if not chunk:
158 drop(fd)
159 continue
160 client.buf += chunk
161 while b"\n" in client.buf:
162 line, client.buf = client.buf.split(b"\n", 1)
163 try:
164 req = json.loads(line)
165 except json.JSONDecodeError as e:
166 reply(fd, {"id": None, "error": f"bad request: {e}"})
167 continue
168 rid = next(ids)
169 pending[rid] = (fd, req.get("id"))
170 write_native(
171 {
172 "id": rid,
173 "method": req.get("method"),
174 "params": req.get("params", {}),
175 }
176 )
177 finally:
178 listener.close()
179 # Only remove the socket if a newer host has not taken it over.
180 with contextlib.suppress(FileNotFoundError):
181 if path.stat().st_ino == inode:
182 path.unlink()
183
184
185# ── MCP server ──────────────────────────────────────────────────────────
186
187
188class BrowserError(Exception):
189 pass
190
191
192async def call(method: str, params: dict[str, Any] | None = None) -> Any:
193 try:
194 reader, writer = await asyncio.open_unix_connection(
195 str(socket_path()), limit=PROTOCOL_LIMIT
196 )
197 except (FileNotFoundError, ConnectionRefusedError) as e:
198 raise BrowserError(
199 "Firefox is not connected: the Browser MCP extension is not "
200 "installed/enabled or Firefox is not running"
201 ) from e
202 try:
203 req = {"id": 1, "method": method, "params": params or {}}
204 writer.write(json.dumps(req).encode() + b"\n")
205 await writer.drain()
206 line = await reader.readline()
207 finally:
208 writer.close()
209 if not line:
210 raise BrowserError("Firefox disconnected before replying")
211 resp: dict[str, Any] = json.loads(line)
212 if "error" in resp:
213 raise BrowserError(resp["error"])
214 return resp.get("result")
215
216
217def as_text(value: Any) -> str:
218 return value if isinstance(value, str) else json.dumps(value, indent=1)
219
220
221def run_server() -> None:
222 mcp = FastMCP(
223 "browser",
224 instructions=(
225 "Controls the user's own running Firefox through an extension "
226 "(no WebDriver, so sites don't flag it as automated). Tabs are "
227 "addressed by tab_id from list_tabs/open_tab. Open new tabs in "
228 "the background unless the user should see them. For "
229 "interaction, take a snapshot to get element refs, then click, "
230 "fill or press_key by ref. Refs are invalid after the page "
231 "navigates. Events are synthetic: captchas and other checks that "
232 "require a real user gesture must be solved by the user."
233 ),
234 )
235
236 @mcp.tool()
237 async def list_tabs() -> str:
238 """List all open tabs: id, windowId, active, status, title, url."""
239 return as_text(await call("list_tabs"))
240
241 @mcp.tool()
242 async def open_tab(url: str, active: bool = False, timeout_ms: int = 15000) -> str:
243 """Open url in a new tab (in the background unless active) and wait for it to load."""
244 return as_text(
245 await call(
246 "open_tab", {"url": url, "active": active, "timeout_ms": timeout_ms}
247 )
248 )
249
250 @mcp.tool()
251 async def navigate(tab_id: int, url: str, timeout_ms: int = 15000) -> str:
252 """Load url in an existing tab and wait for it to load."""
253 return as_text(
254 await call(
255 "navigate", {"tab_id": tab_id, "url": url, "timeout_ms": timeout_ms}
256 )
257 )
258
259 @mcp.tool()
260 async def navigate_history(
261 tab_id: int, direction: str, timeout_ms: int = 15000
262 ) -> str:
263 """Go 'back' or 'forward' in a tab's history and wait for the load."""
264 return as_text(
265 await call(
266 "history",
267 {"tab_id": tab_id, "direction": direction, "timeout_ms": timeout_ms},
268 )
269 )
270
271 @mcp.tool()
272 async def reload(tab_id: int, timeout_ms: int = 15000) -> str:
273 """Reload a tab and wait for it to load."""
274 return as_text(
275 await call("reload", {"tab_id": tab_id, "timeout_ms": timeout_ms})
276 )
277
278 @mcp.tool()
279 async def activate_tab(tab_id: int) -> str:
280 """Bring a tab to the front and focus its window."""
281 return as_text(await call("activate_tab", {"tab_id": tab_id}))
282
283 @mcp.tool()
284 async def close_tab(tab_id: int) -> str:
285 """Close a tab."""
286 return as_text(await call("close_tab", {"tab_id": tab_id}))
287
288 @mcp.tool()
289 async def snapshot(tab_id: int, max_elements: int = 300) -> str:
290 """List the visible interactive elements and headings of a tab with refs.
291
292 Lines look like `[12] button "Sign in"`; pass the number as ref to
293 click, fill or press_key. Covers open shadow roots, not iframes.
294 """
295 return as_text(
296 await call("snapshot", {"tab_id": tab_id, "max_elements": max_elements})
297 )
298
299 @mcp.tool()
300 async def get_text(tab_id: int, max_length: int = 20000) -> str:
301 """Visible text of a tab (document.body.innerText), truncated to max_length."""
302 return as_text(await call("text", {"tab_id": tab_id, "max_length": max_length}))
303
304 @mcp.tool()
305 async def click(tab_id: int, ref: int) -> str:
306 """Click an element from the latest snapshot."""
307 return as_text(await call("click", {"tab_id": tab_id, "ref": ref}))
308
309 @mcp.tool()
310 async def fill(tab_id: int, ref: int, value: str) -> str:
311 """Replace the value of an input/textarea/contenteditable, pick a select
312 option (by value or text), or set a checkbox/radio ("true"/"false")."""
313 return as_text(
314 await call("fill", {"tab_id": tab_id, "ref": ref, "value": value})
315 )
316
317 @mcp.tool()
318 async def press_key(tab_id: int, key: str, ref: int | None = None) -> str:
319 """Dispatch a key (DOM key name, e.g. "Enter", "Escape", "a") to ref or
320 the focused element. Enter in a form input submits the form."""
321 return as_text(
322 await call("press_key", {"tab_id": tab_id, "key": key, "ref": ref})
323 )
324
325 @mcp.tool()
326 async def evaluate(tab_id: int, code: str) -> str:
327 """Run JavaScript in a tab and return the JSON of its result.
328
329 code is the body of an async function: use `return`. It runs in the
330 extension's isolated content-script world with full DOM access;
331 page-defined globals are reachable via window.wrappedJSObject.
332 """
333 return as_text(await call("evaluate", {"tab_id": tab_id, "code": code}))
334
335 @mcp.tool()
336 async def screenshot(tab_id: int) -> Image:
337 """PNG of a tab's visible viewport; works for background tabs too."""
338 data_url = await call("screenshot", {"tab_id": tab_id})
339 _, _, b64 = str(data_url).partition(",")
340 return Image(data=base64.b64decode(b64), format="png")
341
342 mcp.run()
343
344
345def main() -> None:
346 # Firefox starts the host with the manifest path and extension id as
347 # arguments; the MCP server is started without arguments.
348 if len(sys.argv) > 1 and sys.argv[1] == "host":
349 run_host()
350 else:
351 run_server()
352
353
354if __name__ == "__main__":
355 main()