Fix context logging methods to accept Any type per MCP spec

src/mcp/server/mcpserver/context.py

@@ -187,28 +187,22 @@ async def elicit_url(
     async def log(
         self,
         level: Literal["debug", "info", "warning", "error"],
-        message: str,
+        data: Any,
         *,
         logger_name: str | None = None,
-        extra: dict[str, Any] | None = None,
     ) -> None:
         """Send a log message to the client.
 
         Args:
             level: Log level (debug, info, warning, error)
-            message: Log message
+            data: The data to be logged. Any JSON serializable type is allowed
+                (string, dict, list, number, bool, None, etc.)
             logger_name: Optional logger name
-            extra: Optional dictionary with additional structured data to include
         """
 
-        if extra:
-            log_data = {"message": message, **extra}
-        else:
-            log_data = message
-
         await self.request_context.session.send_log_message(
             level=level,
-            data=log_data,
+            data=data,
             logger=logger_name,
             related_request_id=self.request_id,
         )
@@ -261,20 +255,18 @@ async def close_standalone_sse_stream(self) -> None:
             await self._request_context.close_standalone_sse_stream()
 
     # Convenience methods for common log levels
-    async def debug(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
+    async def debug(self, data: Any, *, logger_name: str | None = None) -> None:
         """Send a debug log message."""
-        await self.log("debug", message, logger_name=logger_name, extra=extra)
+        await self.log("debug", data, logger_name=logger_name)
 
-    async def info(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
+    async def info(self, data: Any, *, logger_name: str | None = None) -> None:
         """Send an info log message."""
-        await self.log("info", message, logger_name=logger_name, extra=extra)
+        await self.log("info", data, logger_name=logger_name)
 
-    async def warning(
-        self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None
-    ) -> None:
+    async def warning(self, data: Any, *, logger_name: str | None = None) -> None:
         """Send a warning log message."""
-        await self.log("warning", message, logger_name=logger_name, extra=extra)
+        await self.log("warning", data, logger_name=logger_name)
 
-    async def error(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
+    async def error(self, data: Any, *, logger_name: str | None = None) -> None:
         """Send an error log message."""
-        await self.log("error", message, logger_name=logger_name, extra=extra)
+        await self.log("error", data, logger_name=logger_name)

tests/client/test_logging_callback.py

@@ -30,30 +30,27 @@ async def test_tool() -> bool:
         # The actual tool is very simple and just returns True
         return True
 
-    # Create a function that can send a log notification
+    # Create a function that can send a log notification with a string
     @server.tool("test_tool_with_log")
     async def test_tool_with_log(
         message: str, level: Literal["debug", "info", "warning", "error"], logger: str, ctx: Context
     ) -> bool:
         """Send a log notification to the client."""
-        await ctx.log(level=level, message=message, logger_name=logger)
+        await ctx.log(level=level, data=message, logger_name=logger)
         return True
 
-    @server.tool("test_tool_with_log_extra")
-    async def test_tool_with_log_extra(
-        message: str,
+    # Create a function that can send structured data as a log notification
+    @server.tool("test_tool_with_structured_log")
+    async def test_tool_with_structured_log(
         level: Literal["debug", "info", "warning", "error"],
         logger: str,
-        extra_string: str,
-        extra_dict: dict[str, Any],
         ctx: Context,
     ) -> bool:
-        """Send a log notification to the client with extra fields."""
+        """Send a structured log notification to the client."""
         await ctx.log(
             level=level,
-            message=message,
+            data={"message": "Test log message", "count": 42, "tags": ["a", "b"]},
             logger_name=logger,
-            extra={"extra_string": extra_string, "extra_dict": extra_dict},
         )
         return True
 
@@ -75,7 +72,7 @@ async def message_handler(
         assert isinstance(result.content[0], TextContent)
         assert result.content[0].text == "true"
 
-        # Now send a log message via our tool
+        # Now send a string log message via our tool
         log_result = await client.call_tool(
             "test_tool_with_log",
             {
@@ -84,30 +81,30 @@ async def message_handler(
                 "logger": "test_logger",
             },
         )
-        log_result_with_extra = await client.call_tool(
-            "test_tool_with_log_extra",
+        # Send a structured log message
+        log_result_structured = await client.call_tool(
+            "test_tool_with_structured_log",
             {
-                "message": "Test log message",
                 "level": "info",
                 "logger": "test_logger",
-                "extra_string": "example",
-                "extra_dict": {"a": 1, "b": 2, "c": 3},
             },
         )
         assert log_result.is_error is False
-        assert log_result_with_extra.is_error is False
+        assert log_result_structured.is_error is False
         assert len(logging_collector.log_messages) == 2
-        # Create meta object with related_request_id added dynamically
+
+        # Verify string log
         log = logging_collector.log_messages[0]
         assert log.level == "info"
         assert log.logger == "test_logger"
         assert log.data == "Test log message"
 
-        log_with_extra = logging_collector.log_messages[1]
-        assert log_with_extra.level == "info"
-        assert log_with_extra.logger == "test_logger"
-        assert log_with_extra.data == {
+        # Verify structured log
+        log_structured = logging_collector.log_messages[1]
+        assert log_structured.level == "info"
+        assert log_structured.logger == "test_logger"
+        assert log_structured.data == {
             "message": "Test log message",
-            "extra_string": "example",
-            "extra_dict": {"a": 1, "b": 2, "c": 3},
+            "count": 42,
+            "tags": ["a", "b"],
         }

tests/server/mcpserver/test_server.py

@@ -1078,6 +1078,43 @@ async def logging_tool(msg: str, ctx: Context) -> str:
                 mock_log.assert_any_call(level="warning", data="Warning message", logger=None, related_request_id="1")
                 mock_log.assert_any_call(level="error", data="Error message", logger=None, related_request_id="1")
 
+    async def test_context_logging_structured_data(self):
+        """Test that context logging methods accept any JSON serializable type."""
+        mcp = MCPServer()
+
+        async def structured_logging_tool(ctx: Context) -> str:
+            await ctx.info({"event": "user_login", "user_id": 123})
+            await ctx.debug(["step1", "step2", "step3"])
+            await ctx.warning(42)
+            await ctx.error(None)
+            return "done"
+
+        mcp.add_tool(structured_logging_tool)
+
+        with patch("mcp.server.session.ServerSession.send_log_message") as mock_log:
+            async with Client(mcp) as client:
+                result = await client.call_tool("structured_logging_tool", {})
+                assert len(result.content) == 1
+                content = result.content[0]
+                assert isinstance(content, TextContent)
+                assert content.text == "done"
+
+                assert mock_log.call_count == 4
+                mock_log.assert_any_call(
+                    level="info",
+                    data={"event": "user_login", "user_id": 123},
+                    logger=None,
+                    related_request_id="1",
+                )
+                mock_log.assert_any_call(
+                    level="debug",
+                    data=["step1", "step2", "step3"],
+                    logger=None,
+                    related_request_id="1",
+                )
+                mock_log.assert_any_call(level="warning", data=42, logger=None, related_request_id="1")
+                mock_log.assert_any_call(level="error", data=None, logger=None, related_request_id="1")
+
     async def test_optional_context(self):
         """Test that context is optional."""
         mcp = MCPServer()