Spaces:
Running
Running
File size: 6,239 Bytes
5800457 5bfd7a8 5800457 5bfd7a8 5800457 5bfd7a8 5800457 5bfd7a8 5800457 5bfd7a8 5800457 5bfd7a8 5800457 5bfd7a8 5800457 5bfd7a8 5800457 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | """Comprehensive logging middleware for FastMCP servers."""
import json
import logging
from typing import Any
from .middleware import CallNext, Middleware, MiddlewareContext
class LoggingMiddleware(Middleware):
"""Middleware that provides comprehensive request and response logging.
Logs all MCP messages with configurable detail levels. Useful for debugging,
monitoring, and understanding server usage patterns.
Example:
```python
from fastmcp.server.middleware.logging import LoggingMiddleware
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
mcp = FastMCP("MyServer")
mcp.add_middleware(LoggingMiddleware())
```
"""
def __init__(
self,
logger: logging.Logger | None = None,
log_level: int = logging.INFO,
include_payloads: bool = False,
max_payload_length: int = 1000,
methods: list[str] | None = None,
):
"""Initialize logging middleware.
Args:
logger: Logger instance to use. If None, creates a logger named 'fastmcp.requests'
log_level: Log level for messages (default: INFO)
include_payloads: Whether to include message payloads in logs
max_payload_length: Maximum length of payload to log (prevents huge logs)
methods: List of methods to log. If None, logs all methods.
"""
self.logger = logger or logging.getLogger("fastmcp.requests")
self.log_level = log_level
self.include_payloads = include_payloads
self.max_payload_length = max_payload_length
self.methods = methods
def _format_message(self, context: MiddlewareContext) -> str:
"""Format a message for logging."""
parts = [
f"source={context.source}",
f"type={context.type}",
f"method={context.method or 'unknown'}",
]
if self.include_payloads and hasattr(context.message, "__dict__"):
try:
payload = json.dumps(context.message.__dict__, default=str)
if len(payload) > self.max_payload_length:
payload = payload[: self.max_payload_length] + "..."
parts.append(f"payload={payload}")
except (TypeError, ValueError):
parts.append("payload=<non-serializable>")
return " ".join(parts)
async def on_message(self, context: MiddlewareContext, call_next: CallNext) -> Any:
"""Log all messages."""
message_info = self._format_message(context)
if self.methods and context.method not in self.methods:
return await call_next(context)
self.logger.log(self.log_level, f"Processing message: {message_info}")
try:
result = await call_next(context)
self.logger.log(
self.log_level, f"Completed message: {context.method or 'unknown'}"
)
return result
except Exception as e:
self.logger.log(
logging.ERROR, f"Failed message: {context.method or 'unknown'} - {e}"
)
raise
class StructuredLoggingMiddleware(Middleware):
"""Middleware that provides structured JSON logging for better log analysis.
Outputs structured logs that are easier to parse and analyze with log
aggregation tools like ELK stack, Splunk, or cloud logging services.
Example:
```python
from fastmcp.server.middleware.logging import StructuredLoggingMiddleware
import logging
mcp = FastMCP("MyServer")
mcp.add_middleware(StructuredLoggingMiddleware())
```
"""
def __init__(
self,
logger: logging.Logger | None = None,
log_level: int = logging.INFO,
include_payloads: bool = False,
methods: list[str] | None = None,
):
"""Initialize structured logging middleware.
Args:
logger: Logger instance to use. If None, creates a logger named 'fastmcp.structured'
log_level: Log level for messages (default: INFO)
include_payloads: Whether to include message payloads in logs
methods: List of methods to log. If None, logs all methods.
"""
self.logger = logger or logging.getLogger("fastmcp.structured")
self.log_level = log_level
self.include_payloads = include_payloads
self.methods = methods
def _create_log_entry(
self, context: MiddlewareContext, event: str, **extra_fields
) -> dict:
"""Create a structured log entry."""
entry = {
"event": event,
"timestamp": context.timestamp.isoformat(),
"source": context.source,
"type": context.type,
"method": context.method,
**extra_fields,
}
if self.include_payloads and hasattr(context.message, "__dict__"):
try:
entry["payload"] = context.message.__dict__
except (TypeError, ValueError):
entry["payload"] = "<non-serializable>"
return entry
async def on_message(self, context: MiddlewareContext, call_next: CallNext) -> Any:
"""Log structured message information."""
start_entry = self._create_log_entry(context, "request_start")
if self.methods and context.method not in self.methods:
return await call_next(context)
self.logger.log(self.log_level, json.dumps(start_entry))
try:
result = await call_next(context)
success_entry = self._create_log_entry(
context,
"request_success",
result_type=type(result).__name__ if result else None,
)
self.logger.log(self.log_level, json.dumps(success_entry))
return result
except Exception as e:
error_entry = self._create_log_entry(
context,
"request_error",
error_type=type(e).__name__,
error_message=str(e),
)
self.logger.log(logging.ERROR, json.dumps(error_entry))
raise
|