feat: browser_external_cdp

.lintstagedrc.json

@@ -5,6 +5,6 @@
     "node licenses/update_license.js"
   ],
   "*.{js,jsx}": ["eslint --fix --no-warn-ignored", "prettier --write"],
-  "*.{json,css,md}": ["prettier --write"],
+  "*.{json,css}": ["prettier --write"],
   "*.py": ["node licenses/update_license.js"]
 }

backend/app/agent/factory/browser.py

@@ -13,6 +13,7 @@
 # ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
 
 import platform
+import threading
 import uuid
 
 from camel.messages import BaseMessage
@@ -35,6 +36,116 @@
 from app.utils.file_utils import get_working_directory
 
 
+def _get_browser_port(browser: dict) -> int:
+    """Extract port from a browser config dict, with fallback to env default."""
+    return int(browser.get("port", env("browser_port", "9222")))
+
+
+class CdpBrowserPoolManager:
+    """Manages CDP browser pool occupation to ensure
+    parallel tasks use different browsers."""
+
+    def __init__(self):
+        self._occupied_ports: dict[int, str] = {}
+        self._session_to_port: dict[str, int] = {}
+        self._session_to_task: dict[str, str | None] = {}
+        self._lock = threading.Lock()
+
+    def acquire_browser(
+        self,
+        cdp_browsers: list[dict],
+        session_id: str,
+        task_id: str | None = None,
+    ) -> dict | None:
+        """Acquire an available browser from the pool.
+
+        Args:
+            cdp_browsers: List of browser configurations.
+            session_id: Unique session identifier.
+            task_id: Optional task identifier for ownership tracking.
+
+        Returns:
+            Browser configuration dict or None if all occupied.
+        """
+        with self._lock:
+            for browser in cdp_browsers:
+                port = browser.get("port")
+                if port and port not in self._occupied_ports:
+                    self._occupied_ports[port] = session_id
+                    self._session_to_port[session_id] = port
+                    self._session_to_task[session_id] = task_id
+                    logger.info(
+                        f"Acquired browser on port {port} for session "
+                        f"{session_id}. Occupied: "
+                        f"{list(self._occupied_ports.keys())}"
+                    )
+                    return browser
+            logger.warning(
+                f"No available browsers for session {session_id}. "
+                f"All occupied: {list(self._occupied_ports.keys())}"
+            )
+            return None
+
+    def release_browser(self, port: int, session_id: str):
+        """Release a browser back to the pool."""
+        with self._lock:
+            if (
+                port in self._occupied_ports
+                and self._occupied_ports[port] == session_id
+            ):
+                del self._occupied_ports[port]
+                self._session_to_port.pop(session_id, None)
+                self._session_to_task.pop(session_id, None)
+                logger.info(
+                    f"Released browser on port {port} from session "
+                    f"{session_id}. Occupied: "
+                    f"{list(self._occupied_ports.keys())}"
+                )
+            else:
+                logger.warning(
+                    f"Attempted to release browser on port {port} "
+                    f"but it was not occupied by {session_id}"
+                )
+
+    def release_by_task(self, task_id: str) -> list[int]:
+        """Release all browsers associated with a task_id.
+
+        Returns:
+            List of released ports.
+        """
+        released_ports = []
+        with self._lock:
+            sessions = [
+                s for s, t in self._session_to_task.items() if t == task_id
+            ]
+            for session_id in sessions:
+                port = self._session_to_port.get(session_id)
+                if (
+                    port is not None
+                    and self._occupied_ports.get(port) == session_id
+                ):
+                    del self._occupied_ports[port]
+                    released_ports.append(port)
+                self._session_to_port.pop(session_id, None)
+                self._session_to_task.pop(session_id, None)
+            if released_ports:
+                logger.info(
+                    f"Released {len(released_ports)} browser(s) for "
+                    f"task {task_id}. Occupied: "
+                    f"{list(self._occupied_ports.keys())}"
+                )
+        return released_ports
+
+    def get_occupied_ports(self) -> list[int]:
+        """Get list of currently occupied ports."""
+        with self._lock:
+            return list(self._occupied_ports.keys())
+
+
+# Global CDP browser pool manager instance
+_cdp_pool_manager = CdpBrowserPoolManager()
+
+
 def browser_agent(options: Chat):
     working_directory = get_working_directory(options)
     logger.info(
@@ -47,14 +158,43 @@ def browser_agent(options: Chat):
         ).send_message_to_user
     )
 
+    # Acquire CDP browser from pool or use default port
+    toolkit_session_id = str(uuid.uuid4())[:8]
+    selected_port = None
+    selected_is_external = False
+
+    if options.cdp_browsers:
+        selected_browser = _cdp_pool_manager.acquire_browser(
+            options.cdp_browsers, toolkit_session_id, options.task_id
+        )
+        if selected_browser:
+            selected_port = _get_browser_port(selected_browser)
+            selected_is_external = selected_browser.get("isExternal", False)
+            logger.info(
+                f"Acquired CDP browser from pool (initial): "
+                f"port={selected_port}, isExternal={selected_is_external}, "
+                f"session_id={toolkit_session_id}"
+            )
+        else:
+            selected_port = _get_browser_port(options.cdp_browsers[0])
+            selected_is_external = options.cdp_browsers[0].get(
+                "isExternal", False
+            )
+            logger.warning(
+                f"No available browsers in pool (initial), using first: "
+                f"port={selected_port}, session_id={toolkit_session_id}"
+            )
+    else:
+        selected_port = env("browser_port", "9222")
+
     web_toolkit_custom = HybridBrowserToolkit(
         options.project_id,
         headless=False,
         browser_log_to_file=True,
         stealth=True,
-        session_id=str(uuid.uuid4())[:8],
+        session_id=toolkit_session_id,
         default_start_url="about:blank",
-        cdp_url=f"http://localhost:{env('browser_port', '9222')}",
+        cdp_url=f"http://localhost:{selected_port}",
         enabled_tools=[
             "browser_click",
             "browser_type",
@@ -113,14 +253,28 @@ def browser_agent(options: Chat):
         *search_tools,
     ]
 
+    # Build external browser notice
+    external_browser_notice = ""
+    if selected_is_external:
+        external_browser_notice = (
+            "\n<external_browser_connection>\n"
+            "**IMPORTANT**: You are connected to an external browser instance. "
+            "The browser may already be open with active sessions and logged-in "
+            "websites. When you use browser tools, you will connect to this "
+            "existing browser and can immediately access its current state and "
+            "pages.\n"
+            "</external_browser_connection>\n"
+        )
+
     system_message = BROWSER_SYS_PROMPT.format(
         platform_system=platform.system(),
         platform_machine=platform.machine(),
         working_directory=working_directory,
         now_str=NOW_STR,
+        external_browser_notice=external_browser_notice,
     )
 
-    return agent_model(
+    agent = agent_model(
         Agents.browser_agent,
         BaseMessage.make_assistant_message(
             role_name="Browser Agent",
@@ -139,3 +293,45 @@ def browser_agent(options: Chat):
         toolkits_to_register_agent=[web_toolkit_for_agent_registration],
         enable_snapshot_clean=True,
     )
+
+    # Attach CDP management callbacks and info to the agent
+    def acquire_cdp_for_agent(agent_instance):
+        """Acquire a CDP browser from pool for a cloned agent."""
+        if not options.cdp_browsers:
+            return
+        session_id = str(uuid.uuid4())[:8]
+        selected = _cdp_pool_manager.acquire_browser(
+            options.cdp_browsers, session_id, options.task_id
+        )
+        if selected:
+            agent_instance._cdp_port = _get_browser_port(selected)
+        else:
+            agent_instance._cdp_port = _get_browser_port(
+                options.cdp_browsers[0]
+            )
+        agent_instance._cdp_session_id = session_id
+        logger.info(
+            f"Acquired CDP for cloned agent {agent_instance.agent_id}: "
+            f"port={agent_instance._cdp_port}, session={session_id}"
+        )
+
+    def release_cdp_from_agent(agent_instance):
+        """Release CDP browser back to pool."""
+        port = getattr(agent_instance, "_cdp_port", None)
+        session_id = getattr(agent_instance, "_cdp_session_id", None)
+        if port is not None and session_id is not None:
+            _cdp_pool_manager.release_browser(port, session_id)
+            logger.info(
+                f"Released CDP for agent {agent_instance.agent_id}: "
+                f"port={port}, session={session_id}"
+            )
+
+    agent._cdp_acquire_callback = acquire_cdp_for_agent
+    agent._cdp_release_callback = release_cdp_from_agent
+    agent._cdp_port = selected_port
+    agent._cdp_session_id = toolkit_session_id
+    agent._cdp_task_id = options.task_id
+    agent._cdp_options = options
+    agent._browser_toolkit = web_toolkit_for_agent_registration
+
+    return agent

backend/app/agent/listen_chat_agent.py

@@ -15,6 +15,7 @@
 import asyncio
 import json
 import logging
+import threading
 from collections.abc import Callable
 from threading import Event
 from typing import Any
@@ -52,6 +53,10 @@
 
 
 class ListenChatAgent(ChatAgent):
+    _cdp_clone_lock = (
+        threading.Lock()
+    )  # Protects CDP URL mutation during clone
+
     def __init__(
         self,
         api_task_id: str,
@@ -692,8 +697,62 @@ def clone(self, with_memory: bool = False) -> ChatAgent:
         """Please see super.clone()"""
         system_message = None if with_memory else self._original_system_message
 
-        # Clone tools and collect toolkits that need registration
-        cloned_tools, toolkits_to_register = self._clone_tools()
+        # If this agent has CDP acquire callback, acquire CDP BEFORE cloning
+        # tools so that HybridBrowserToolkit clones with the correct CDP port
+        new_cdp_port = None
+        new_cdp_session = None
+        has_cdp = hasattr(self, "_cdp_acquire_callback") and callable(
+            getattr(self, "_cdp_acquire_callback", None)
+        )
+
+        need_cdp_clone = False
+        if has_cdp and hasattr(self, "_cdp_options"):
+            options = self._cdp_options
+            cdp_browsers = getattr(options, "cdp_browsers", [])
+            if cdp_browsers and hasattr(self, "_browser_toolkit"):
+                need_cdp_clone = True
+                import uuid as _uuid
+
+                from app.agent.factory.browser import _cdp_pool_manager
+
+                new_cdp_session = str(_uuid.uuid4())[:8]
+                selected = _cdp_pool_manager.acquire_browser(
+                    cdp_browsers,
+                    new_cdp_session,
+                    getattr(self, "_cdp_task_id", None),
+                )
+                from app.agent.factory.browser import _get_browser_port
+
+                if selected:
+                    new_cdp_port = _get_browser_port(selected)
+                else:
+                    new_cdp_port = _get_browser_port(cdp_browsers[0])
+
+        if need_cdp_clone:
+            # Temporarily override the browser toolkit's CDP URL.
+            # Lock prevents concurrent clones from clobbering each
+            # other's cdp_url on the shared parent toolkit.
+            toolkit = self._browser_toolkit
+            with ListenChatAgent._cdp_clone_lock:
+                original_cdp_url = (
+                    toolkit.config_loader.get_browser_config().cdp_url
+                )
+                toolkit.config_loader.get_browser_config().cdp_url = (
+                    f"http://localhost:{new_cdp_port}"
+                )
+                try:
+                    cloned_tools, toolkits_to_register = self._clone_tools()
+                except Exception:
+                    _cdp_pool_manager.release_browser(
+                        new_cdp_port, new_cdp_session
+                    )
+                    raise
+                finally:
+                    toolkit.config_loader.get_browser_config().cdp_url = (
+                        original_cdp_url
+                    )
+        else:
+            cloned_tools, toolkits_to_register = self._clone_tools()
 
         new_agent = ListenChatAgent(
             api_task_id=self.api_task_id,
@@ -726,6 +785,31 @@ def clone(self, with_memory: bool = False) -> ChatAgent:
 
         new_agent.process_task_id = self.process_task_id
 
+        # Copy CDP management data to cloned agent
+        if has_cdp:
+            new_agent._cdp_acquire_callback = self._cdp_acquire_callback
+            new_agent._cdp_release_callback = self._cdp_release_callback
+            if hasattr(self, "_cdp_options"):
+                new_agent._cdp_options = self._cdp_options
+            if hasattr(self, "_cdp_task_id"):
+                new_agent._cdp_task_id = self._cdp_task_id
+
+            # Find and store the cloned browser toolkit on the new agent
+            for tk in toolkits_to_register:
+                if tk.__class__.__name__ == "HybridBrowserToolkit":
+                    new_agent._browser_toolkit = tk
+                    break
+
+            # Set CDP info on cloned agent
+            if new_cdp_port is not None and new_cdp_session is not None:
+                new_agent._cdp_port = new_cdp_port
+                new_agent._cdp_session_id = new_cdp_session
+            else:
+                if hasattr(self, "_cdp_port"):
+                    new_agent._cdp_port = self._cdp_port
+                if hasattr(self, "_cdp_session_id"):
+                    new_agent._cdp_session_id = self._cdp_session_id
+
         # Copy memory if requested
         if with_memory:
             # Get all records from the current memory

backend/app/agent/prompt.py

@@ -596,7 +596,7 @@
 </capabilities>
 
 <web_search_workflow>
-Your approach depends on available search tools:
+{external_browser_notice}Your approach depends on available search tools:
 
 **If Google Search is Available:**
 - Initial Search: Start with `search_google` to get a list of relevant URLs