message / plugins /browser /main.py
hunian
refactor(plugins): 插件短名并统一 MCP tool 为 {plugin}-{tool}
cc826a1
Raw
History Blame Contribute Delete
3.66 kB
"""
浏览器MCP插件主模块(v3.0 纯 Lightpanda 透传)。
提供 28 个通用浏览器 MCP tool,对任意环境 agent 暴露与 Lightpanda 上游一致的
浏览器能力。纯 Lightpanda 单后端,单例长驻子进程。
"""
import logging
logger = logging.getLogger(__name__)
PLUGIN_VERSION = "3.0.0"
# 上游工具数(与 tools_meta 一致,供状态展示)
TOOLS_COUNT = 28
class BrowserMCPPlugin:
"""浏览器MCP插件。"""
def __init__(self):
self.name = "browser"
self.version = PLUGIN_VERSION
self.enabled = False
self._init_error = None
# 缓存二进制探测结果,避免 check_readiness / get_status 重复探测
self._binary_check_cache = None
def on_enable(self):
"""插件启用时调用。不预启动子进程——首次工具调用时按需拉起。"""
self.enabled = True
logger.info(f"{self.name} v{self.version} 已启用(lightpanda 单后端,懒启动)")
def on_disable(self):
"""插件禁用时调用。best-effort 关闭子进程。"""
self.enabled = False
try:
from plugins.browser.browser.client import LightpandaClient
LightpandaClient.instance().close_sync()
except Exception as e:
logger.warning(f"{self.name} 关闭子进程失败: {e}")
logger.info(f"{self.name} 已禁用")
def get_status(self) -> dict:
"""获取插件状态。"""
from plugins.browser.browser.client import LightpandaClient
process_alive = False
try:
process_alive = LightpandaClient.instance().process_alive
except Exception:
process_alive = False
return {
"name": self.name,
"version": self.version,
"enabled": self.enabled,
"backend": "lightpanda",
"binary_check": self._get_binary_check(),
"process_alive": process_alive,
"tools_count": TOOLS_COUNT,
"error": self._init_error,
}
def check_readiness(self) -> dict:
"""readiness hook,plugin_manager 可选调用。
Lightpanda 二进制不可用时返回 ok=False → DEPENDENCY_ERROR;
任意探测异常都返回 ok=True(不阻断其他插件)。
"""
try:
from plugins.browser.browser.binary_locator import locate_lightpanda
if self._binary_check_cache is None:
self._binary_check_cache = locate_lightpanda()
check = self._binary_check_cache
return {
"ok": check.available,
"reason": None if check.available else (check.error or "lightpanda unavailable"),
"details": {
"binary_path": check.path,
"version": check.version,
"source": check.source,
"platform": check.platform,
},
}
except Exception as e:
# 探测本身失败不应阻断插件运行
logger.warning(f"{self.name} readiness check 异常: {e}")
return {"ok": True, "reason": None, "details": {}}
def _get_binary_check(self) -> dict:
"""获取二进制探测结果(dict 形式,带内存缓存)。"""
if self._binary_check_cache is None:
from plugins.browser.browser.binary_locator import locate_lightpanda
self._binary_check_cache = locate_lightpanda()
return self._binary_check_cache.to_dict()
# 创建插件实例(plugin_loader 按 *Plugin 命名查找)
plugin = BrowserMCPPlugin()