Structured client-side logging (#1326)

docs/clients/logging.mdx

@@ -13,17 +13,39 @@ MCP servers can emit log messages to clients. The client can handle these logs t
 
 ## Log Handler
 
-Provide a `log_handler` function when creating the client:
+Provide a `log_handler` function when creating the client. For robust logging, the log messages can be integrated with Python's standard `logging` module.
 
 ```python
+import logging
 from fastmcp import Client
 from fastmcp.client.logging import LogMessage
 
+# In a real app, you might configure this in your main entry point
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+
+# Get a logger for the module where the client is used
+logger = logging.getLogger(__name__)
+
+# This mapping is useful for converting MCP level strings to Python's levels
+LOGGING_LEVEL_MAP = logging.getLevelNamesMapping()
+
 async def log_handler(message: LogMessage):
-    level = message.level.upper()
-    logger = message.logger or 'server'
-    data = message.data
-    print(f"[{level}] {logger}: {data}")
+    """
+    Handles incoming logs from the MCP server and forwards them
+    to the standard Python logging system.
+    """
+    msg = message.data.get('msg')
+    extra = message.data.get('extra')
+
+    # Convert the MCP log level to a Python log level
+    level = LOGGING_LEVEL_MAP.get(message.level.upper(), logging.INFO)
+
+    # Log the message using the standard logging library
+    logger.log(level, msg, extra=extra)
+
 
 client = Client(
     "my_mcp_server.py",
@@ -31,6 +53,16 @@ client = Client(
 )
 ```
 
+## Handling Structured Logs
+
+The `message.data` attribute is a dictionary that contains the log payload from the server. This enables structured logging, allowing you to receive rich, contextual information.
+
+The dictionary contains two keys:
+- `msg`: The string log message.
+- `extra`: A dictionary containing any extra data sent from the server.
+
+This structure is preserved even when logs are forwarded through a FastMCP proxy, making it a powerful tool for debugging complex, multi-server applications.
+
 ### Handler Parameters
 
 The `log_handler` is called every time a log message is received. It receives a `LogMessage` object:
@@ -46,8 +78,8 @@ The `log_handler` is called every time a log message is received. It receives a
       The logger name (optional, may be None)
     </ResponseField>
 
-    <ResponseField name="data" type="Any">
-      The actual log message content
+    <ResponseField name="data" type="dict">
+      The log payload, containing `msg` and `extra` keys.
     </ResponseField>
   </Expandable>
 </ResponseField>
@@ -55,12 +87,15 @@ The `log_handler` is called every time a log message is received. It receives a
 
 ```python
 async def detailed_log_handler(message: LogMessage):
+    msg = message.data.get('msg')
+    extra = message.data.get('extra')
+
     if message.level == "error":
-        print(f"ERROR: {message.data}")
+        print(f"ERROR: {msg} | Details: {extra}")
     elif message.level == "warning":
-        print(f"WARNING: {message.data}")
+        print(f"WARNING: {msg} | Details: {extra}")
     else:
-        print(f"{message.level.upper()}: {message.data}")
+        print(f"{message.level.upper()}: {msg}")
 ```
 
 ## Default Log Handling
@@ -73,4 +108,4 @@ client = Client("my_mcp_server.py")
 async with client:
     # Server logs will be emitted at DEBUG level automatically
     await client.call_tool("some_tool")
-```
+```

docs/servers/logging.mdx

@@ -53,6 +53,24 @@ async def analyze_data(data: list[float], ctx: Context) -> dict:
         raise
 ```
 
+## Structured Logging with `extra`
+
+All logging methods (`debug`, `info`, `warning`, `error`, `log`) now accept an `extra` parameter, which is a dictionary of arbitrary data. This allows you to send structured data to the client, which is useful for creating rich, queryable logs.
+
+```python
+@mcp.tool
+async def process_transaction(transaction_id: str, amount: float, ctx: Context):
+    await ctx.info(
+        f"Processing transaction {transaction_id}",
+        extra={
+            "transaction_id": transaction_id,
+            "amount": amount,
+            "currency": "USD"
+        }
+    )
+    # ... processing logic ...
+```
+
 ## Logging Methods
 
 <Card icon="code" title="Context Logging Methods">
@@ -63,6 +81,9 @@ async def analyze_data(data: list[float], ctx: Context) -> dict:
     <ResponseField name="message" type="str">
       The debug message to send to the client
     </ResponseField>
+    <ResponseField name="extra" type="dict | None" default="None">
+      Optional dictionary for structured logging data
+    </ResponseField>
   </Expandable>
 </ResponseField>
 
@@ -73,6 +94,9 @@ async def analyze_data(data: list[float], ctx: Context) -> dict:
     <ResponseField name="message" type="str">
       The information message to send to the client
     </ResponseField>
+    <ResponseField name="extra" type="dict | None" default="None">
+      Optional dictionary for structured logging data
+    </ResponseField>
   </Expandable>
 </ResponseField>
 
@@ -83,6 +107,9 @@ async def analyze_data(data: list[float], ctx: Context) -> dict:
     <ResponseField name="message" type="str">
       The warning message to send to the client
     </ResponseField>
+    <ResponseField name="extra" type="dict | None" default="None">
+      Optional dictionary for structured logging data
+    </ResponseField>
   </Expandable>
 </ResponseField>
 
@@ -93,6 +120,9 @@ async def analyze_data(data: list[float], ctx: Context) -> dict:
     <ResponseField name="message" type="str">
       The error message to send to the client
     </ResponseField>
+    <ResponseField name="extra" type="dict | None" default="None">
+      Optional dictionary for structured logging data
+    </ResponseField>
   </Expandable>
 </ResponseField>
 
@@ -111,6 +141,9 @@ async def analyze_data(data: list[float], ctx: Context) -> dict:
     <ResponseField name="logger_name" type="str | None" default="None">
       Optional custom logger name for categorizing messages
     </ResponseField>
+    <ResponseField name="extra" type="dict | None" default="None">
+      Optional dictionary for structured logging data
+    </ResponseField>
   </Expandable>
 </ResponseField>
 </Card>
@@ -153,10 +186,16 @@ Use for potentially harmful situations that don't prevent execution:
 async def validate_config(config: dict, ctx: Context) -> dict:
     """Validate configuration with warnings for deprecated options."""
     if "old_api_key" in config:
-        await ctx.warning("Using deprecated 'old_api_key' field. Please use 'api_key' instead")
+        await ctx.warning(
+            "Using deprecated 'old_api_key' field. Please use 'api_key' instead",
+            extra={"deprecated_field": "old_api_key"}
+        )
     
     if config.get("timeout", 30) > 300:
-        await ctx.warning("Timeout value is very high (>5 minutes), this may cause issues")
+        await ctx.warning(
+            "Timeout value is very high (>5 minutes), this may cause issues",
+            extra={"timeout_value": config.get("timeout")}
+        )
     
     return {"status": "valid", "warnings": "see logs"}
 ```
@@ -176,7 +215,10 @@ async def batch_process(items: list[str], ctx: Context) -> dict:
             # Process item
             successful += 1
         except Exception as e:
-            await ctx.error(f"Failed to process item '{item}': {str(e)}")
+            await ctx.error(
+                f"Failed to process item '{item}': {str(e)}",
+                extra={"failed_item": item}
+            )
             failed += 1
     
     return {"successful": successful, "failed": failed}

src/fastmcp/server/context.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import asyncio
 import copy
 import warnings
-from collections.abc import Generator
+from collections.abc import Generator, Mapping
 from contextlib import contextmanager
 from contextvars import ContextVar, Token
 from dataclasses import dataclass
@@ -47,6 +47,18 @@ _current_context: ContextVar[Context | None] = ContextVar("context", default=Non
 _flush_lock = asyncio.Lock()
 
 
+@dataclass
+class LogData:
+    """Data object for passing log arguments to client-side handlers.
+
+    This provides an interface to match the Python standard library logging,
+    for compatibility with structured logging.
+    """
+
+    msg: str
+    extra: Mapping[str, Any] | None = None
+
+
 @contextmanager
 def set_context(context: Context) -> Generator[Context, None, None]:
     token = _current_context.set(context)
@@ -184,6 +196,7 @@ class Context:
         message: str,
         level: LoggingLevel | None = None,
         logger_name: str | None = None,
+        extra: Mapping[str, Any] | None = None,
     ) -> None:
         """Send a log message to the client.
 
@@ -192,12 +205,14 @@ class Context:
             level: Optional log level. One of "debug", "info", "notice", "warning", "error", "critical",
                 "alert", or "emergency". Default is "info".
             logger_name: Optional logger name
+            extra: Optional mapping for additional arguments
         """
         if level is None:
             level = "info"
+        data = LogData(msg=message, extra=extra)
         await self.session.send_log_message(
             level=level,
-            data=message,
+            data=data,
             logger=logger_name,
             related_request_id=self.request_id,
         )
@@ -266,21 +281,49 @@ class Context:
         return self.request_context.session
 
     # Convenience methods for common log levels
-    async def debug(self, message: str, logger_name: str | None = None) -> None:
+    async def debug(
+        self,
+        message: str,
+        logger_name: str | None = None,
+        extra: Mapping[str, Any] | None = None,
+    ) -> None:
         """Send a debug log message."""
-        await self.log(level="debug", message=message, logger_name=logger_name)
+        await self.log(
+            level="debug", message=message, logger_name=logger_name, extra=extra
+        )
 
-    async def info(self, message: str, logger_name: str | None = None) -> None:
+    async def info(
+        self,
+        message: str,
+        logger_name: str | None = None,
+        extra: Mapping[str, Any] | None = None,
+    ) -> None:
         """Send an info log message."""
-        await self.log(level="info", message=message, logger_name=logger_name)
+        await self.log(
+            level="info", message=message, logger_name=logger_name, extra=extra
+        )
 
-    async def warning(self, message: str, logger_name: str | None = None) -> None:
+    async def warning(
+        self,
+        message: str,
+        logger_name: str | None = None,
+        extra: Mapping[str, Any] | None = None,
+    ) -> None:
         """Send a warning log message."""
-        await self.log(level="warning", message=message, logger_name=logger_name)
+        await self.log(
+            level="warning", message=message, logger_name=logger_name, extra=extra
+        )
 
-    async def error(self, message: str, logger_name: str | None = None) -> None:
+    async def error(
+        self,
+        message: str,
+        logger_name: str | None = None,
+        extra: Mapping[str, Any] | None = None,
+    ) -> None:
         """Send an error log message."""
-        await self.log(level="error", message=message, logger_name=logger_name)
+        await self.log(
+            level="error", message=message, logger_name=logger_name, extra=extra
+        )
 
     async def list_roots(self) -> list[Root]:
         """List the roots available to the server, as indicated by the client."""

src/fastmcp/server/proxy.py

@@ -591,7 +591,9 @@ class ProxyClient(Client[ClientTransportT]):
         A handler that forwards the log notification from the remote server to the proxy's connected clients.
         """
         ctx = get_context()
-        await ctx.log(message.data, level=message.level, logger_name=message.logger)
+        msg = message.data.get("msg")
+        extra = message.data.get("extra")
+        await ctx.log(msg, level=message.level, logger_name=message.logger, extra=extra)
 
     @classmethod
     async def default_progress_handler(

tests/client/test_logs.py

@@ -1,3 +1,5 @@
+import logging
+
 import pytest
 from mcp import LoggingLevel
 
@@ -8,10 +10,23 @@ from fastmcp.client.logging import LogMessage
 class LogHandler:
     def __init__(self):
         self.logs: list[LogMessage] = []
+        self.logger = logging.getLogger(__name__)
+        # Backwards-compatible way to get the log level mapping
+        if hasattr(logging, "getLevelNamesMapping"):
+            # For Python 3.11+
+            self.LOGGING_LEVEL_MAP = logging.getLevelNamesMapping()  # pyright: ignore [reportAttributeAccessIssue]
+        else:
+            # For older Python versions
+            self.LOGGING_LEVEL_MAP = logging._nameToLevel
 
     async def handle_log(self, message: LogMessage) -> None:
         self.logs.append(message)
 
+        level = self.LOGGING_LEVEL_MAP[message.level.upper()]
+        msg = message.data.get("msg")
+        extra = message.data.get("extra")
+        self.logger.log(level, msg, extra=extra)
+
 
 @pytest.fixture
 def fastmcp_server():
@@ -34,27 +49,42 @@ def fastmcp_server():
 
 
 class TestClientLogs:
-    async def test_log(self, fastmcp_server: FastMCP):
+    async def test_log(self, fastmcp_server: FastMCP, caplog):
+        caplog.set_level(logging.INFO, logger=__name__)
+
         log_handler = LogHandler()
         async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
             await client.call_tool("log", {})
 
         assert len(log_handler.logs) == 1
-        assert log_handler.logs[0].data == "hello?"
+        assert log_handler.logs[0].data["msg"] == "hello?"
         assert log_handler.logs[0].level == "info"
 
-    async def test_echo_log(self, fastmcp_server: FastMCP):
+        assert len(caplog.records) == 1
+        assert caplog.records[0].msg == "hello?"
+        assert caplog.records[0].levelname == "INFO"
+
+    async def test_echo_log(self, fastmcp_server: FastMCP, caplog):
+        caplog.set_level(logging.INFO, logger=__name__)
+
         log_handler = LogHandler()
         async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
             await client.call_tool("echo_log", {"message": "this is a log"})
 
             assert len(log_handler.logs) == 1
+            assert len(caplog.records) == 1
             await client.call_tool(
                 "echo_log", {"message": "this is a warning log", "level": "warning"}
             )
             assert len(log_handler.logs) == 2
+            assert len(caplog.records) == 2
 
-        assert log_handler.logs[0].data == "this is a log"
+        assert log_handler.logs[0].data["msg"] == "this is a log"
         assert log_handler.logs[0].level == "info"
-        assert log_handler.logs[1].data == "this is a warning log"
+        assert log_handler.logs[1].data["msg"] == "this is a warning log"
         assert log_handler.logs[1].level == "warning"
+
+        assert caplog.records[0].msg == "this is a log"
+        assert caplog.records[0].levelname == "INFO"
+        assert caplog.records[1].msg == "this is a warning log"
+        assert caplog.records[1].levelname == "WARNING"

tests/test_mcp_config.py

@@ -1,4 +1,5 @@
 import inspect
+import logging
 import tempfile
 from collections.abc import AsyncGenerator
 from pathlib import Path
@@ -281,10 +282,12 @@ async def test_remote_config_with_oauth_literal():
     assert isinstance(client.transport.transport.auth, OAuthClientProvider)
 
 
-async def test_multi_client_with_logging(tmp_path: Path):
+async def test_multi_client_with_logging(tmp_path: Path, caplog):
     """
     Tests that logging is properly forwarded to the ultimate client.
     """
+    caplog.set_level(logging.INFO, logger=__name__)
+
     server_script = inspect.cleandoc("""
         from fastmcp import FastMCP, Context
 
@@ -317,14 +320,31 @@ async def test_multi_client_with_logging(tmp_path: Path):
 
     MESSAGES = []
 
+    logger = logging.getLogger(__name__)
+    # Backwards-compatible way to get the log level mapping
+    if hasattr(logging, "getLevelNamesMapping"):
+        # For Python 3.11+
+        LOGGING_LEVEL_MAP = logging.getLevelNamesMapping()  # pyright: ignore [reportAttributeAccessIssue]
+    else:
+        # For older Python versions
+        LOGGING_LEVEL_MAP = logging._nameToLevel
+
     async def log_handler(message: LogMessage):
         MESSAGES.append(message)
 
+        level = LOGGING_LEVEL_MAP[message.level.upper()]
+        msg = message.data.get("msg")
+        extra = message.data.get("extra")
+        logger.log(level, msg, extra=extra)
+
     async with Client(config, log_handler=log_handler) as client:
         result = await client.call_tool("test_server_log_test", {"message": "test 42"})
         assert result.data == 42
         assert len(MESSAGES) == 1
-        assert MESSAGES[0].data == "test 42"
+        assert MESSAGES[0].data["msg"] == "test 42"
+
+        assert len(caplog.records) == 1
+        assert caplog.records[0].msg == "test 42"
 
 
 async def test_multi_client_with_transforms(tmp_path: Path):