Gemini Web Automation MCP

Instructions

Implementation Reference

• server.py:145-199 (handler)

  The primary handler function for the 'start_web_task' MCP tool. Decorated with @mcp.tool() for registration. Creates a task ID via task_manager and starts it in background thread, returning immediately with task_id.
  @mcp.tool() async def start_web_task(task: str, url: str = "https://www.google.com") -> dict[str, Any]: """ Start a web browsing task in the background and return immediately. Use this for tasks that might take a while (30+ seconds). The task runs asynchronously while you continue working. Check progress with check_web_task(). Args: task: What you want to accomplish on the web url: Starting webpage (defaults to Google) Returns: Dictionary containing: - ok: Boolean indicating task was started successfully - task_id: Unique ID to check progress later - status: Will be "running" - message: Instructions for checking progress Examples: - start_web_task("Research top 10 AI companies and their products") - start_web_task("Find and compare prices for MacBook Pro on 5 different sites") Next steps: Use check_web_task(task_id) to monitor progress. Wait at least 5 seconds between status checks. """ logger.info(f"Starting async web browsing task: {task}") # Create task task_id = task_manager.create_task(task, url) # Start task in background using anyio (FastMCP best practice) # Use anyio.to_thread.run_sync to run the blocking start_task in a thread # We await it but start_task itself just spawns the thread and returns immediately success = await anyio.to_thread.run_sync( task_manager.start_task, task_id, logger ) if not success: return { "ok": False, "error": "Failed to start task" } logger.info(f"Task {task_id} started in background, returning immediately") return { "ok": True, "task_id": task_id, "status": "running", "message": f"Task started. Use check_web_task('{task_id}') to monitor progress." }
• task_manager.py:100-116 (helper)

  Helper method in BrowserTaskManager to create a new task instance with PENDING status and generate unique task_id.
  def create_task(self, task_description: str, url: str = "https://www.google.com") -> str: """Create a new browser automation task. Args: task_description: Description of the browsing task url: Starting URL Returns: task_id: Unique identifier for the task """ task_id = str(uuid.uuid4()) task = BrowserTask(task_id, task_description, url) with self._lock: self.tasks[task_id] = task return task_id
• task_manager.py:118-148 (helper)

  Helper method that transitions task to RUNNING status and spawns a daemon thread to execute _execute_task.
  def start_task(self, task_id: str, logger=None) -> bool: """Start executing a task in the background. Args: task_id: Task identifier logger: Optional logger instance Returns: True if task started, False if task not found or already running """ with self._lock: task = self.tasks.get(task_id) if not task or task.status != TaskStatus.PENDING: return False task.status = TaskStatus.RUNNING task.started_at = datetime.now(timezone.utc).isoformat() # Run task in background thread (don't store reference) thread = threading.Thread( target=self._execute_task, args=(task_id, logger), daemon=True, name=f"BrowserTask-{task_id[:8]}" ) thread.start() if logger: logger.info(f"Started background thread for task {task_id[:8]}... Thread: {thread.name}") return True
• task_manager.py:150-186 (helper)

  Private helper that runs in background thread: creates GeminiBrowserAgent, executes the web task, updates result and status, cleans up.
  def _execute_task(self, task_id: str, logger=None): """Execute the browser automation task (runs in background thread).""" if logger: logger.info(f"[Thread {threading.current_thread().name}] Starting execution for task {task_id[:8]}...") with self._lock: task = self.tasks.get(task_id) if not task: if logger: logger.error(f"Task {task_id} not found in _execute_task") return try: # Create browser agent agent = GeminiBrowserAgent(logger=logger) task.agent = agent # Execute the task result = agent.execute_task(task.task_description, task.url) with self._lock: task.result = result task.progress_updates = agent.progress_updates.copy() task.status = TaskStatus.COMPLETED task.completed_at = datetime.now(timezone.utc).isoformat() except Exception as e: with self._lock: task.error = str(e) task.status = TaskStatus.FAILED task.completed_at = datetime.now(timezone.utc).isoformat() finally: # Clean up browser if task.agent: task.agent.cleanup_browser()
• server.py:147-171 (schema)

  Docstring defining the input parameters, output format, and usage examples for the start_web_task tool.
  """ Start a web browsing task in the background and return immediately. Use this for tasks that might take a while (30+ seconds). The task runs asynchronously while you continue working. Check progress with check_web_task(). Args: task: What you want to accomplish on the web url: Starting webpage (defaults to Google) Returns: Dictionary containing: - ok: Boolean indicating task was started successfully - task_id: Unique ID to check progress later - status: Will be "running" - message: Instructions for checking progress Examples: - start_web_task("Research top 10 AI companies and their products") - start_web_task("Find and compare prices for MacBook Pro on 5 different sites") Next steps: Use check_web_task(task_id) to monitor progress. Wait at least 5 seconds between status checks. """