Gemini Web Automation MCP

Instructions

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.

Implementation Reference

• server.py:145-199 (handler)

  The primary handler for the 'start_web_task' tool. This function is registered via the @mcp.tool() decorator. It creates a background task using task_manager.create_task and initiates execution asynchronously via task_manager.start_task, returning a task_id immediately for later polling.

  @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 that creates a new BrowserTask instance and stores it, returning the unique task_id. Called by start_web_task.

  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 the task to RUNNING status and spawns a daemon thread to execute _execute_task in the background. Called by start_web_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 method run in background thread that instantiates GeminiBrowserAgent and calls its execute_task method to perform the actual web browsing. Handles completion, error, and cleanup.

  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()