ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-28 10:18:08 +02:00
Code Issues Projects Releases Packages Wiki Activity

Structured client-side logging (#1326)

Browse source
This commit is contained in:
Colin Jermain 2025-08-01 14:47:21 -04:00 • committed by GitHub
commit 7f3c3cc24b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 204 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

57
docs/clients/logging.mdx
View file

@ -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")
```
```

48
docs/servers/logging.mdx
View file

@ -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}

63
src/fastmcp/server/context.py
View file

@ -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."""

4
src/fastmcp/server/proxy.py
View file

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

40
tests/client/test_logs.py
View file

@ -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"

24
tests/test_mcp_config.py
View file

@ -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):