Gemini Web Automation MCP

Instructions

Browse the web to complete a task using AI-powered browser automation.

The AI agent can navigate websites, click buttons, fill forms, search for information,
and interact with web pages just like a human user. This runs synchronously and returns
when the task is complete.

Args:
    task: What you want to accomplish (e.g., "Find the top 3 gaming laptops on Amazon")
    url: Starting webpage (defaults to Google)

Returns:
    Dictionary containing:
    - ok: Boolean indicating success
    - data: Task completion message with results
    - screenshot_dir: Path to saved screenshots
    - session_id: Unique session identifier
    - progress: List of actions taken during browsing
    - error: Error message (if task failed)

Examples:
    - "Search for Python tutorials and summarize the top result"
    - "Go to example.com and click the login button"
    - "Find product reviews for iPhone 15 Pro"

Note: For long-running tasks, consider using start_web_task instead.

Implementation Reference

• server.py:43-43 (registration)

  Registers 'browse_web' as an MCP tool using the FastMCP @mcp.tool() decorator.

  @mcp.tool()

• server.py:44-88 (handler)

  The main handler function for the 'browse_web' tool. It creates a GeminiBrowserAgent, runs the task synchronously via executor, and returns the result.

  async def browse_web(task: str, url: str = "https://www.google.com") -> dict[str, Any]:
      """
      Browse the web to complete a task using AI-powered browser automation.
  
      The AI agent can navigate websites, click buttons, fill forms, search for information,
      and interact with web pages just like a human user. This runs synchronously and returns
      when the task is complete.
  
      Args:
          task: What you want to accomplish (e.g., "Find the top 3 gaming laptops on Amazon")
          url: Starting webpage (defaults to Google)
  
      Returns:
          Dictionary containing:
          - ok: Boolean indicating success
          - data: Task completion message with results
          - screenshot_dir: Path to saved screenshots
          - session_id: Unique session identifier
          - progress: List of actions taken during browsing
          - error: Error message (if task failed)
  
      Examples:
          - "Search for Python tutorials and summarize the top result"
          - "Go to example.com and click the login button"
          - "Find product reviews for iPhone 15 Pro"
  
      Note: For long-running tasks, consider using start_web_task instead.
      """
      logger.info(f"Received web browsing request: {task}")
  
      # Create agent instance (browser will be cleaned up automatically)
      agent = GeminiBrowserAgent(logger=logger)
  
      try:
          # Execute task in thread pool to avoid blocking
          loop = asyncio.get_event_loop()
          result = await loop.run_in_executor(None, agent.execute_task, task, url)
  
          logger.info(f"Task completed with status: {result.get('ok')}")
          return result
  
      finally:
          # Clean up browser resources
          agent.cleanup_browser()

• server.py:45-71 (schema)

  Docstring defining the input parameters (task, url) and output format for the browse_web tool, used by FastMCP for schema inference.

  """
  Browse the web to complete a task using AI-powered browser automation.
  
  The AI agent can navigate websites, click buttons, fill forms, search for information,
  and interact with web pages just like a human user. This runs synchronously and returns
  when the task is complete.
  
  Args:
      task: What you want to accomplish (e.g., "Find the top 3 gaming laptops on Amazon")
      url: Starting webpage (defaults to Google)
  
  Returns:
      Dictionary containing:
      - ok: Boolean indicating success
      - data: Task completion message with results
      - screenshot_dir: Path to saved screenshots
      - session_id: Unique session identifier
      - progress: List of actions taken during browsing
      - error: Error message (if task failed)
  
  Examples:
      - "Search for Python tutorials and summarize the top result"
      - "Go to example.com and click the login button"
      - "Find product reviews for iPhone 15 Pro"
  
  Note: For long-running tasks, consider using start_web_task instead.
  """

• browser_agent.py:106-157 (helper)

  GeminiBrowserAgent.execute_task method called by the handler; sets up the browser, navigates to URL, runs the automation loop, handles screenshots and returns results.

  def execute_task(
      self, task: str, url: Optional[str] = "https://www.google.com"
  ) -> Dict[str, Any]:
      """
      Execute a browser automation task.
  
      Args:
          task: Description of the browsing task to perform
          url: Optional starting URL (defaults to Google)
  
      Returns:
          Dictionary with ok status and either data or error
      """
      try:
          self.logger.info(f"Task: {task}")
          self.logger.info(f"Starting URL: {url}")
          self.logger.info(f"Session ID: {self.session_id}")
  
          # Setup browser if not already done
          if not self.page:
              self.setup_browser()
  
          # Navigate to starting URL if provided
          if url:
              self.page.goto(url, wait_until="domcontentloaded", timeout=10000)
              self.logger.info(f"Navigated to: {url}")
          else:
              # Start with a search engine
              self.page.goto(
                  "https://www.google.com", wait_until="domcontentloaded", timeout=10000
              )
              self.logger.info("Starting from Google")
  
          # Run the browser automation loop
          result = self._run_browser_automation_loop(task)
  
          self.logger.info(
              f"Task completed! Screenshots saved to: {self.screenshot_dir}"
          )
  
          return {
              "ok": True,
              "data": result,
              "screenshot_dir": str(self.screenshot_dir),
              "session_id": self.session_id,
              "progress": self.progress_updates,
          }
  
      except Exception as exc:
          self.logger.exception("Browser automation failed")
          return {"ok": False, "error": str(exc)}

• browser_agent.py:158-278 (helper)

  Core automation loop in GeminiBrowserAgent; configures Gemini Computer Use tool, runs iterative agent turns with screenshots and function calls until task completion.

  def _run_browser_automation_loop(self, task: str, max_turns: int = 30) -> str:
      """
      Run the Gemini Computer Use agent loop to complete the task.
  
      Args:
          task: The browsing task to complete
          max_turns: Maximum number of agent turns
  
      Returns:
          The final result as a string
      """
      # Configure Gemini with Computer Use
      config = types.GenerateContentConfig(
          tools=[
              types.Tool(
                  computer_use=types.ComputerUse(
                      environment=types.Environment.ENVIRONMENT_BROWSER
                  )
              )
          ],
      )
  
      # Initial screenshot - take once and save
      initial_screenshot = self.page.screenshot(type="png")
      timestamp = datetime.now().strftime("%H%M%S")
      screenshot_path = (
          self.screenshot_dir
          / f"step_{self.screenshot_counter:02d}_initial_{timestamp}.png"
      )
      with open(screenshot_path, "wb") as f:
          f.write(initial_screenshot)
      self.logger.info(f"Saved initial screenshot: {screenshot_path}")
      self.screenshot_counter += 1
  
      # Build initial contents
      contents = [
          Content(
              role="user",
              parts=[
                  Part(text=task),
                  Part.from_bytes(data=initial_screenshot, mime_type="image/png"),
              ],
          )
      ]
  
      self.logger.info(f"Starting browser automation loop for task: {task}")
      self._add_progress("Started browser automation", "info")
  
      # Agent loop
      for turn in range(max_turns):
          self.logger.info(f"Turn {turn + 1}/{max_turns}")
          self._add_progress(f"Turn {turn + 1}/{max_turns}", "turn")
  
          try:
              # Get response from Gemini
              response = self.gemini_client.models.generate_content(
                  model=GEMINI_MODEL,
                  contents=contents,
                  config=config,
              )
  
              candidate = response.candidates[0]
              contents.append(candidate.content)
  
              # Check if there are function calls
              has_function_calls = any(
                  part.function_call for part in candidate.content.parts
              )
  
              if not has_function_calls:
                  # No more actions - extract final text response
                  text_response = " ".join(
                      [part.text for part in candidate.content.parts if part.text]
                  )
                  self.logger.info(f"Agent finished: {text_response}")
  
                  # Save final screenshot
                  timestamp = datetime.now().strftime("%H%M%S")
                  screenshot_path = (
                      self.screenshot_dir
                      / f"step_{self.screenshot_counter:02d}_final_{timestamp}.png"
                  )
                  self.page.screenshot(path=str(screenshot_path))
                  self.logger.info(f"Saved final screenshot: {screenshot_path}")
                  self.screenshot_counter += 1
  
                  return text_response
  
              # Execute function calls
              self.logger.info("Executing browser actions...")
              self._add_progress("Executing browser actions", "action")
              results = self._execute_gemini_function_calls(candidate)
  
              # Get function responses with new screenshot
              function_responses = self._get_gemini_function_responses(results)
  
              # Save screenshot after actions
              timestamp = datetime.now().strftime("%H%M%S")
              screenshot_path = (
                  self.screenshot_dir
                  / f"step_{self.screenshot_counter:02d}_{timestamp}.png"
              )
              self.page.screenshot(path=str(screenshot_path))
              self.logger.info(f"Saved screenshot: {screenshot_path}")
              self.screenshot_counter += 1
  
              # Add function responses to contents
              contents.append(
                  Content(
                      role="user",
                      parts=[Part(function_response=fr) for fr in function_responses],
                  )
              )
  
          except Exception as e:
              self.logger.error(f"Error in browser automation loop: {e}")
              raise
  
      # If we hit max turns, return what we have
      return f"Task reached maximum turns ({max_turns}). Please check browser state."

• browser_agent.py:279-391 (helper)

  Executes specific browser actions (navigate, click, type, scroll, etc.) dispatched from Gemini's function calls using Playwright.

  def _execute_gemini_function_calls(self, candidate) -> list:
      """Execute Gemini Computer Use function calls using Playwright."""
      results = []
      function_calls = [
          part.function_call for part in candidate.content.parts if part.function_call
      ]
  
      for function_call in function_calls:
          fname = function_call.name
          args = function_call.args
          self.logger.info(f"Executing Gemini action: {fname}")
          self._add_progress(f"Action: {fname}", "function_call")
  
          action_result = {}
  
          try:
              if fname == "open_web_browser":
                  pass  # Already open
              elif fname == "wait_5_seconds":
                  time.sleep(5)
              elif fname == "go_back":
                  self.page.go_back()
              elif fname == "go_forward":
                  self.page.go_forward()
              elif fname == "search":
                  self.page.goto("https://www.google.com")
              elif fname == "navigate":
                  self.page.goto(args["url"], wait_until="domcontentloaded", timeout=10000)
              elif fname == "click_at":
                  actual_x = self._denormalize_x(args["x"])
                  actual_y = self._denormalize_y(args["y"])
                  self.page.mouse.click(actual_x, actual_y)
              elif fname == "hover_at":
                  actual_x = self._denormalize_x(args["x"])
                  actual_y = self._denormalize_y(args["y"])
                  self.page.mouse.move(actual_x, actual_y)
              elif fname == "type_text_at":
                  actual_x = self._denormalize_x(args["x"])
                  actual_y = self._denormalize_y(args["y"])
                  text = args["text"]
                  press_enter = args.get("press_enter", True)
                  clear_before = args.get("clear_before_typing", True)
  
                  self.page.mouse.click(actual_x, actual_y)
                  if clear_before:
                      self.page.keyboard.press("Meta+A")
                      self.page.keyboard.press("Backspace")
                  self.page.keyboard.type(text)
                  if press_enter:
                      self.page.keyboard.press("Enter")
              elif fname == "key_combination":
                  keys = args["keys"]
                  self.page.keyboard.press(keys)
              elif fname == "scroll_document":
                  direction = args["direction"]
                  if direction == "down":
                      self.page.keyboard.press("PageDown")
                  elif direction == "up":
                      self.page.keyboard.press("PageUp")
                  elif direction == "left":
                      self.page.keyboard.press("ArrowLeft")
                  elif direction == "right":
                      self.page.keyboard.press("ArrowRight")
              elif fname == "scroll_at":
                  actual_x = self._denormalize_x(args["x"])
                  actual_y = self._denormalize_y(args["y"])
                  direction = args["direction"]
                  magnitude = args.get("magnitude", 800)
  
                  # Scroll by moving to position and using wheel
                  self.page.mouse.move(actual_x, actual_y)
                  scroll_amount = int(magnitude * SCREEN_HEIGHT / 1000)
                  if direction == "down":
                      self.page.mouse.wheel(0, scroll_amount)
                  elif direction == "up":
                      self.page.mouse.wheel(0, -scroll_amount)
                  elif direction == "left":
                      self.page.mouse.wheel(-scroll_amount, 0)
                  elif direction == "right":
                      self.page.mouse.wheel(scroll_amount, 0)
              elif fname == "drag_and_drop":
                  x = self._denormalize_x(args["x"])
                  y = self._denormalize_y(args["y"])
                  dest_x = self._denormalize_x(args["destination_x"])
                  dest_y = self._denormalize_y(args["destination_y"])
  
                  self.page.mouse.move(x, y)
                  self.page.mouse.down()
                  self.page.mouse.move(dest_x, dest_y)
                  self.page.mouse.up()
              else:
                  self.logger.warning(f"Unimplemented action: {fname}")
  
              # Quick stability check - only wait if navigation occurred
              if fname in ["navigate", "go_back", "go_forward", "search"]:
                  self.page.wait_for_load_state("domcontentloaded", timeout=3000)
              else:
                  time.sleep(0.3)  # Brief pause for UI updates
  
          except Exception as e:
              self.logger.error(f"Error executing {fname}: {e}")
              action_result = {"error": str(e)}
  
          # Get safety decision from the function call if present
          safety_decision = None
          if hasattr(function_call, 'safety_decision'):
              safety_decision = function_call.safety_decision
              self.logger.info(f"Safety decision present for {fname}: {safety_decision}")
  
          results.append((fname, action_result, safety_decision))
  
      return results