Stream server logs to clients via notifications/message

Add an MCPLogBridge logging.Handler that, while serve_stdio is
running, forwards every project-logger record to the MCP client as
a notifications/message. The handler is attached / detached
automatically per stdio session. logging/setLevel requests retune
the bridge level on the fly, and the initialize handshake now
advertises the logging capability so clients know to expect them.

Author: JE-Chen

je_auto_control/utils/mcp_server/__init__.py

@@ -17,6 +17,7 @@
     FakeState, fake_state, install_fake_backend, reset_fake_state,
     uninstall_fake_backend,
 )
+from je_auto_control.utils.mcp_server.log_bridge import MCPLogBridge
 from je_auto_control.utils.mcp_server.rate_limit import RateLimiter
 from je_auto_control.utils.mcp_server.http_transport import (
     HttpMCPServer, start_mcp_http_server,
@@ -33,10 +34,11 @@
 )
 
 __all__ = [
-    "AuditLogger", "FakeState", "HttpMCPServer", "MCPContent", "MCPPrompt",
-    "MCPPromptArgument", "MCPResource", "MCPServer", "MCPTool",
-    "MCPToolAnnotations", "OperationCancelledError", "PromptProvider",
-    "RateLimiter", "ResourceProvider", "ToolCallContext",
+    "AuditLogger", "FakeState", "HttpMCPServer", "MCPContent",
+    "MCPLogBridge", "MCPPrompt", "MCPPromptArgument", "MCPResource",
+    "MCPServer", "MCPTool", "MCPToolAnnotations",
+    "OperationCancelledError", "PromptProvider", "RateLimiter",
+    "ResourceProvider", "ToolCallContext",
     "build_default_tool_registry",
     "default_prompt_provider", "default_resource_provider",
     "fake_state", "install_fake_backend", "make_plugin_tool",

je_auto_control/utils/mcp_server/log_bridge.py

@@ -0,0 +1,90 @@
+"""Bridge Python logging records onto MCP ``notifications/message``.
+
+Once attached to the project logger, every record at or above the
+configured level is forwarded to the MCP client as a notification so
+the client can mirror server-side activity in its UI. The handler is
+no-op when the server's notifier is not yet connected — useful for
+unit tests that don't actually start a transport.
+"""
+import logging
+from typing import Any, Callable, Dict, Optional
+
+# MCP log levels (RFC 5424 syslog names) mapped from stdlib logging levels.
+_LEVEL_NAME_FROM_LEVEL = {
+    logging.DEBUG: "debug",
+    logging.INFO: "info",
+    logging.WARNING: "warning",
+    logging.ERROR: "error",
+    logging.CRITICAL: "critical",
+}
+
+_LEVEL_FROM_MCP_NAME = {
+    "debug": logging.DEBUG,
+    "info": logging.INFO,
+    "notice": logging.INFO,
+    "warning": logging.WARNING,
+    "error": logging.ERROR,
+    "critical": logging.CRITICAL,
+    "alert": logging.CRITICAL,
+    "emergency": logging.CRITICAL,
+}
+
+
+def mcp_level_to_logging(name: str) -> Optional[int]:
+    """Return the :mod:`logging` level for an MCP log name, or ``None``."""
+    return _LEVEL_FROM_MCP_NAME.get(str(name).strip().lower())
+
+
+def logging_level_to_mcp(level: int) -> str:
+    """Return the closest MCP level name for a stdlib logging level."""
+    closest = max(
+        (lvl for lvl in _LEVEL_NAME_FROM_LEVEL if lvl <= int(level)),
+        default=logging.DEBUG,
+    )
+    return _LEVEL_NAME_FROM_LEVEL[closest]
+
+
+class MCPLogBridge(logging.Handler):
+    """Logging handler that forwards records as ``notifications/message``."""
+
+    def __init__(self, notifier: Optional[
+            Callable[[str, Dict[str, Any]], None]] = None,
+                 logger_name: str = "je_auto_control",
+                 level: int = logging.INFO) -> None:
+        super().__init__(level=level)
+        self._notifier = notifier
+        self._logger_name = str(logger_name)
+
+    def set_notifier(self, notifier: Optional[
+            Callable[[str, Dict[str, Any]], None]]) -> None:
+        self._notifier = notifier
+
+    def emit(self, record: logging.LogRecord) -> None:
+        notifier = self._notifier
+        if notifier is None:
+            return
+        try:
+            text = record.getMessage()
+        except (TypeError, ValueError):
+            text = str(record.msg)
+        params: Dict[str, Any] = {
+            "level": logging_level_to_mcp(record.levelno),
+            "logger": self._logger_name,
+            "data": {
+                "logger": record.name,
+                "message": text,
+                "module": record.module,
+                "func": record.funcName,
+                "line": record.lineno,
+            },
+        }
+        try:
+            notifier("notifications/message", params)
+        except (OSError, RuntimeError, ValueError):
+            # The bridge must never crash the producer.
+            pass
+
+
+__all__ = [
+    "MCPLogBridge", "logging_level_to_mcp", "mcp_level_to_logging",
+]

je_auto_control/utils/mcp_server/server.py

@@ -14,10 +14,15 @@
 from typing import Any, Callable, Dict, List, Optional, TextIO
 
 from je_auto_control.utils.logging.logging_instance import autocontrol_logger
+import logging
+
 from je_auto_control.utils.mcp_server.audit import AuditLogger
 from je_auto_control.utils.mcp_server.context import (
     OperationCancelledError, ToolCallContext,
 )
+from je_auto_control.utils.mcp_server.log_bridge import (
+    MCPLogBridge, mcp_level_to_logging,
+)
 from je_auto_control.utils.mcp_server.rate_limit import RateLimiter
 from je_auto_control.utils.mcp_server.prompts import (
     PromptProvider, default_prompt_provider,
@@ -55,6 +60,7 @@ def __init__(self, tools: Optional[List[MCPTool]] = None,
                  concurrent_tools: bool = False,
                  audit_logger: Optional[AuditLogger] = None,
                  rate_limiter: Optional[RateLimiter] = None,
+                 log_bridge: Optional[MCPLogBridge] = None,
                  ) -> None:
         registry = tools if tools is not None else build_default_tool_registry()
         self._tools: Dict[str, MCPTool] = {tool.name: tool for tool in registry}
@@ -66,6 +72,7 @@ def __init__(self, tools: Optional[List[MCPTool]] = None,
         self._audit = (audit_logger if audit_logger is not None
                         else AuditLogger())
         self._rate_limiter = rate_limiter
+        self._log_bridge = log_bridge
         self._stop = threading.Event()
         self._initialized = False
         self._notifier: Optional[Callable[[str, Dict[str, Any]], None]] = None
@@ -129,6 +136,7 @@ def serve_stdio(self, stdin: Optional[TextIO] = None,
         # Stdio always opts into concurrent tool execution so sampling
         # requests issued by tool handlers don't block the reader.
         self._concurrent_tools = True
+        self._attach_log_bridge_if_configured()
         try:
             while not self._stop.is_set():
                 line = in_stream.readline()
@@ -141,6 +149,7 @@ def serve_stdio(self, stdin: Optional[TextIO] = None,
                 if response is not None:
                     self._write_message(out_stream, response)
         finally:
+            self._detach_log_bridge_if_configured()
             self._notifier = prior_notifier
             self._writer = prior_writer
             self._concurrent_tools = prior_concurrent
@@ -163,6 +172,23 @@ def set_notifier(self,
         """
         self._notifier = notifier
 
+    def _attach_log_bridge_if_configured(self) -> None:
+        """Wire the log bridge into the project logger and notifier."""
+        if self._log_bridge is None:
+            self._log_bridge = MCPLogBridge()
+        self._log_bridge.set_notifier(self._notifier)
+        if self._log_bridge not in autocontrol_logger.handlers:
+            autocontrol_logger.addHandler(self._log_bridge)
+
+    def _detach_log_bridge_if_configured(self) -> None:
+        if self._log_bridge is None:
+            return
+        self._log_bridge.set_notifier(None)
+        try:
+            autocontrol_logger.removeHandler(self._log_bridge)
+        except ValueError:
+            pass
+
     def set_writer(self, writer: Optional[Callable[[str], None]]) -> None:
         """Install a callback used to write any outbound JSON-RPC line.
 
@@ -357,8 +383,26 @@ def _dispatch(self, msg_id: Any, method: Optional[str],
                                  for prompt in self._prompts.list()]}
         if method == "prompts/get":
             return self._handle_prompts_get(params)
+        if method == "logging/setLevel":
+            return self._handle_logging_set_level(params)
         raise _MCPError(-32601, f"Method not found: {method}")
 
+    def _handle_logging_set_level(self,
+                                  params: Dict[str, Any]) -> Dict[str, Any]:
+        name = params.get("level")
+        if not isinstance(name, str):
+            raise _MCPError(-32602, "logging/setLevel requires string 'level'")
+        level = mcp_level_to_logging(name)
+        if level is None:
+            raise _MCPError(-32602, f"unknown log level: {name!r}")
+        if self._log_bridge is None:
+            self._log_bridge = MCPLogBridge()
+        self._log_bridge.setLevel(level)
+        autocontrol_logger.setLevel(min(autocontrol_logger.level or level,
+                                         level) if autocontrol_logger.level
+                                    else level)
+        return {}
+
     def _handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
         client_version = params.get("protocolVersion", PROTOCOL_VERSION)
         client_caps = params.get("capabilities") or {}
@@ -369,6 +413,7 @@ def _handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
             "resources": {"listChanged": False, "subscribe": False},
             "prompts": {"listChanged": False},
             "sampling": {},
+            "logging": {},
         }
         if "roots" in self._client_capabilities:
             capabilities["roots"] = {"listChanged": True}

test/unit_test/headless/test_mcp_server.py

@@ -158,9 +158,10 @@ def test_serve_stdio_processes_messages_until_eof():
     stdout = io.StringIO()
     server.serve_stdio(stdin=stdin, stdout=stdout)
     out_lines = [line for line in stdout.getvalue().splitlines() if line]
-    assert len(out_lines) == 2  # initialize + tools/call (notification has no reply)
-    last = _decode(out_lines[-1])
-    assert last["result"]["content"][0]["text"] == "pong"
+    responses = [_decode(line) for line in out_lines
+                  if '"id":' in line and '"method"' not in line]
+    assert len(responses) == 2  # initialize + tools/call
+    assert responses[-1]["result"]["content"][0]["text"] == "pong"
 
 
 def test_tool_descriptor_includes_annotations():
@@ -1193,6 +1194,51 @@ def run_refresh():
     assert os.path.realpath(fs_provider.root) == os.path.realpath(str(target))
 
 
+def test_initialize_advertises_logging_capability():
+    server = MCPServer(tools=[])
+    response = _decode(server.handle_line(_request("initialize", params={})))
+    assert "logging" in response["result"]["capabilities"]
+
+
+def test_log_bridge_emits_notification_for_log_record():
+    from je_auto_control.utils.mcp_server.log_bridge import MCPLogBridge
+    captured = []
+    bridge = MCPLogBridge(
+        notifier=lambda method, params: captured.append((method, params)),
+    )
+    import logging
+    record = logging.LogRecord(
+        name="je_auto_control.tests", level=logging.WARNING,
+        pathname=__file__, lineno=10, msg="something %s", args=("happened",),
+        exc_info=None,
+    )
+    bridge.emit(record)
+    assert captured
+    method, params = captured[0]
+    assert method == "notifications/message"
+    assert params["level"] == "warning"
+    assert params["data"]["message"] == "something happened"
+
+
+def test_logging_set_level_request_updates_bridge_level():
+    from je_auto_control.utils.mcp_server.log_bridge import MCPLogBridge
+    server = MCPServer(tools=[], log_bridge=MCPLogBridge())
+    response = _decode(server.handle_line(_request("logging/setLevel", params={
+        "level": "error",
+    })))
+    assert response["result"] == {}
+    import logging
+    assert server._log_bridge.level == logging.ERROR
+
+
+def test_logging_set_level_rejects_unknown_name():
+    server = MCPServer(tools=[])
+    response = _decode(server.handle_line(_request("logging/setLevel", params={
+        "level": "telepathy",
+    })))
+    assert response["error"]["code"] == -32602
+
+
 def test_default_registry_lists_core_automation_tools():
     names = {tool.name for tool in build_default_tool_registry()}
     expected = {