from collections.abc import Callable from typing import TYPE_CHECKING, Any from pydantic import BaseModel, ConfigDict from browser_use.browser import BrowserSession from browser_use.filesystem.file_system import FileSystem from browser_use.llm.base import BaseChatModel if TYPE_CHECKING: pass class RegisteredAction(BaseModel): """Model for a registered action""" name: str description: str function: Callable param_model: type[BaseModel] # If True, this action is known to change the page (e.g. navigate, search, go_back, switch). # multi_act() will abort remaining queued actions after executing a terminates_sequence action. terminates_sequence: bool = False # filters: provide specific domains to determine whether the action should be available on the given URL or not domains: list[str] | None = None # e.g. ['*.google.com', 'www.bing.com', 'yahoo.*] model_config = ConfigDict(arbitrary_types_allowed=True) def prompt_description(self) -> str: """Get a description of the action for the prompt in unstructured format""" schema = self.param_model.model_json_schema() params = [] if 'properties' in schema: for param_name, param_info in schema['properties'].items(): # Build parameter description param_desc = param_name # Add type information if available if 'type' in param_info: param_type = param_info['type'] param_desc += f'={param_type}' # Add description as comment if available if 'description' in param_info: param_desc += f' ({param_info["description"]})' params.append(param_desc) # Format: action_name: Description. (param1=type, param2=type, ...) if params: return f'{self.name}: {self.description}. ({", ".join(params)})' else: return f'{self.name}: {self.description}' class ActionModel(BaseModel): """Base model for dynamically created action models""" # this will have all the registered actions, e.g. # click_element = param_model = ClickElementParams # done = param_model = None # model_config = ConfigDict(arbitrary_types_allowed=True, extra='forbid') def get_index(self) -> int | None: """Get the index of the action""" # {'clicked_element': {'index':5}} params = self.model_dump(exclude_unset=True).values() if not params: return None for param in params: if param is not None and 'index' in param: return param['index'] return None def set_index(self, index: int): """Overwrite the index of the action""" # Get the action name and params action_data = self.model_dump(exclude_unset=True) action_name = next(iter(action_data.keys())) action_params = getattr(self, action_name) # Update the index directly on the model if hasattr(action_params, 'index'): action_params.index = index class ActionRegistry(BaseModel): """Model representing the action registry""" actions: dict[str, RegisteredAction] = {} @staticmethod def _match_domains(domains: list[str] | None, url: str) -> bool: """ Match a list of domain glob patterns against a URL. Args: domains: A list of domain patterns that can include glob patterns (* wildcard) url: The URL to match against Returns: True if the URL's domain matches the pattern, False otherwise """ if domains is None or not url: return True # Use the centralized URL matching logic from utils from browser_use.utils import match_url_with_domain_pattern for domain_pattern in domains: if match_url_with_domain_pattern(url, domain_pattern): return True return False def get_prompt_description(self, page_url: str | None = None) -> str: """Get a description of all actions for the prompt Args: page_url: If provided, filter actions by URL using domain filters. Returns: A string description of available actions. - If page is None: return only actions with no page_filter and no domains (for system prompt) - If page is provided: return only filtered actions that match the current page (excluding unfiltered actions) """ if page_url is None: # For system prompt (no URL provided), include only actions with no filters return '\n'.join(action.prompt_description() for action in self.actions.values() if action.domains is None) # only include filtered actions for the current page URL filtered_actions = [] for action in self.actions.values(): if not action.domains: # skip actions with no filters, they are already included in the system prompt continue # Check domain filter if self._match_domains(action.domains, page_url): filtered_actions.append(action) return '\n'.join(action.prompt_description() for action in filtered_actions) class SpecialActionParameters(BaseModel): """Model defining all special parameters that can be injected into actions""" model_config = ConfigDict(arbitrary_types_allowed=True) # optional user-provided context object passed down from Agent(context=...) # e.g. can contain anything, external db connections, file handles, queues, runtime config objects, etc. # that you might want to be able to access quickly from within many of your actions # browser-use code doesn't use this at all, we just pass it down to your actions for convenience context: Any | None = None # browser-use session object, can be used to create new tabs, navigate, access CDP browser_session: BrowserSession | None = None # Current page URL for filtering and context page_url: str | None = None # CDP client for direct Chrome DevTools Protocol access cdp_client: Any | None = None # CDPClient type from cdp_use # extra injected config if the action asks for these arg names page_extraction_llm: BaseChatModel | None = None file_system: FileSystem | None = None available_file_paths: list[str] | None = None has_sensitive_data: bool = False extraction_schema: dict | None = None @classmethod def get_browser_requiring_params(cls) -> set[str]: """Get parameter names that require browser_session""" return {'browser_session', 'cdp_client', 'page_url'}