fix: make context logging methods accept Any data type per MCP spec

src/mcp/server/mcpserver/context.py

@@ -187,22 +187,29 @@ 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,
     ) -> None:
         """Send a log message to the client.
 
+        Per the MCP spec, the data to be logged can be any JSON-serializable type
+        (string, dict, list, number, bool, etc.), not just strings.
+
         Args:
             level: Log level (debug, info, warning, error)
-            message: Log message
+            message: Any JSON-serializable data to log
             logger_name: Optional logger name
-            extra: Optional dictionary with additional structured data to include
+            extra: Optional dictionary with additional structured data to include.
+                When provided, data is wrapped in a dict with the extra fields merged in.
         """
 
         if extra:
-            log_data = {"message": message, **extra}
+            if isinstance(message, dict):
+                log_data = {**message, **extra}
+            else:
+                log_data = {"message": message, **extra}
         else:
             log_data = message
 
@@ -261,20 +268,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/server/mcpserver/test_server.py

@@ -1078,6 +1078,70 @@ 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_any_data(self):
+        """Test that context logging methods accept any JSON-serializable data per MCP spec."""
+        mcp = MCPServer()
+
+        async def logging_any_tool(ctx: Context) -> str:
+            await ctx.info({"event": "user_login", "user_id": 42})
+            await ctx.debug(["step1", "step2", "step3"])
+            await ctx.warning(12345)
+            await ctx.error({"code": 500, "details": {"reason": "timeout"}})
+            return "done"
+
+        mcp.add_tool(logging_any_tool)
+
+        with patch("mcp.server.session.ServerSession.send_log_message") as mock_log:
+            async with Client(mcp) as client:
+                result = await client.call_tool("logging_any_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": 42},
+                    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=12345, logger=None, related_request_id="1"
+                )
+                mock_log.assert_any_call(
+                    level="error",
+                    data={"code": 500, "details": {"reason": "timeout"}},
+                    logger=None,
+                    related_request_id="1",
+                )
+
+    async def test_context_logging_dict_with_extra(self):
+        """Test that dict data is merged with extra fields."""
+        mcp = MCPServer()
+
+        async def logging_extra_tool(ctx: Context) -> str:
+            await ctx.info({"event": "request"}, extra={"trace_id": "abc123"})
+            return "done"
+
+        mcp.add_tool(logging_extra_tool)
+
+        with patch("mcp.server.session.ServerSession.send_log_message") as mock_log:
+            async with Client(mcp) as client:
+                await client.call_tool("logging_extra_tool", {})
+                mock_log.assert_any_call(
+                    level="info",
+                    data={"event": "request", "trace_id": "abc123"},
+                    logger=None,
+                    related_request_id="1",
+                )
+
     async def test_optional_context(self):
         """Test that context is optional."""
         mcp = MCPServer()