fix: make Context logging methods spec-compliant by accepting Any data type

Per the MCP specification, the data field in LoggingMessageNotificationParams
allows any JSON serializable type, not just strings. The Context.log() and
convenience methods (debug, info, warning, error) were typed as message: str,
which prevented users from logging structured data like dicts, lists, numbers, etc.

Changes:
- Change message: str to data: Any in Context.log() and all convenience methods
- Remove the extra parameter (structured data can now be passed directly as data)
- Update docstrings to document that any JSON serializable type is accepted
- Update class docstring with examples of structured data logging
- Add test for logging structured data (dict, list, number, boolean, None)

Fixes #397

Author: Jimil-Joshi

src/mcp/server/mcpserver/server.py

@@ -1122,6 +1122,11 @@ async def my_tool(x: int, ctx: Context) -> str:
         await ctx.warning("Warning message")
         await ctx.error("Error message")
 
+        # Log structured data (any JSON serializable type)
+        await ctx.info({"event": "processing", "input": x})
+        await ctx.debug(["step1", "step2", "step3"])
+        await ctx.info(42)
+
         # Report progress
         await ctx.report_progress(50, 100)
 
@@ -1272,28 +1277,25 @@ 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.
 
+        Per the MCP specification, the data can be any JSON serializable type,
+        such as a string message, a dictionary, a list, a number, or any other
+        JSON-compatible value.
+
         Args:
             level: Log level (debug, info, warning, error)
-            message: Log message
+            data: The data to be logged. Any JSON serializable type is allowed.
             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,
         )
@@ -1346,20 +1348,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:
-        """Send a debug log message."""
-        await self.log("debug", message, logger_name=logger_name, extra=extra)
+    async def debug(self, data: Any, *, logger_name: str | None = None) -> None:
+        """Send a debug log message. Data can be any JSON serializable type."""
+        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:
-        """Send an info log message."""
-        await self.log("info", message, logger_name=logger_name, extra=extra)
+    async def info(self, data: Any, *, logger_name: str | None = None) -> None:
+        """Send an info log message. Data can be any JSON serializable type."""
+        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:
-        """Send a warning log message."""
-        await self.log("warning", message, logger_name=logger_name, extra=extra)
+    async def warning(self, data: Any, *, logger_name: str | None = None) -> None:
+        """Send a warning log message. Data can be any JSON serializable type."""
+        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:
-        """Send an error log message."""
-        await self.log("error", message, logger_name=logger_name, extra=extra)
+    async def error(self, data: Any, *, logger_name: str | None = None) -> None:
+        """Send an error log message. Data can be any JSON serializable type."""
+        await self.log("error", data, logger_name=logger_name)

tests/server/mcpserver/test_server.py

@@ -1070,6 +1070,50 @@ async def logging_tool(msg: str, ctx: Context[ServerSession, None]) -> 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 per MCP spec."""
+        mcp = MCPServer()
+
+        async def structured_logging_tool(ctx: Context[ServerSession, None]) -> str:
+            # Log a dictionary
+            await ctx.info({"event": "processing", "count": 5})
+            # Log a list
+            await ctx.debug(["step1", "step2", "step3"])
+            # Log a number
+            await ctx.warning(42)
+            # Log a boolean
+            await ctx.error(True)
+            # Log None
+            await ctx.info(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 == 5
+                mock_log.assert_any_call(
+                    level="info",
+                    data={"event": "processing", "count": 5},
+                    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=True, logger=None, related_request_id="1")
+                mock_log.assert_any_call(level="info", data=None, logger=None, related_request_id="1")
+
     async def test_optional_context(self):
         """Test that context is optional."""
         mcp = MCPServer()