vent_deep_agent/tools/mcp_logger.py

# -*- coding: utf-8 -*-
"""
MCP 调用日志工具

从 .env 读取 LOG_LEVEL 配置，低于该级别的日志静默忽略。
MCP 调用的详细日志（工具名、参数、返回值）统一通过此模块输出。

注意：fastmcp 库自身会输出大量内部日志（连接握手、SSE 消息等），
本模块在导入时将 fastmcp 相关 logger 强制设为 WARNING，避免噪音。

日志级别（从低到高）：DEBUG < INFO < WARNING < ERROR

使用方式：
    from tools.mcp_logger import log_mcp_call
    log_mcp_call(tool_name, arguments, result_raw)
"""

import json
import logging
import os
from pathlib import Path
from dotenv import dotenv_values

# ── 压制 fastmcp 库的内部日志噪音 ──
for _name in ("mcp", "mcp.client", "mcp.server", "fastmcp", "httpx", "httpcore"):
    logging.getLogger(_name).setLevel(logging.WARNING)

# ── 读取 .env 中的 LOG_LEVEL ──
_env_file = Path(__file__).parent.parent / ".env"
_cfg = dotenv_values(str(_env_file))
_level_str = _cfg.get("LOG_LEVEL", "info").strip().upper()

_level_map = {
    "DEBUG": logging.DEBUG,
    "INFO": logging.INFO,
    "WARNING": logging.WARNING,
    "ERROR": logging.ERROR,
    "CRITICAL": logging.CRITICAL,
}
_level = _level_map.get(_level_str, logging.INFO)

# ── 配置业务 logger（与 fastmcp 隔离）──
_logger = logging.getLogger("vent.mcp")
_logger.setLevel(_level)

if not _logger.handlers:
    _handler = logging.StreamHandler()
    _handler.setLevel(_level)
    _handler.setFormatter(logging.Formatter(
        "[MCP] %(asctime)s %(message)s",
        datefmt="%H:%M:%S",
    ))
    _logger.addHandler(_handler)


def log_mcp_call(tool_name: str, arguments: dict, result_raw: str, error: str | None = None):
    """记录一次 MCP 工具调用的完整信息（INFO 级别，始终可见）。

    输出内容：
    - 工具名 + 参数
    - 返回结果总字符数
    - 返回结果前 1000 字符（截断标记）
    - 若出错，输出错误信息（ERROR 级别）

    Args:
        tool_name: MCP 工具名称
        arguments: 调用参数字典
        result_raw: 返回的原始字符串
        error: 异常信息（成功时为 None）
    """
    params_str = json.dumps(arguments, ensure_ascii=False)
    result_len = len(result_raw) if result_raw else 0

    if error:
        _logger.error(
            "✗ %s 调用失败 | 参数: %s | 错误: %s",
            tool_name, params_str, error,
        )
    else:
        preview = result_raw[:1000]
        suffix = "…" if len(result_raw) > 1000 else ""
        _logger.info(
            "→ %s | 参数: %s\n"
            "← 返回 %d 字符 | 预览: %s%s",
            tool_name, params_str,
            result_len, preview, suffix,
        )