#!/usr/bin/env python3 """ Minimal stdio MCP server that exposes browser interaction tools. Launched as a subprocess by the Claude Agent SDK. Proxies tool calls to the OpenSwarm backend via HTTP, which bridges them to the Electron frontend via WebSocket where the actual webview lives. """ import base64 import json import sys import os import urllib.request import urllib.error from io import BytesIO try: from PIL import Image HAS_PIL = True except ImportError: HAS_PIL = False BACKEND_PORT = os.environ.get("OPENSWARM_PORT", "8324") BACKEND_URL = f"http://127.0.0.1:{BACKEND_PORT}/api/browser/command" TAB_ID_PROP = { "type": "string", "description": "Optional tab ID within the browser card. If omitted, targets the active tab.", } TOOLS = [ { "name": "BrowserScreenshot", "description": ( "Capture a screenshot of the browser page. Returns the screenshot as a " "base64-encoded PNG image. Use this to see what is currently displayed." ), "inputSchema": { "type": "object", "properties": { "browser_id": { "type": "string", "description": "The browser card ID to capture. Use the ID from the selected browser card context.", }, "tab_id": TAB_ID_PROP, }, "required": ["browser_id"], }, }, { "name": "BrowserGetText", "description": ( "Get the visible text content of the browser page. Returns the page's " "innerText (up to 15000 characters)." ), "inputSchema": { "type": "object", "properties": { "browser_id": { "type": "string", "description": "The browser card ID.", }, "tab_id": TAB_ID_PROP, }, "required": ["browser_id"], }, }, { "name": "BrowserNavigate", "description": "Navigate the browser to a URL.", "inputSchema": { "type": "object", "properties": { "browser_id": { "type": "string", "description": "The browser card ID.", }, "tab_id": TAB_ID_PROP, "url": { "type": "string", "description": "The URL to navigate to.", }, }, "required": ["browser_id", "url"], }, }, { "name": "BrowserClick", "description": ( "Click an element in the browser page identified by a CSS selector." ), "inputSchema": { "type": "object", "properties": { "browser_id": { "type": "string", "description": "The browser card ID.", }, "tab_id": TAB_ID_PROP, "selector": { "type": "string", "description": "CSS selector of the element to click.", }, }, "required": ["browser_id", "selector"], }, }, { "name": "BrowserType", "description": ( "Type text into an input element in the browser page. Clears the " "existing value first, then types the new text." ), "inputSchema": { "type": "object", "properties": { "browser_id": { "type": "string", "description": "The browser card ID.", }, "tab_id": TAB_ID_PROP, "selector": { "type": "string", "description": "CSS selector of the input element.", }, "text": { "type": "string", "description": "The text to type.", }, }, "required": ["browser_id", "selector", "text"], }, }, { "name": "BrowserEvaluate", "description": ( "Evaluate a JavaScript expression in the browser page and return the result. " "The expression is run via executeJavaScript on the webview." ), "inputSchema": { "type": "object", "properties": { "browser_id": { "type": "string", "description": "The browser card ID.", }, "tab_id": TAB_ID_PROP, "expression": { "type": "string", "description": "JavaScript expression to evaluate.", }, }, "required": ["browser_id", "expression"], }, }, { "name": "BrowserGetElements", "description": ( "Get a list of interactive elements on the page with their CSS selectors. " "Returns clickable elements, inputs, links, and buttons with selector paths " "you can use with BrowserClick and BrowserType. Call this BEFORE attempting " "to click or type so you know which selectors are valid." ), "inputSchema": { "type": "object", "properties": { "browser_id": { "type": "string", "description": "The browser card ID.", }, "tab_id": TAB_ID_PROP, "selector": { "type": "string", "description": ( "Optional CSS selector to scope the search " "(e.g. 'form', '#main'). Defaults to 'body'." ), }, }, "required": ["browser_id"], }, }, { "name": "BrowserScroll", "description": ( "Scroll the page up or down. Automatically finds the correct scrollable " "container (works on SPAs like Notion, Gmail, etc. that use nested scroll " "containers instead of window-level scrolling). Returns scroll position info " "including whether top/bottom has been reached." ), "inputSchema": { "type": "object", "properties": { "browser_id": { "type": "string", "description": "The browser card ID.", }, "tab_id": TAB_ID_PROP, "direction": { "type": "string", "enum": ["up", "down"], "description": "Scroll direction. Defaults to 'down'.", }, "amount": { "type": "number", "description": "Pixels to scroll. Defaults to 500.", }, }, "required": ["browser_id"], }, }, { "name": "BrowserWait", "description": ( "Wait for a specified duration. Useful after navigation or actions that " "trigger page loads, animations, or async content rendering. " "Min 100ms, max 10000ms." ), "inputSchema": { "type": "object", "properties": { "browser_id": { "type": "string", "description": "The browser card ID.", }, "tab_id": TAB_ID_PROP, "milliseconds": { "type": "number", "description": "Duration to wait in milliseconds. Defaults to 1000.", }, }, "required": ["browser_id"], }, }, ] def send_response(id_, result=None, error=None): msg = {"jsonrpc": "2.0", "id": id_} if error is not None: msg["error"] = error else: msg["result"] = result sys.stdout.write(json.dumps(msg) + "\n") sys.stdout.flush() def send_notification(method, params=None): msg = {"jsonrpc": "2.0", "method": method} if params is not None: msg["params"] = params sys.stdout.write(json.dumps(msg) + "\n") sys.stdout.flush() def call_backend(action: str, browser_id: str, params: dict | None = None, tab_id: str = "") -> dict: payload = json.dumps({ "action": action, "browser_id": browser_id, "tab_id": tab_id, "params": params or {}, }).encode() req = urllib.request.Request( BACKEND_URL, data=payload, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: body = e.read().decode() if e.fp else str(e) return {"error": f"HTTP {e.code}: {body}"} except Exception as e: return {"error": str(e)} MAX_IMAGE_B64_BYTES = 400_000 def compress_screenshot(b64_png: str) -> tuple[str, str] | None: """Resize and re-encode as JPEG to stay under the stdio buffer limit.""" if not HAS_PIL: return None try: raw = base64.b64decode(b64_png) img = Image.open(BytesIO(raw)) max_width = 1024 if img.width > max_width: ratio = max_width / img.width img = img.resize((max_width, int(img.height * ratio)), Image.LANCZOS) buf = BytesIO() img.convert("RGB").save(buf, format="JPEG", quality=45) return base64.b64encode(buf.getvalue()).decode(), "image/jpeg" except Exception: return None def handle_tool_call(tool_name: str, arguments: dict) -> dict: browser_id = arguments.get("browser_id", "") tab_id = arguments.get("tab_id", "") if not browser_id: return {"content": [{"type": "text", "text": "Error: browser_id is required"}], "isError": True} action_map = { "BrowserScreenshot": "screenshot", "BrowserGetText": "get_text", "BrowserNavigate": "navigate", "BrowserClick": "click", "BrowserType": "type", "BrowserEvaluate": "evaluate", "BrowserGetElements": "get_elements", "BrowserScroll": "scroll", "BrowserWait": "wait", } action = action_map.get(tool_name) if not action: return {"content": [{"type": "text", "text": f"Unknown tool: {tool_name}"}], "isError": True} params = {k: v for k, v in arguments.items() if k not in ("browser_id", "tab_id")} result = call_backend(action, browser_id, params, tab_id=tab_id) if "error" in result: return {"content": [{"type": "text", "text": f"Error: {result['error']}"}], "isError": True} if action == "screenshot" and result.get("image"): image_data = result["image"] mime_type = "image/png" if len(image_data) > MAX_IMAGE_B64_BYTES: compressed = compress_screenshot(image_data) if compressed: image_data, mime_type = compressed if len(image_data) > MAX_IMAGE_B64_BYTES: return { "content": [ {"type": "text", "text": ( f"Screenshot too large to return ({len(image_data)} bytes base64). " f"URL: {result.get('url', 'unknown')}. " "Use BrowserGetText to read the page content instead." )}, ], } return { "content": [ {"type": "image", "data": image_data, "mimeType": mime_type}, {"type": "text", "text": f"Screenshot captured. URL: {result.get('url', 'unknown')}"}, ], } text = result.get("text", result.get("data", json.dumps(result))) return {"content": [{"type": "text", "text": str(text)}]} def main(): for line in sys.stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except json.JSONDecodeError: continue method = msg.get("method") id_ = msg.get("id") params = msg.get("params", {}) if method == "initialize": send_response(id_, { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": { "name": "openswarm-browser", "version": "1.0.0", }, }) elif method == "notifications/initialized": pass elif method == "tools/list": send_response(id_, {"tools": TOOLS}) elif method == "tools/call": tool_name = params.get("name", "") arguments = params.get("arguments", {}) result = handle_tool_call(tool_name, arguments) send_response(id_, result) elif method == "ping": send_response(id_, {}) elif id_ is not None: send_response(id_, error={"code": -32601, "message": f"Method not found: {method}"}) if __name__ == "__main__": main()