plugins/browser/mcp.py

"""
浏览器MCP插件 MCP 工具定义（v3.0 纯 Lightpanda 透传）。

28 个工具，每个由工厂从 tools_meta 生成，统一入参 params: dict 透传给 Lightpanda
上游，平台层不定义任何浏览器语义：
- 工具名/参数/返回/错误码均由 Lightpanda 上游决定。
- 平台只负责：run_log 埋点、调用透传、错误码归一为 BrowserError。
- 单例长驻子进程，复用会话，支持 goto→click→extract 链式操作。
"""

import logging
from typing import Any, Callable, Dict, Optional

from app.mcp.decorators import mcp_tool
from app.plugins.run_log import get_run_log_service, EventLevel, RunStatus

from .browser.client import LightpandaClient, DEFAULT_TOOL_TIMEOUT
from .browser.errors import BrowserError, BrowserErrorCode
from .browser.tools_meta import TOOLS_META, ToolMeta

logger = logging.getLogger(__name__)


def _annotations(meta: ToolMeta) -> Dict[str, Any]:
    """从上游谓词推导 MCP annotations，让 agent 知道工具是否只读。"""
    return {
        "readOnlyHint": meta.read_only,
        "destructiveHint": meta.destructive,
    }


def _summarize_args(args: Optional[dict]) -> str:
    """生成参数摘要用于 run_log（截断长值，避免日志爆炸）。"""
    if not args:
        return "{}"
    parts = []
    for k, v in args.items():
        s = v if isinstance(v, str) else repr(v)
        if len(s) > 80:
            s = s[:77] + "..."
        parts.append(f"{k}={s}")
    return ", ".join(parts)


async def _run_with_log(tool: str, params: Optional[dict]) -> Dict[str, Any]:
    """统一执行包装：create_run → 透传调用 → finish_run。

    返回 {success, run_id, tool, result/error_code}。
    """
    run_service = get_run_log_service()
    run = run_service.create_run("browser")
    run_service.add_event(run.run_id, "init", f"调用工具: {tool}")
    args = params or {}
    run_service.add_event(run.run_id, "call", f"{tool}({_summarize_args(args)})")

    try:
        text = await LightpandaClient.instance().call(tool, args, timeout=DEFAULT_TOOL_TIMEOUT)
    except BrowserError as e:
        _safe_event(run_service, run.run_id, "error", f"{e.code.value}: {e}", level=EventLevel.ERROR)
        _safe_finish(run_service, run.run_id, RunStatus.FAILED, error=str(e))
        return {
            "success": False,
            "run_id": run.run_id,
            "tool": tool,
            "error": str(e),
            "error_code": e.code.value,
        }
    except Exception as e:
        # 未归一的异常兜底
        _safe_event(run_service, run.run_id, "error", f"未捕获: {e}", level=EventLevel.ERROR)
        _safe_finish(run_service, run.run_id, RunStatus.FAILED, error=str(e))
        return {
            "success": False,
            "run_id": run.run_id,
            "tool": tool,
            "error": str(e),
            "error_code": BrowserErrorCode.TOOL_ERROR.value,
        }

    _safe_event(run_service, run.run_id, "complete", f"{tool} 完成")
    result = {
        "success": True,
        "run_id": run.run_id,
        "tool": tool,
        "result": text,
    }
    # run_log 是辅助观测，持久化失败不应拖垮工具调用的正常返回
    _safe_finish(run_service, run.run_id, RunStatus.SUCCEEDED, result=result)
    return result


def _safe_event(run_service, run_id: str, stage: str, message: str, level: EventLevel = EventLevel.INFO) -> None:
    """记录 run_log 事件，失败仅记日志不抛（不阻断主链路）。"""
    try:
        run_service.add_event(run_id=run_id, stage=stage, message=message, level=level)
    except Exception as e:
        logger.warning("run_log add_event 失败（忽略）: %s", e)


def _safe_finish(run_service, run_id: str, status: RunStatus, result=None, error=None) -> None:
    """结束 run，失败仅记日志不抛（不阻断主链路）。"""
    try:
        run_service.finish_run(run_id, status=status, result=result, error=error)
    except Exception as e:
        logger.warning("run_log finish_run 失败（忽略）: %s", e)


def _make_tool(meta: ToolMeta) -> Callable:
    """为单个工具元数据生成一个 @mcp_tool 装饰的透传函数。

    入参统一 params: dict，透传给上游，校验由 Lightpanda 负责。
    """
    # 命名：{插件名}-{上游工具名}，保留上游 camelCase
    tool_name = f"browser-{meta.name}"

    @mcp_tool(
        name=tool_name,
        title=meta.title,
        description=meta.description,
        annotations=_annotations(meta),
        risk_level="low" if meta.read_only else "medium",
    )
    async def _tool(params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """透传调用 Lightpanda 上游工具。params 为该工具的入参字典。"""
        return await _run_with_log(meta.name, params)

    # 函数名与 doc 影响可读性；description 已由装饰器记录
    _tool.__name__ = f"browser_{meta.name}"
    _tool.__doc__ = meta.description
    return _tool


# 模块级生成 28 个工具函数，供 plugin_registry._scan_tools 扫描注册。
# globals() 注入使 dir(module) 能发现带 _mcp_tool_def 的属性。
for _meta in TOOLS_META:
    globals()[f"browser_{_meta.name}"] = _make_tool(_meta)


def list_tool_defs() -> list[dict]:
    """列出本模块所有已注册工具的 _mcp_tool_def（供 api.py / 测试用）。"""
    import sys
    mod = sys.modules[__name__]
    defs = []
    for attr_name in dir(mod):
        attr = getattr(mod, attr_name, None)
        if callable(attr) and hasattr(attr, "_mcp_tool_def"):
            defs.append(attr._mcp_tool_def)
    return defs