"""Tool argument validation and browser operation dispatch."""
from __future__ import annotations
from collections.abc import Callable
from urllib.parse import urlsplit
from .browser_session import BrowserSession
from .errors import FirefoxMcpError
from .tool_schemas import TOOL_NAMES
class InvalidArgumentsError(ValueError):
"""Report invalid MCP tool-call arguments to JSON-RPC dispatch."""
def _object(arguments: object) -> dict[str, object]:
if not isinstance(arguments, dict):
raise InvalidArgumentsError("Tool arguments must be an object.")
return arguments
def _keys(
arguments: dict[str, object],
allowed: set[str],
required: set[str] = frozenset(),
) -> None:
extra = set(arguments) - allowed
missing = required - set(arguments)
if extra:
raise InvalidArgumentsError(
f"Unexpected tool argument: {sorted(extra)[0]}"
)
if missing:
raise InvalidArgumentsError(
f"Missing tool argument: {sorted(missing)[0]}"
)
def _string(
arguments: dict[str, object],
name: str,
*,
maximum: int | None = None,
allow_empty: bool = True,
) -> str:
value = arguments.get(name)
if not isinstance(value, str):
raise InvalidArgumentsError(f"{name} must be a string.")
if not allow_empty and not value:
raise InvalidArgumentsError(f"{name} must not be empty.")
if maximum is not None and len(value) > maximum:
raise InvalidArgumentsError(
f"{name} must contain at most {maximum} code points."
)
return value
def _integer(
arguments: dict[str, object],
name: str,
default: int,
minimum: int,
maximum: int,
) -> int:
value = arguments.get(name, default)
if isinstance(value, bool) or not isinstance(value, int):
raise InvalidArgumentsError(f"{name} must be an integer.")
if not minimum <= value <= maximum:
raise InvalidArgumentsError(
f"{name} must be between {minimum} and {maximum}."
)
return value
def _boolean(
arguments: dict[str, object],
name: str,
default: bool,
) -> bool:
value = arguments.get(name, default)
if not isinstance(value, bool):
raise InvalidArgumentsError(f"{name} must be a boolean.")
return value
def successResult(
text: str,
value: object,
) -> dict[str, object]:
"""Create a human-readable and structured successful MCP tool result."""
return {
"content": [{"type": "text", "text": text}],
"structuredContent": value,
"isError": False,
}
def errorResult(error: FirefoxMcpError) -> dict[str, object]:
"""Translate a browser-domain exception into a stable tool error."""
structured: dict[str, object] = {
"error": {
"code": error.code,
"message": error.message,
"retryable": error.retryable,
}
}
if error.details:
error_value = structured["error"]
if isinstance(error_value, dict):
error_value["details"] = error.details
return {
"content": [{"type": "text", "text": error.message}],
"structuredContent": structured,
"isError": True,
}
class ToolHandlers:
"""Validate all tool inputs before dispatching to BrowserSession."""
def __init__(self, browser: BrowserSession) -> None:
self.browser = browser
async def call(
self,
name: str,
raw_arguments: object,
cancelled: Callable[[], bool],
) -> dict[str, object]:
"""Execute one known tool and return its MCP result object."""
if name not in TOOL_NAMES:
raise InvalidArgumentsError(f"Unknown tool: {name}")
arguments = _object(raw_arguments)
try:
value = await self._dispatch(name, arguments, cancelled)
except FirefoxMcpError as error:
return errorResult(error)
return successResult(self._summary(name, value), value)
async def _dispatch(
self,
name: str,
arguments: dict[str, object],
cancelled: Callable[[], bool],
) -> object:
if name == "browser_status":
_keys(arguments, set())
return await self.browser.status(cancelled)
if name == "list_tabs":
_keys(arguments, set())
return {"tabs": await self.browser.listTabs(cancelled)}
if name == "select_tab":
return await self._selectTab(arguments, cancelled)
if name == "snapshot_page":
return await self._snapshot(arguments, cancelled)
if name == "click":
_keys(arguments, {"element", "wait_after_ms"}, {"element"})
return await self.browser.click(
_string(arguments, "element", allow_empty=False),
_integer(arguments, "wait_after_ms", 250, 0, 5000),
cancelled,
)
if name == "fill":
_keys(
arguments,
{"element", "value", "submit"},
{"element", "value"},
)
return await self.browser.fill(
_string(arguments, "element", allow_empty=False),
_string(arguments, "value", maximum=100_000),
_boolean(arguments, "submit", False),
cancelled,
)
if name == "type_text":
_keys(
arguments,
{"element", "text", "delay_ms"},
{"element", "text"},
)
return await self.browser.typeText(
_string(arguments, "element", allow_empty=False),
_string(arguments, "text", maximum=10_000),
_integer(arguments, "delay_ms", 0, 0, 1000),
cancelled,
)
if name == "press_key":
return await self._pressKey(arguments, cancelled)
if name == "run_javascript":
_keys(arguments, {"script", "timeout_ms"}, {"script"})
return await self.browser.runJavaScript(
_string(
arguments,
"script",
maximum=100_000,
allow_empty=False,
),
_integer(
arguments,
"timeout_ms",
self.browser.config.request_timeout_ms,
1,
120_000,
),
cancelled,
)
if name == "navigate":
return await self._navigate(arguments, cancelled)
if name == "wait_for":
return await self._waitFor(arguments, cancelled)
if name == "upload_files":
return await self._uploadFiles(arguments, cancelled)
raise AssertionError(f"Unhandled known tool: {name}")
async def _selectTab(
self,
arguments: dict[str, object],
cancelled: Callable[[], bool],
) -> object:
_keys(arguments, {"tab_id", "index"})
if ("tab_id" in arguments) == ("index" in arguments):
raise InvalidArgumentsError(
"select_tab requires exactly one of tab_id or index."
)
if "tab_id" in arguments:
tab_id = _string(arguments, "tab_id", allow_empty=False)
return await self.browser.selectTab(
tab_id=tab_id,
cancelled=cancelled,
)
index = _integer(arguments, "index", 0, 0, 2_147_483_647)
return await self.browser.selectTab(
index=index,
cancelled=cancelled,
)
async def _snapshot(
self,
arguments: dict[str, object],
cancelled: Callable[[], bool],
) -> object:
_keys(
arguments,
{"include_all_text", "max_elements", "max_text_chars"},
)
return await self.browser.snapshot(
include_all_text=_boolean(
arguments,
"include_all_text",
False,
),
max_elements=_integer(
arguments,
"max_elements",
self.browser.config.max_elements,
1,
self.browser.config.max_elements,
),
max_text_chars=_integer(
arguments,
"max_text_chars",
self.browser.config.max_text_chars,
1,
self.browser.config.max_text_chars,
),
cancelled=cancelled,
)
async def _pressKey(
self,
arguments: dict[str, object],
cancelled: Callable[[], bool],
) -> object:
_keys(arguments, {"element", "key"}, {"key"})
key = _string(arguments, "key")
allowed = {
"Enter", "Tab", "Escape", "ArrowUp", "ArrowDown",
"ArrowLeft", "ArrowRight", "Backspace", "Delete",
"Home", "End",
}
if key not in allowed:
raise InvalidArgumentsError("key is not supported.")
element = None
if "element" in arguments:
element = _string(arguments, "element", allow_empty=False)
return await self.browser.pressKey(element, key, cancelled)
async def _navigate(
self,
arguments: dict[str, object],
cancelled: Callable[[], bool],
) -> object:
_keys(
arguments,
{"url", "wait", "timeout_ms"},
{"url"},
)
url = _string(arguments, "url", allow_empty=False)
parsed = urlsplit(url)
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
raise InvalidArgumentsError(
"url must be an absolute HTTP or HTTPS URL."
)
wait = arguments.get("wait", "dom")
if wait not in {"none", "dom", "complete"}:
raise InvalidArgumentsError(
"wait must be none, dom, or complete."
)
timeout = _integer(
arguments,
"timeout_ms",
self.browser.config.request_timeout_ms,
0,
120_000,
)
return await self.browser.navigate(
url,
str(wait),
timeout,
cancelled,
)
async def _waitFor(
self,
arguments: dict[str, object],
cancelled: Callable[[], bool],
) -> object:
conditions = {"text", "selector", "url_contains"}
_keys(
arguments,
conditions | {"state", "timeout_ms", "poll_ms"},
)
present = conditions & set(arguments)
if len(present) != 1:
raise InvalidArgumentsError(
"wait_for requires exactly one condition."
)
condition = next(iter(present))
value = _string(arguments, condition)
state = arguments.get("state", "present")
if state not in {"present", "absent"}:
raise InvalidArgumentsError("state must be present or absent.")
return await self.browser.waitFor(
condition,
value,
str(state),
_integer(arguments, "timeout_ms", 30_000, 0, 120_000),
_integer(arguments, "poll_ms", 250, 50, 2000),
cancelled,
)
async def _uploadFiles(
self,
arguments: dict[str, object],
cancelled: Callable[[], bool],
) -> object:
_keys(arguments, {"element", "paths"}, {"element", "paths"})
paths = arguments.get("paths")
if not isinstance(paths, list) or not paths:
raise InvalidArgumentsError("paths must be a non-empty array.")
if not all(isinstance(path, str) and path for path in paths):
raise InvalidArgumentsError(
"Every paths entry must be a non-empty string."
)
return await self.browser.uploadFiles(
_string(arguments, "element", allow_empty=False),
paths,
cancelled,
)
@staticmethod
def _summary(name: str, value: object) -> str:
if name == "snapshot_page" and isinstance(value, dict):
return str(value.get("snapshot", "Page snapshot completed."))
if name == "list_tabs" and isinstance(value, dict):
tabs = value.get("tabs", [])
return f"Found {len(tabs) if isinstance(tabs, list) else 0} tabs."
if name == "select_tab" and isinstance(value, dict):
return f"Selected tab: {value.get('title', '')}"
if name == "browser_status" and isinstance(value, dict):
status = "connected" if value.get("connected") else "disconnected"
return f"Firefox RDP is {status}."
return f"{name} completed."