Fix context logging methods to accept any JSON serializable type per MCP spec

The MCP spec defines the logging data field as 'unknown' (any JSON
serializable type), but Context.log() and convenience methods (debug,
info, warning, error) only accepted str. This changes the parameter
from 'message: str' to 'data: Any' to match the spec and the existing
ServerSession.send_log_message signature.

Also removes the 'extra' parameter which was a workaround for the
str-only limitation and caused confusion (as noted in the issue).

Fixes #397

Author: Christian-Sidak

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()