mirror of
https://github.com/usestrix/strix.git
synced 2026-09-21 00:21:13 +00:00
qol
This commit is contained in:
STJ
parent
4c2b368976
commit
3718943134
5 changed files with 379 additions and 366 deletions
|
|
@ -138,11 +138,8 @@ module = [
|
|||
"opentelemetry.*",
|
||||
"scrubadub.*",
|
||||
"traceloop.*",
|
||||
"browser_use",
|
||||
"browser_use.*",
|
||||
"cdp_use",
|
||||
"cdp_use.*",
|
||||
"aiohttp",
|
||||
"aiohttp.*",
|
||||
]
|
||||
ignore_missing_imports = true
|
||||
|
|
@ -251,21 +248,6 @@ ignore = [
|
|||
"strix/tools/**/*.py" = [
|
||||
"ARG001", # Unused function argument (tools may have unused args for interface consistency)
|
||||
]
|
||||
"strix/tools/browser/**/*.py" = [
|
||||
"ARG002", # Unused method argument (interface methods may not use all args)
|
||||
"PLR0911", # Too many return statements (dispatchers and complex browser logic)
|
||||
"PLR0912", # Too many branches (dispatch functions)
|
||||
"PLR0915", # Too many statements (complex browser task handling)
|
||||
]
|
||||
"strix/interface/tool_components/browser_renderer.py" = [
|
||||
"PLR0911", # Too many return statements (action dispatcher)
|
||||
"PLR0912", # Too many branches
|
||||
"PLR0915", # Too many statements
|
||||
]
|
||||
"strix/telemetry/tracer.py" = [
|
||||
"PLR0912", # Too many branches (save_run_data is legitimately complex)
|
||||
"PLR0915", # Too many statements
|
||||
]
|
||||
|
||||
[tool.ruff.lint.isort]
|
||||
force-single-line = false
|
||||
|
|
|
|||
|
|
@ -1,5 +1,11 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from functools import cache
|
||||
from typing import Any, ClassVar
|
||||
from typing import TYPE_CHECKING, Any, ClassVar
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
|
||||
from pygments.lexers import get_lexer_by_name
|
||||
from pygments.styles import get_style_by_name
|
||||
|
|
@ -103,11 +109,29 @@ class BrowserRenderer(BaseToolRenderer):
|
|||
status: str,
|
||||
result: Any,
|
||||
) -> Text:
|
||||
if action == "run":
|
||||
return cls._build_run(args, status, result)
|
||||
# Dispatch to action-specific builders
|
||||
builders: dict[str, Callable[[], Text]] = {
|
||||
"run": lambda: cls._build_run(args, status, result),
|
||||
"launch": lambda: cls._build_launch(args, status, result),
|
||||
"navigate": lambda: cls._build_navigate(args, status),
|
||||
"search": lambda: cls._build_search(args, status),
|
||||
"click": lambda: cls._build_click(args, status),
|
||||
"input": lambda: cls._build_input(args, status),
|
||||
"upload_file": lambda: cls._build_upload_file(args, status),
|
||||
"scroll": lambda: cls._build_scroll(args, status),
|
||||
"find_text": lambda: cls._build_find_text(args, status),
|
||||
"send_keys": lambda: cls._build_send_keys(args, status),
|
||||
"search_page": lambda: cls._build_search_page(args, status),
|
||||
"find_elements": lambda: cls._build_find_elements(args, status),
|
||||
"dropdown_options": lambda: cls._build_dropdown_options(args, status),
|
||||
"select_dropdown": lambda: cls._build_select_dropdown(args, status),
|
||||
"evaluate": lambda: cls._build_evaluate(args, status),
|
||||
"wait": lambda: cls._build_wait(args, status),
|
||||
"switch": lambda: cls._build_switch(args, status),
|
||||
"done": lambda: cls._build_done(args, status),
|
||||
}
|
||||
|
||||
# --- simple one-liners ----------------------------------------
|
||||
simple = {
|
||||
simple_actions = {
|
||||
"go_back": "going back",
|
||||
"close_browser": "closing browser",
|
||||
"close_tab": "closing tab",
|
||||
|
|
@ -116,236 +140,253 @@ class BrowserRenderer(BaseToolRenderer):
|
|||
"extract": "extracting content",
|
||||
"read_long_content": "reading long content",
|
||||
}
|
||||
if action in simple:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append(simple[action], style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
# --- launch ----------------------------------------------------
|
||||
if action == "launch":
|
||||
mode = "local" if args.get("use_local") else "sandboxed"
|
||||
text = Text("◈ ", style=cls.LIFE)
|
||||
text.append("launching browser", style=f"bold {cls.LIFE}")
|
||||
text.append(f" {mode}", style=cls.DIM)
|
||||
res = result if isinstance(result, dict) else {}
|
||||
warning = res.get("warning")
|
||||
if warning:
|
||||
text.append(f"\n ⚠ {warning}", style=f"italic {cls.DIM}")
|
||||
return text
|
||||
file_actions = {"write_file", "read_file", "replace_file"}
|
||||
|
||||
# --- navigate / search ----------------------------------------
|
||||
if action == "navigate":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("navigating to ", style=cls.DIM)
|
||||
url = args.get("url", "")
|
||||
if len(url) > 80:
|
||||
url = url[:77] + "..."
|
||||
text.append(url, style=f"{cls.NAV} underline")
|
||||
if args.get("new_tab"):
|
||||
text.append(" in new tab", style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
if action in builders:
|
||||
return builders[action]()
|
||||
if action in simple_actions:
|
||||
return cls._build_simple(simple_actions[action], status)
|
||||
if action in file_actions:
|
||||
return cls._build_file_operation(action, args, status)
|
||||
return cls._build_fallback(action, status)
|
||||
|
||||
if action == "search":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("searching for ", style=cls.DIM)
|
||||
query = args.get("query", "")
|
||||
if query:
|
||||
preview = query if len(query) <= 80 else query[:77] + "..."
|
||||
text.append(f'"{preview}"', style=cls.INTERACT)
|
||||
engine = args.get("engine")
|
||||
if engine:
|
||||
text.append(f" via {engine}", style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_simple(cls, label: str, status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append(label, style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
# --- pointer actions -------------------------------------------
|
||||
if action == "click":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("clicking", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(f" #{index}", style=f"bold {cls.INTERACT}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_launch(cls, args: dict[str, Any], status: str, result: Any) -> Text:
|
||||
mode = "local" if args.get("use_local") else "sandboxed"
|
||||
text = Text("◈ ", style=cls.LIFE)
|
||||
text.append("launching browser", style=f"bold {cls.LIFE}")
|
||||
text.append(f" {mode}", style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
res = result if isinstance(result, dict) else {}
|
||||
warning = res.get("warning")
|
||||
if warning:
|
||||
text.append(f"\n ⚠ {warning}", style=f"italic {cls.DIM}")
|
||||
return text
|
||||
|
||||
# --- text / file input ----------------------------------------
|
||||
if action == "input":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("inputting", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(f" #{index}", style=f"bold {cls.INTERACT}")
|
||||
value = args.get("text")
|
||||
if value:
|
||||
preview = value if len(value) <= 60 else value[:57] + "..."
|
||||
text.append(f' "{preview}"', style=cls.INTERACT)
|
||||
if args.get("clear"):
|
||||
text.append(" (clear)", style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_navigate(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("navigating to ", style=cls.DIM)
|
||||
url = args.get("url", "")
|
||||
if len(url) > 80:
|
||||
url = url[:77] + "..."
|
||||
text.append(url, style=f"{cls.NAV} underline")
|
||||
if args.get("new_tab"):
|
||||
text.append(" in new tab", style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
if action == "upload_file":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("uploading file", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(f" #{index}", style=f"bold {cls.INTERACT}")
|
||||
path = args.get("path", "")
|
||||
if path:
|
||||
text.append(" ", style=cls.DIM)
|
||||
text.append(path, style=cls.NAV)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_search(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("searching for ", style=cls.DIM)
|
||||
query = args.get("query", "")
|
||||
if query:
|
||||
preview = query if len(query) <= 80 else query[:77] + "..."
|
||||
text.append(f'"{preview}"', style=cls.INTERACT)
|
||||
engine = args.get("engine")
|
||||
if engine:
|
||||
text.append(f" via {engine}", style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
# --- scroll / find --------------------------------------------
|
||||
if action == "scroll":
|
||||
direction = "down" if args.get("down", True) else "up"
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("scrolling ", style=cls.DIM)
|
||||
text.append(direction, style=cls.INTERACT)
|
||||
pages = args.get("pages")
|
||||
if pages is not None:
|
||||
text.append(f" {pages} page(s)", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(" on ", style=cls.DIM)
|
||||
text.append(f"#{index}", style=f"bold {cls.INTERACT}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_click(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("clicking", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(f" #{index}", style=f"bold {cls.INTERACT}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
if action == "find_text":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("finding text ", style=cls.DIM)
|
||||
value = args.get("text", "")
|
||||
if value:
|
||||
preview = value if len(value) <= 80 else value[:77] + "..."
|
||||
text.append(f'"{preview}"', style=cls.OBSERVE)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_input(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("inputting", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(f" #{index}", style=f"bold {cls.INTERACT}")
|
||||
value = args.get("text")
|
||||
if value:
|
||||
preview = value if len(value) <= 60 else value[:57] + "..."
|
||||
text.append(f' "{preview}"', style=cls.INTERACT)
|
||||
if args.get("clear"):
|
||||
text.append(" (clear)", style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
# --- keyboard --------------------------------------------------
|
||||
if action == "send_keys":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("pressing ", style=cls.DIM)
|
||||
text.append(args.get("keys", ""), style=f"bold {cls.INTERACT}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_upload_file(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("uploading file", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(f" #{index}", style=f"bold {cls.INTERACT}")
|
||||
path = args.get("path", "")
|
||||
if path:
|
||||
text.append(" ", style=cls.DIM)
|
||||
text.append(path, style=cls.NAV)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
# --- search/extract helpers -----------------------------------
|
||||
if action == "search_page":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("searching page for ", style=cls.DIM)
|
||||
pattern = args.get("pattern", "")
|
||||
if pattern:
|
||||
preview = pattern if len(pattern) <= 80 else pattern[:77] + "..."
|
||||
text.append(f'"{preview}"', style=cls.OBSERVE)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_scroll(cls, args: dict[str, Any], status: str) -> Text:
|
||||
direction = "down" if args.get("down", True) else "up"
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("scrolling ", style=cls.DIM)
|
||||
text.append(direction, style=cls.INTERACT)
|
||||
pages = args.get("pages")
|
||||
if pages is not None:
|
||||
text.append(f" {pages} page(s)", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(" on ", style=cls.DIM)
|
||||
text.append(f"#{index}", style=f"bold {cls.INTERACT}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
if action == "find_elements":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("finding elements ", style=cls.DIM)
|
||||
selector = args.get("selector", "")
|
||||
if selector:
|
||||
text.append(selector, style=cls.OBSERVE)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_find_text(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("finding text ", style=cls.DIM)
|
||||
value = args.get("text", "")
|
||||
if value:
|
||||
preview = value if len(value) <= 80 else value[:77] + "..."
|
||||
text.append(f'"{preview}"', style=cls.OBSERVE)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
# --- dropdowns -------------------------------------------------
|
||||
if action == "dropdown_options":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("reading dropdown options", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(f" #{index}", style=f"bold {cls.INTERACT}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_send_keys(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("pressing ", style=cls.DIM)
|
||||
text.append(args.get("keys", ""), style=f"bold {cls.INTERACT}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
if action == "select_dropdown":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("selecting dropdown value", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(f" #{index}", style=f"bold {cls.INTERACT}")
|
||||
value = args.get("text", "")
|
||||
if value:
|
||||
text.append(f' "{value}"', style=f"bold {cls.INTERACT}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_search_page(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("searching page for ", style=cls.DIM)
|
||||
pattern = args.get("pattern", "")
|
||||
if pattern:
|
||||
preview = pattern if len(pattern) <= 80 else pattern[:77] + "..."
|
||||
text.append(f'"{preview}"', style=cls.OBSERVE)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
# --- eval js ---------------------------------------------------
|
||||
if action == "evaluate":
|
||||
js = args.get("code")
|
||||
@classmethod
|
||||
def _build_find_elements(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("finding elements ", style=cls.DIM)
|
||||
selector = args.get("selector", "")
|
||||
if selector:
|
||||
text.append(selector, style=cls.OBSERVE)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("executing javascript", style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
if js:
|
||||
text.append("\n")
|
||||
text.append_text(cls._highlight_js(js))
|
||||
@classmethod
|
||||
def _build_dropdown_options(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("reading dropdown options", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(f" #{index}", style=f"bold {cls.INTERACT}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
return text
|
||||
@classmethod
|
||||
def _build_select_dropdown(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("selecting dropdown value", style=cls.DIM)
|
||||
index = args.get("index")
|
||||
if index is not None:
|
||||
text.append(f" #{index}", style=f"bold {cls.INTERACT}")
|
||||
value = args.get("text", "")
|
||||
if value:
|
||||
text.append(f' "{value}"', style=f"bold {cls.INTERACT}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
# --- wait ------------------------------------------------------
|
||||
if action == "wait":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
seconds = args.get("seconds")
|
||||
if status == "completed":
|
||||
text.append("waited", style=cls.DIM)
|
||||
else:
|
||||
text.append("waiting", style=cls.DIM)
|
||||
if seconds is not None:
|
||||
text.append(f" {seconds}s", style=cls.INTERACT)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_evaluate(cls, args: dict[str, Any], status: str) -> Text:
|
||||
js = args.get("code")
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("executing javascript", style=cls.DIM)
|
||||
text.append_text(cls._status_mark(status))
|
||||
if js:
|
||||
text.append("\n")
|
||||
text.append_text(cls._highlight_js(js))
|
||||
return text
|
||||
|
||||
# --- tab switch ------------------------------------------------
|
||||
if action == "switch":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("switching to tab ", style=cls.DIM)
|
||||
text.append(str(args.get("tab_id", "?")), style=f"bold {cls.NAV}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_wait(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
seconds = args.get("seconds")
|
||||
if status == "completed":
|
||||
text.append("waited", style=cls.DIM)
|
||||
else:
|
||||
text.append("waiting", style=cls.DIM)
|
||||
if seconds is not None:
|
||||
text.append(f" {seconds}s", style=cls.INTERACT)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
# --- file operations ------------------------------------------
|
||||
if action in {"write_file", "read_file", "replace_file"}:
|
||||
labels = {
|
||||
"write_file": "writing file",
|
||||
"read_file": "reading file",
|
||||
"replace_file": "replacing file content",
|
||||
}
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append(labels[action], style=cls.DIM)
|
||||
file_name = args.get("file_name", "")
|
||||
if file_name:
|
||||
text.append(" ", style=cls.DIM)
|
||||
text.append(file_name, style=cls.NAV)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
@classmethod
|
||||
def _build_switch(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append("switching to tab ", style=cls.DIM)
|
||||
text.append(str(args.get("tab_id", "?")), style=f"bold {cls.NAV}")
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
# --- completion ------------------------------------------------
|
||||
if action == "done":
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
success = args.get("success")
|
||||
if success is True:
|
||||
text.append("marking task done", style=f"bold {cls.OK}")
|
||||
elif success is False:
|
||||
text.append("marking task failed", style=f"bold {cls.ERR}")
|
||||
else:
|
||||
text.append("marking task done", style=cls.DIM)
|
||||
@classmethod
|
||||
def _build_file_operation(cls, action: str, args: dict[str, Any], status: str) -> Text:
|
||||
labels = {
|
||||
"write_file": "writing file",
|
||||
"read_file": "reading file",
|
||||
"replace_file": "replacing file content",
|
||||
}
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
text.append(labels[action], style=cls.DIM)
|
||||
file_name = args.get("file_name", "")
|
||||
if file_name:
|
||||
text.append(" ", style=cls.DIM)
|
||||
text.append(file_name, style=cls.NAV)
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
summary = args.get("text", "")
|
||||
if summary:
|
||||
preview = summary if len(summary) <= 120 else summary[:117] + "..."
|
||||
text.append("\n ")
|
||||
text.append(preview, style=cls.DIM)
|
||||
@classmethod
|
||||
def _build_done(cls, args: dict[str, Any], status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
success = args.get("success")
|
||||
if success is True:
|
||||
text.append("marking task done", style=f"bold {cls.OK}")
|
||||
elif success is False:
|
||||
text.append("marking task failed", style=f"bold {cls.ERR}")
|
||||
else:
|
||||
text.append("marking task done", style=cls.DIM)
|
||||
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
summary = args.get("text", "")
|
||||
if summary:
|
||||
preview = summary if len(summary) <= 120 else summary[:117] + "..."
|
||||
text.append("\n ")
|
||||
text.append(preview, style=cls.DIM)
|
||||
|
||||
# --- fallback --------------------------------------------------
|
||||
text.append_text(cls._status_mark(status))
|
||||
return text
|
||||
|
||||
@classmethod
|
||||
def _build_fallback(cls, action: str, status: str) -> Text:
|
||||
text = Text("@ ", style=cls.DIM)
|
||||
if action:
|
||||
text.append(action, style=cls.DIM)
|
||||
|
|
|
|||
|
|
@ -139,6 +139,109 @@ async def _execute_task(session: BrowserSession, operation: Any, desc: str) -> d
|
|||
return {"error": "Failed after 2 attempts", "is_running": False}
|
||||
|
||||
|
||||
async def _run_browser_agent(
|
||||
session: BrowserSession,
|
||||
task: str,
|
||||
return_fields: list[str] | None,
|
||||
) -> dict[str, Any]:
|
||||
llm = _build_llm()
|
||||
agent: Any = Agent(
|
||||
task=task,
|
||||
llm=llm,
|
||||
browser=session.browser,
|
||||
flash_mode=True,
|
||||
use_vision=llm_supports_vision(),
|
||||
)
|
||||
|
||||
async def log_step(step: Any) -> None:
|
||||
logger.info("Agent step completed: %s", step)
|
||||
|
||||
try:
|
||||
result = await agent.run(on_step_end=log_step)
|
||||
|
||||
if hasattr(result, "is_successful") and not result.is_successful():
|
||||
final_result = (
|
||||
result.final_result() if callable(result.final_result) else result.final_result
|
||||
)
|
||||
return {"error": final_result or "Agent failed", "is_running": False}
|
||||
|
||||
final_result = (
|
||||
result.final_result()
|
||||
if hasattr(result, "final_result") and callable(result.final_result)
|
||||
else getattr(result, "final_result", str(result))
|
||||
)
|
||||
out = {"message": "Task completed", "result": final_result, "is_running": False}
|
||||
|
||||
if return_fields:
|
||||
fields = {f: getattr(result, f, None) for f in return_fields}
|
||||
out["fields"] = fields
|
||||
|
||||
return out
|
||||
finally:
|
||||
for name in ("close", "stop"):
|
||||
if fn := getattr(agent, name, None):
|
||||
try:
|
||||
if asyncio.iscoroutine(coro := fn()):
|
||||
await asyncio.wait_for(coro, timeout=5)
|
||||
except Exception: # noqa: BLE001,S110
|
||||
pass
|
||||
|
||||
|
||||
def _fix_json_in_xml(kws: dict[str, Any]) -> dict[str, Any]:
|
||||
result = {}
|
||||
for k, v in kws.items():
|
||||
if v is None:
|
||||
continue
|
||||
if isinstance(v, str) and v.startswith(("[", "{")):
|
||||
try:
|
||||
result[k] = json.loads(v)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
result[k] = v
|
||||
else:
|
||||
result[k] = v
|
||||
return result
|
||||
|
||||
|
||||
async def _run_browser_tool(
|
||||
session: BrowserSession,
|
||||
action: str,
|
||||
params: dict[str, Any],
|
||||
) -> Any:
|
||||
if not session.browser.is_cdp_connected:
|
||||
await session.browser.start()
|
||||
|
||||
llm = _build_llm()
|
||||
|
||||
if session.local:
|
||||
from pathlib import Path
|
||||
|
||||
from browser_use.filesystem.file_system import FileSystem
|
||||
|
||||
base_dir = Path.cwd() / "browser_files"
|
||||
base_dir.mkdir(parents=True, exist_ok=True)
|
||||
file_system = FileSystem(base_dir=str(base_dir), create_default_files=False)
|
||||
else:
|
||||
|
||||
class StubFileSystem:
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
def soft_error(*args: Any, **kwargs: Any) -> dict[str, str]:
|
||||
error_msg = f"File operation '{name}' not available in sandboxed environment"
|
||||
logger.warning(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
return soft_error
|
||||
|
||||
file_system = StubFileSystem()
|
||||
|
||||
return await Tools().registry.execute_action(
|
||||
action,
|
||||
params=params,
|
||||
browser_session=session.browser,
|
||||
page_extraction_llm=llm,
|
||||
file_system=file_system,
|
||||
)
|
||||
|
||||
|
||||
@register_tool(sandbox_execution=False)
|
||||
async def browser_actions(
|
||||
action: BrowserUseLocalAction,
|
||||
|
|
@ -155,11 +258,9 @@ async def browser_actions(
|
|||
|
||||
agent_id = get_current_agent_id()
|
||||
|
||||
# Launch
|
||||
if action == "launch":
|
||||
if use_local:
|
||||
session = await _launch_local_browser(agent_id, profile_directory)
|
||||
|
||||
result = {
|
||||
"message": "Local browser ready",
|
||||
"mode": "local",
|
||||
|
|
@ -170,7 +271,6 @@ async def browser_actions(
|
|||
cdp_url, auth_token = _resolve_cdp_url(agent_state)
|
||||
session = await _launch_browser(cdp_url, agent_id, auth_token)
|
||||
ws_url = re.sub(r"[?&]token=[^&]+", "", session.ws_url)
|
||||
|
||||
result = {
|
||||
"message": "Browser ready",
|
||||
"mode": "sandboxed",
|
||||
|
|
@ -183,129 +283,28 @@ async def browser_actions(
|
|||
|
||||
return result
|
||||
|
||||
# Close
|
||||
if action == "close_browser":
|
||||
await _close_session(agent_id)
|
||||
return {"message": "Browser closed", "is_running": False}
|
||||
|
||||
session = _get_session(agent_id)
|
||||
|
||||
# Agent mode
|
||||
if action == "run":
|
||||
if not task:
|
||||
msg = "task required for run action"
|
||||
return {"error": msg, "is_running": False}
|
||||
return {"error": "task required for run action", "is_running": False}
|
||||
|
||||
async def run_agent() -> dict[str, Any]:
|
||||
llm = _build_llm()
|
||||
|
||||
agent: Any = Agent(
|
||||
task=task,
|
||||
llm=llm,
|
||||
browser=session.browser,
|
||||
flash_mode=True,
|
||||
use_vision=llm_supports_vision(),
|
||||
)
|
||||
|
||||
async def log_step(step: Any) -> None:
|
||||
logger.info("Agent step completed: %s", step)
|
||||
|
||||
try:
|
||||
result = await agent.run(on_step_end=log_step)
|
||||
|
||||
# Extract result
|
||||
if hasattr(result, "is_successful") and not result.is_successful():
|
||||
final_result = (
|
||||
result.final_result()
|
||||
if callable(result.final_result)
|
||||
else result.final_result
|
||||
)
|
||||
return {
|
||||
"error": final_result or "Agent failed",
|
||||
"is_running": False,
|
||||
}
|
||||
|
||||
final_result = (
|
||||
result.final_result()
|
||||
if hasattr(result, "final_result") and callable(result.final_result)
|
||||
else getattr(result, "final_result", str(result))
|
||||
)
|
||||
out = {"message": "Task completed", "result": final_result, "is_running": False}
|
||||
|
||||
# Return fields
|
||||
if return_fields:
|
||||
fields = {f: getattr(result, f, None) for f in return_fields}
|
||||
out["fields"] = fields
|
||||
|
||||
return out
|
||||
finally:
|
||||
# Cleanup
|
||||
for name in ("close", "stop"):
|
||||
if fn := getattr(agent, name, None):
|
||||
try:
|
||||
if asyncio.iscoroutine(coro := fn()):
|
||||
await asyncio.wait_for(coro, timeout=5)
|
||||
except Exception: # noqa: BLE001,S110
|
||||
pass
|
||||
|
||||
return await _execute_task(session, run_agent, task)
|
||||
|
||||
# [fix] parse json nested in xml
|
||||
def fix_json_in_xml(kws: dict[str, Any]) -> dict[str, Any]:
|
||||
result = {}
|
||||
for k, v in kws.items():
|
||||
if v is None:
|
||||
continue
|
||||
if isinstance(v, str) and v.startswith(("[", "{")):
|
||||
try:
|
||||
result[k] = json.loads(v)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
result[k] = v
|
||||
else:
|
||||
result[k] = v
|
||||
return result
|
||||
|
||||
params = fix_json_in_xml(kwargs)
|
||||
|
||||
async def run_tool() -> Any:
|
||||
# [resiliency] this happens really randomly. Better to be proactive
|
||||
if not session.browser.is_cdp_connected:
|
||||
await session.browser.start()
|
||||
|
||||
llm = _build_llm()
|
||||
|
||||
if session.local:
|
||||
from pathlib import Path
|
||||
|
||||
from browser_use.filesystem.file_system import FileSystem
|
||||
|
||||
base_dir = Path.cwd() / "browser_files"
|
||||
base_dir.mkdir(parents=True, exist_ok=True)
|
||||
file_system = FileSystem(base_dir=str(base_dir), create_default_files=False)
|
||||
else:
|
||||
# [monkeypatch] this is to make the screenshot tool work
|
||||
class StubFileSystem:
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
def soft_error(*args: Any, **kwargs: Any) -> dict[str, str]:
|
||||
error_msg = (
|
||||
f"File operation '{name}' not available in sandboxed environment"
|
||||
)
|
||||
logger.warning(error_msg)
|
||||
return {"error": error_msg}
|
||||
|
||||
return soft_error
|
||||
|
||||
file_system = StubFileSystem()
|
||||
|
||||
return await Tools().registry.execute_action(
|
||||
action,
|
||||
params=params,
|
||||
browser_session=session.browser,
|
||||
page_extraction_llm=llm,
|
||||
file_system=file_system,
|
||||
return await _execute_task(
|
||||
session,
|
||||
lambda: _run_browser_agent(session, task, return_fields),
|
||||
task,
|
||||
)
|
||||
|
||||
return await _execute_task(session, run_tool, f"{action}({list(params.keys())[:3]})")
|
||||
params = _fix_json_in_xml(kwargs)
|
||||
return await _execute_task(
|
||||
session,
|
||||
lambda: _run_browser_tool(session, action, params),
|
||||
f"{action}({list(params.keys())[:3]})",
|
||||
)
|
||||
|
||||
except Exception as error:
|
||||
logger.exception("browser_actions error: %s", action)
|
||||
|
|
|
|||
|
|
@ -146,33 +146,24 @@ class BrowserSession:
|
|||
return f"Failed to refresh local browser session: {exc}"
|
||||
return None
|
||||
|
||||
if self.needs_refresh:
|
||||
cdp_alive = await asyncio.to_thread(_check_cdp_alive, self.cdp_url, self.auth_token)
|
||||
if not cdp_alive and not await _wait_for_cdp_recovery(self, task_num):
|
||||
return (
|
||||
f"Chromium CDP at {self.cdp_url} is not responding "
|
||||
f"after {_CDP_RECOVERY_TIMEOUT}s"
|
||||
)
|
||||
try:
|
||||
await self.refresh()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return f"Failed to refresh browser session: {exc}"
|
||||
cdp_alive = await asyncio.to_thread(_check_cdp_alive, self.cdp_url, self.auth_token)
|
||||
|
||||
if not self.needs_refresh and cdp_alive:
|
||||
return None
|
||||
|
||||
if await asyncio.to_thread(_check_cdp_alive, self.cdp_url, self.auth_token):
|
||||
return None
|
||||
|
||||
if not await _wait_for_cdp_recovery(self, task_num):
|
||||
self.invalidated = True
|
||||
if not cdp_alive and not await _wait_for_cdp_recovery(self, task_num):
|
||||
if not self.needs_refresh:
|
||||
self.invalidated = True
|
||||
return (
|
||||
f"Chromium CDP at {self.cdp_url} is not responding after {_CDP_RECOVERY_TIMEOUT}s"
|
||||
)
|
||||
|
||||
try:
|
||||
await self.refresh()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self.invalidated = True
|
||||
return f"Chromium restarted but reconnection failed: {exc}"
|
||||
|
||||
if not self.needs_refresh:
|
||||
self.invalidated = True
|
||||
return f"Failed to refresh browser session: {exc}"
|
||||
return None
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -129,11 +129,11 @@ class ChatLiteLLM(BaseChatModel):
|
|||
**kwargs: Any,
|
||||
) -> ChatInvokeCompletion[T]: ...
|
||||
|
||||
async def ainvoke(
|
||||
async def ainvoke( # noqa: PLR0912
|
||||
self,
|
||||
messages: list[BaseMessage],
|
||||
output_format: type[T] | None = None,
|
||||
**kwargs: Any,
|
||||
**kwargs: Any, # noqa: ARG002
|
||||
) -> ChatInvokeCompletion[T] | ChatInvokeCompletion[str]:
|
||||
"""Invoke the model via litellm.
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue