Spaces:
Running
Running
File size: 1,670 Bytes
b05b338 741b0a2 6fe9090 741b0a2 c27f039 741b0a2 b05b338 741b0a2 c27f039 b05b338 fbfd90f c27f039 b05b338 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | from collections.abc import Awaitable, Callable
from typing import TypeAlias
from mcp.client.session import LoggingFnT
from mcp.types import LoggingMessageNotificationParams
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
LogMessage: TypeAlias = LoggingMessageNotificationParams
LogHandler: TypeAlias = Callable[[LogMessage], Awaitable[None]]
async def default_log_handler(message: LogMessage) -> None:
"""Default handler that properly routes server log messages to appropriate log levels."""
msg = message.data.get("msg", str(message))
extra = message.data.get("extra", {})
# Map MCP log levels to Python logging levels
level_map = {
"debug": logger.debug,
"info": logger.info,
"notice": logger.info, # Python doesn't have 'notice', map to info
"warning": logger.warning,
"error": logger.error,
"critical": logger.critical,
"alert": logger.critical, # Map alert to critical
"emergency": logger.critical, # Map emergency to critical
}
# Get the appropriate logging function based on the message level
log_fn = level_map.get(message.level.lower(), logger.info)
# Include logger name if available
if message.logger:
msg = f"[{message.logger}] {msg}"
# Log with appropriate level and extra data
log_fn(f"Server log: {msg}", extra=extra)
def create_log_callback(handler: LogHandler | None = None) -> LoggingFnT:
if handler is None:
handler = default_log_handler
async def log_callback(params: LoggingMessageNotificationParams) -> None:
await handler(params)
return log_callback
|