fix: Make context logging functions spec-compliant - Fixes #397

src/mcp/server/mcpserver/context.py

@@ -187,7 +187,7 @@ async def elicit_url(
     async def log(
         self,
         level: Literal["debug", "info", "warning", "error"],
-        message: str,
+        message: Any,
         *,
         logger_name: str | None = None,
         extra: dict[str, Any] | None = None,
@@ -196,7 +196,7 @@ async def log(
 
         Args:
             level: Log level (debug, info, warning, error)
-            message: Log message
+            message: Log payload (any JSON-serializable type)
             logger_name: Optional logger name
             extra: Optional dictionary with additional structured data to include
         """
@@ -261,20 +261,20 @@ 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, message: Any, *, 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 info(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
+    async def info(self, message: Any, *, 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 warning(
-        self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None
+        self, message: Any, *, 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 error(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
+    async def error(self, message: Any, *, 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)

tests/client/test_logging_callback.py

@@ -53,7 +53,10 @@ async def test_tool_with_log_extra(
             level=level,
             message=message,
             logger_name=logger,
-            extra={"extra_string": extra_string, "extra_dict": extra_dict},
+            extra={
+                "extra_string": extra_string,
+                "extra_dict": extra_dict,
+            },
         )
         return True
 

tests/server/mcpserver/test_server.py

@@ -1069,6 +1069,87 @@ 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_with_structured_data(self):
+        """Test that context logging accepts structured data per MCP spec (issue #397)."""
+        mcp = MCPServer()
+
+        async def structured_logging_tool(msg: str, ctx: Context) -> str:
+            # Test with dictionary
+            await ctx.info({"status": "success", "message": msg, "count": 42})
+            # Test with list
+            await ctx.debug([1, 2, 3, "item"])
+            # Test with number
+            await ctx.warning(404)
+            # Test with boolean
+            await ctx.error(True)
+            # Test with null
+            await ctx.info(None)
+            # Test string still works (backward compatibility)
+            await ctx.info("Plain string message")
+            return f"Logged structured data for {msg}"
+
+        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", {"msg": "test"})
+                assert len(result.content) == 1
+                content = result.content[0]
+                assert isinstance(content, TextContent)
+                assert "Logged structured data for test" in content.text
+
+                # Verify all log calls were made with correct data types
+                assert mock_log.call_count == 6
+
+                # Check dictionary logging
+                mock_log.assert_any_call(
+                    level="info",
+                    data={"status": "success", "message": "test", "count": 42},
+                    logger=None,
+                    related_request_id="1",
+                )
+
+                # Check list logging
+                mock_log.assert_any_call(
+                    level="debug",
+                    data=[1, 2, 3, "item"],
+                    logger=None,
+                    related_request_id="1",
+                )
+
+                # Check number logging
+                mock_log.assert_any_call(
+                    level="warning",
+                    data=404,
+                    logger=None,
+                    related_request_id="1",
+                )
+
+                # Check boolean logging
+                mock_log.assert_any_call(
+                    level="error",
+                    data=True,
+                    logger=None,
+                    related_request_id="1",
+                )
+
+                # Check null logging
+                mock_log.assert_any_call(
+                    level="info",
+                    data=None,
+                    logger=None,
+                    related_request_id="1",
+                )
+
+                # Check string still works
+                mock_log.assert_any_call(
+                    level="info",
+                    data="Plain string message",
+                    logger=None,
+                    related_request_id="1",
+                )
+
+    @pytest.mark.anyio
     async def test_optional_context(self):
         """Test that context is optional."""
         mcp = MCPServer()