From 75d2d11dd09d1c398d90be4de116c98ed4b1290b Mon Sep 17 00:00:00 2001 From: Jimil Joshi Date: Mon, 2 Mar 2026 11:45:16 +0530 Subject: [PATCH] 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 --- src/mcp/server/mcpserver/server.py | 48 +++++++++++++-------------- tests/server/mcpserver/test_server.py | 44 ++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 24 deletions(-) diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py index 9c7105a7b..929542a7a 100644 --- a/src/mcp/server/mcpserver/server.py +++ b/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) diff --git a/tests/server/mcpserver/test_server.py b/tests/server/mcpserver/test_server.py index cfbe6587b..ba8fda5fe 100644 --- a/tests/server/mcpserver/test_server.py +++ b/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()