Skip to content

datarobot_genai.drmcp.core.logging

logging

SecretRedactingFormatter

Bases: Formatter

Custom formatter that redacts sensitive information from logs.

Source code in datarobot_genai/drmcp/core/logging.py 
31
32
33
34
35
36
class SecretRedactingFormatter(logging.Formatter):
    """Custom formatter that redacts sensitive information from logs."""

    def format(self, record: logging.LogRecord) -> str:
        msg = super().format(record)
        return redact_secrets(msg)

MCPLogging

MCP Logging class.

Source code in datarobot_genai/drmcp/core/logging.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class MCPLogging:
    """MCP Logging class."""

    def __init__(self, level: str = "INFO") -> None:
        """Initialize the MCP logging."""
        self._level = level
        self._setup_logging()

    def _setup_logging(self) -> None:
        """Configure logging with secret redaction and set log level."""
        # Remove all existing handlers
        logging.root.handlers.clear()

        # Add a console handler with our formatter
        handler = logging.StreamHandler()
        logger_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
        formatter = SecretRedactingFormatter(logger_format)
        handler.setFormatter(formatter)
        logging.root.addHandler(handler)
        logging.root.setLevel(self._level)

__init__ 

__init__(level: str = 'INFO') -> None

Initialize the MCP logging.

Source code in datarobot_genai/drmcp/core/logging.py 
42
43
44
45
def __init__(self, level: str = "INFO") -> None:
    """Initialize the MCP logging."""
    self._level = level
    self._setup_logging()

log_execution 

log_execution(func: F) -> F

Log execution with error handling.

Source code in datarobot_genai/drmcp/core/logging.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def log_execution(func: F) -> F:
    """Log execution with error handling."""
    logger = logging.getLogger(func.__module__)

    @functools.wraps(func)
    async def wrapper(*args: Any, **kwargs: Any) -> Any:
        try:
            logger.info(f"Starting {func.__name__}")
            logger.debug(f"Arguments: {args}, {kwargs}")
            result = await func(*args, **kwargs)
            logger.info(f"Completed {func.__name__}")
            return result
        except FastMCPToolError as e:
            _log_error(logger, func.__name__, e, args=args, kwargs=kwargs)
            raise
        except DRToolError as e:
            _log_error(logger, func.__name__, e, args=args, kwargs=kwargs)
            raise FastMCPToolError(_mcp_message_for_dr_tool_error(e)) from e
        except (PydanticValidationError, FastMCPValidationError) as e:
            error_msg = _log_error(logger, func.__name__, e, args=args, kwargs=kwargs)
            raise FastMCPToolError(f"[{ToolErrorKind.SCHEMA.value}] {error_msg}") from e
        except Exception as e:
            error_msg = _log_error(logger, func.__name__, e, args=args, kwargs=kwargs)
            kind = _kind_for_wrapped_exception(e)
            raise FastMCPToolError(f"[{kind.value}] {error_msg}") from e

    return wrapper  # type: ignore[return-value]