feat(mcp): surface MCP server log notifications in agent.log (#57416)

Port from anomalyco/opencode#34529: MCP servers can emit
notifications/message logging notifications (RFC 5424 levels), but the
MCP SDK's default logging_callback silently discards them — server-side
warnings/errors during tool calls were invisible.

- tools/mcp_tool.py: pass a logging_callback to every ClientSession
  (stdio, SSE, streamable HTTP old+new API paths via the shared
  sampling_kwargs sites), mapping the 8 MCP log levels onto Python
  logging levels and tagging entries with [server/logger] origin.
- JSON-serialize non-string payloads, cap at 2000 chars so a chatty
  server can't flood agent.log, never raise from the handler.
- Gated on SDK support (_check_logging_callback_support) mirroring the
  existing message_handler gate for old SDK versions.
- tests/tools/test_mcp_server_log_notifications.py: 10 tests covering
  level mapping, origin tagging, JSON payloads, truncation, and the
  never-raise contract.

tests/tools/test_mcp_server_log_notifications.py

@@ -0,0 +1,126 @@
+"""Tests for MCP server log notification handling (port of anomalyco/opencode#34529).
+
+MCP servers can emit ``notifications/message`` logging notifications
+(RFC 5424 syslog levels). The MCP SDK's default ``logging_callback``
+silently discards them; Hermes now passes ``_make_logging_callback()``
+to ``ClientSession`` so server-side diagnostics land in agent.log,
+tagged with the server name.
+"""
+
+import logging
+from types import SimpleNamespace
+
+import pytest
+
+from tools.mcp_tool import (
+    _MCP_LOG_LEVEL_MAP,
+    _MCP_LOGGING_CALLBACK_SUPPORTED,
+    MCPServerTask,
+)
+
+
+def _params(level="info", data="hello", logger_name=None):
+    return SimpleNamespace(level=level, data=data, logger=logger_name)
+
+
+class TestLogLevelMap:
+    def test_all_mcp_levels_mapped(self):
+        # MCP spec (RFC 5424) defines these eight levels.
+        for lvl in ("debug", "info", "notice", "warning",
+                    "error", "critical", "alert", "emergency"):
+            assert lvl in _MCP_LOG_LEVEL_MAP
+
+    def test_severity_ordering(self):
+        assert _MCP_LOG_LEVEL_MAP["debug"] == logging.DEBUG
+        assert _MCP_LOG_LEVEL_MAP["notice"] == logging.INFO
+        assert _MCP_LOG_LEVEL_MAP["warning"] == logging.WARNING
+        assert _MCP_LOG_LEVEL_MAP["emergency"] == logging.ERROR
+
+
+class TestLoggingCallback:
+    @pytest.mark.asyncio
+    async def test_routes_to_hermes_logger_with_server_tag(self, caplog):
+        server = MCPServerTask("log_srv")
+        callback = server._make_logging_callback()
+        with caplog.at_level(logging.INFO, logger="tools.mcp_tool"):
+            await callback(_params(level="info", data="server started"))
+        assert any(
+            "MCP server log [log_srv]: server started" in rec.getMessage()
+            for rec in caplog.records
+        )
+
+    @pytest.mark.asyncio
+    async def test_includes_sub_logger_name(self, caplog):
+        server = MCPServerTask("log_srv")
+        callback = server._make_logging_callback()
+        with caplog.at_level(logging.WARNING, logger="tools.mcp_tool"):
+            await callback(_params(level="warning", data="rate limited",
+                                   logger_name="http"))
+        assert any(
+            "MCP server log [log_srv/http]: rate limited" in rec.getMessage()
+            and rec.levelno == logging.WARNING
+            for rec in caplog.records
+        )
+
+    @pytest.mark.asyncio
+    async def test_error_family_maps_to_error_level(self, caplog):
+        server = MCPServerTask("log_srv")
+        callback = server._make_logging_callback()
+        with caplog.at_level(logging.ERROR, logger="tools.mcp_tool"):
+            for lvl in ("error", "critical", "alert", "emergency"):
+                await callback(_params(level=lvl, data=f"boom-{lvl}"))
+        errors = [r for r in caplog.records if r.levelno == logging.ERROR]
+        assert len(errors) == 4
+
+    @pytest.mark.asyncio
+    async def test_non_string_data_is_json_serialized(self, caplog):
+        server = MCPServerTask("log_srv")
+        callback = server._make_logging_callback()
+        with caplog.at_level(logging.INFO, logger="tools.mcp_tool"):
+            await callback(_params(data={"event": "connect", "port": 8080}))
+        assert any(
+            '"event": "connect"' in rec.getMessage() for rec in caplog.records
+        )
+
+    @pytest.mark.asyncio
+    async def test_unknown_level_defaults_to_info(self, caplog):
+        server = MCPServerTask("log_srv")
+        callback = server._make_logging_callback()
+        with caplog.at_level(logging.INFO, logger="tools.mcp_tool"):
+            await callback(_params(level="bogus", data="odd level"))
+        assert any(
+            rec.levelno == logging.INFO and "odd level" in rec.getMessage()
+            for rec in caplog.records
+        )
+
+    @pytest.mark.asyncio
+    async def test_oversized_payload_truncated(self, caplog):
+        server = MCPServerTask("log_srv")
+        callback = server._make_logging_callback()
+        with caplog.at_level(logging.INFO, logger="tools.mcp_tool"):
+            await callback(_params(data="x" * 10_000))
+        msg = next(
+            rec.getMessage() for rec in caplog.records
+            if "MCP server log" in rec.getMessage()
+        )
+        assert "... [truncated]" in msg
+        assert len(msg) < 3000
+
+    @pytest.mark.asyncio
+    async def test_handler_never_raises(self):
+        server = MCPServerTask("log_srv")
+        callback = server._make_logging_callback()
+        # A params object missing every attribute must not blow up the
+        # SDK's notification dispatch loop.
+        await callback(object())
+
+
+class TestSDKSupportGate:
+    def test_current_sdk_supports_logging_callback(self):
+        # The pinned MCP SDK in this repo supports logging_callback; if this
+        # starts failing after an SDK downgrade the feature silently degrades
+        # (by design), but we want to know.
+        import inspect
+        from mcp import ClientSession
+        expected = "logging_callback" in inspect.signature(ClientSession).parameters
+        assert _MCP_LOGGING_CALLBACK_SUPPORTED == expected

tools/mcp_tool.py

@@ -280,6 +280,38 @@ _MCP_MESSAGE_HANDLER_SUPPORTED = _check_message_handler_support()
 if _MCP_AVAILABLE and not _MCP_MESSAGE_HANDLER_SUPPORTED:
     logger.debug("MCP SDK does not support message_handler -- dynamic tool discovery disabled")
 
+
+def _check_logging_callback_support() -> bool:
+    """Check if ClientSession accepts the ``logging_callback`` kwarg.
+
+    Mirrors ``_check_message_handler_support`` for backward compatibility
+    with older MCP SDK versions.  Without a logging_callback, the SDK's
+    default handler silently discards every ``notifications/message`` a
+    server emits, so server-side diagnostics never reach Hermes' logs.
+    """
+    if not _MCP_AVAILABLE:
+        return False
+    try:
+        return "logging_callback" in inspect.signature(ClientSession).parameters
+    except (TypeError, ValueError):
+        return False
+
+
+_MCP_LOGGING_CALLBACK_SUPPORTED = _check_logging_callback_support()
+
+# MCP logging levels (RFC 5424 syslog severities) -> Python logging levels.
+# Port of anomalyco/opencode#34529's serverLog mapping.
+_MCP_LOG_LEVEL_MAP = {
+    "debug": logging.DEBUG,
+    "info": logging.INFO,
+    "notice": logging.INFO,
+    "warning": logging.WARNING,
+    "error": logging.ERROR,
+    "critical": logging.ERROR,
+    "alert": logging.ERROR,
+    "emergency": logging.ERROR,
+}
+
 # ---------------------------------------------------------------------------
 # Constants
 # ---------------------------------------------------------------------------
@@ -1539,6 +1571,40 @@ class MCPServerTask:
         task.add_done_callback(self._pending_refresh_tasks.discard)
         return task
 
+    def _make_logging_callback(self):
+        """Build a ``logging_callback`` for ``ClientSession``.
+
+        Routes MCP ``notifications/message`` log notifications from the
+        server into Hermes' logging (agent.log via hermes_logging), tagged
+        with the server name.  Without this, the SDK's default callback
+        silently discards them, so server-side warnings/errors during a
+        tool call were invisible.  Port of anomalyco/opencode#34529.
+        """
+        async def _on_log(params):
+            try:
+                level = _MCP_LOG_LEVEL_MAP.get(
+                    str(getattr(params, "level", "info")).lower(), logging.INFO,
+                )
+                data = getattr(params, "data", None)
+                if not isinstance(data, str):
+                    try:
+                        data = json.dumps(data, ensure_ascii=False, default=str)
+                    except (TypeError, ValueError):
+                        data = str(data)
+                # Cap pathological payloads so a chatty/broken server can't
+                # flood agent.log with megabyte lines.
+                if len(data) > 2000:
+                    data = data[:2000] + "... [truncated]"
+                logger_name = getattr(params, "logger", None)
+                origin = f"{self.name}/{logger_name}" if logger_name else self.name
+                logger.log(level, "MCP server log [%s]: %s", origin, data)
+            except Exception:
+                logger.debug(
+                    "Failed to handle MCP log notification from '%s'",
+                    self.name, exc_info=True,
+                )
+        return _on_log
+
     def _make_message_handler(self):
         """Build a ``message_handler`` callback for ``ClientSession``.
 
@@ -1864,6 +1930,8 @@ class MCPServerTask:
             sampling_kwargs.update(self._elicitation.session_kwargs())
         if _MCP_NOTIFICATION_TYPES and _MCP_MESSAGE_HANDLER_SUPPORTED:
             sampling_kwargs["message_handler"] = self._make_message_handler()
+        if _MCP_LOGGING_CALLBACK_SUPPORTED:
+            sampling_kwargs["logging_callback"] = self._make_logging_callback()
 
         # Snapshot child PIDs before spawning so we can track the new one.
         pids_before = _snapshot_child_pids()
@@ -2079,6 +2147,8 @@ class MCPServerTask:
             sampling_kwargs.update(self._elicitation.session_kwargs())
         if _MCP_NOTIFICATION_TYPES and _MCP_MESSAGE_HANDLER_SUPPORTED:
             sampling_kwargs["message_handler"] = self._make_message_handler()
+        if _MCP_LOGGING_CALLBACK_SUPPORTED:
+            sampling_kwargs["logging_callback"] = self._make_logging_callback()
 
         # SSE transport (for MCP servers that implement the SSE transport protocol
         # rather than Streamable HTTP). Configure with ``transport: sse`` in the