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

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