« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""MCP logging handler — structured log forwarding from server to client.
3Implements the MCP Logging specification so that server-side log events
4can be forwarded to the MCP client via notifications.
6https://spec.modelcontextprotocol.io/specification/server/utilities/logging/
7"""
9from __future__ import annotations
11from typing import TYPE_CHECKING, Any
13from lexigram.logging import (
14 get_logger,
15)
16from lexigram.result import Ok, Result
18if TYPE_CHECKING:
19 from lexigram.contracts.mcp.exceptions import MCPError
21logger = get_logger(__name__)
23# MCP-defined log levels (syslog-inspired, lower = more severe)
24_MCP_LEVELS = frozenset(
25 ["debug", "info", "notice", "warning", "error", "critical", "alert", "emergency"]
26)
29class LoggingHandler:
30 """MCP logging capability — forward structured logs to the connected client.
32 The MCP spec allows the server to emit ``notifications/message`` events
33 that are delivered to the client as log entries. This handler keeps a
34 configurable minimum level and holds a reference to a *send callback* so
35 it can push notifications asynchronously.
37 Usage in provider wiring::
39 logging_handler = LoggingHandler()
40 server.register_logging_handler(logging_handler)
42 The transport layer calls ``set_notify_callback`` once the client
43 connection is established.
45 Example client-directed log emission::
47 await logging_handler.log("info", "Ingestion started", data={"docs": 5})
48 """
50 def __init__(self, *, min_level: str = "info") -> None:
51 """Initialize the logging handler.
53 Args:
54 min_level: Minimum log level to forward to the client
55 (one of the MCP syslog levels). Default is ``"info"``.
56 """
57 if min_level not in _MCP_LEVELS:
58 msg = f"Invalid MCP log level '{min_level}'. Must be one of {sorted(_MCP_LEVELS)}"
59 raise ValueError(msg)
60 self._min_level = min_level
61 self._notify_callback: Any | None = None
62 self._level_order = list(_MCP_LEVELS) # preserved insertion order
64 def set_notify_callback(self, callback: Any) -> None:
65 """Register a coroutine callback for sending log notifications.
67 The callback receives a single ``dict`` which is a full JSON-RPC
68 ``notifications/message`` payload.
70 Args:
71 callback: Async callable ``callback(notification: dict) -> None``.
72 """
73 self._notify_callback = callback
75 async def set_level(
76 self, level: str = "info", **_kwargs: Any
77 ) -> Result[dict[str, Any], MCPError]:
78 """Handle ``logging/setLevel`` MCP method from the client.
80 Args:
81 level: New minimum level (MCP syslog name).
82 **_kwargs: Ignored extra params for forward-compatibility.
84 Returns:
85 ``Ok({})`` (MCP spec requires an empty result payload).
86 """
87 if level not in _MCP_LEVELS:
88 level = "info"
89 self._min_level = level
90 logger.info("mcp_log_level_changed", level=level)
91 return Ok({})
93 async def log(
94 self,
95 level: str,
96 message: str,
97 data: dict[str, Any] | None = None,
98 logger_name: str | None = None,
99 ) -> None:
100 """Emit a structured log entry to the MCP client.
102 If the *level* is below the current minimum, the entry is silently
103 dropped. When no notify callback is registered (no connected client),
104 the entry is still written to the local logger.
106 Args:
107 level: MCP syslog level name.
108 message: Human-readable log message.
109 data: Optional structured data payload.
110 logger_name: Optional logger name shown to the client.
111 """
112 if not self._is_at_or_above_min(level):
113 return
115 # Always emit locally regardless of client connection
116 local_log = getattr(logger, _map_level(level), logger.info)
117 local_log("mcp_client_log", level=level, message=message, **(data or {}))
119 if self._notify_callback is None:
120 return
122 notification: dict[str, Any] = {
123 "jsonrpc": "2.0",
124 "method": "notifications/message",
125 "params": {
126 "level": level,
127 "message": message,
128 },
129 }
130 if data:
131 notification["params"]["data"] = data
132 if logger_name:
133 notification["params"]["logger"] = logger_name
135 try:
136 await self._notify_callback(notification)
137 except (RuntimeError, ValueError, TypeError, LookupError, OSError) as exc:
138 logger.warning("mcp_log_notify_failed", error=str(exc))
140 # ------------------------------------------------------------------
141 # Internal helpers
142 # ------------------------------------------------------------------
144 def _is_at_or_above_min(self, level: str) -> bool:
145 """Return True if *level* is at or above the current minimum.
147 Uses the ordered index of the MCP syslog levels list.
148 """
149 levels_ordered = [
150 "debug",
151 "info",
152 "notice",
153 "warning",
154 "error",
155 "critical",
156 "alert",
157 "emergency",
158 ]
159 try:
160 min_idx = levels_ordered.index(self._min_level)
161 lvl_idx = levels_ordered.index(level)
162 return lvl_idx >= min_idx
163 except ValueError:
164 return True
167def _map_level(mcp_level: str) -> str:
168 """Map an MCP syslog level to a structlog method name."""
169 mapping = {
170 "debug": "debug",
171 "info": "info",
172 "notice": "info",
173 "warning": "warning",
174 "error": "error",
175 "critical": "critical",
176 "alert": "critical",
177 "emergency": "critical",
178 }
179 return mapping.get(mcp_level, "info")
182__all__ = ["LoggingHandler"]