logging_config.py•4.75 kB
"""Logging configuration for delegation MCP server."""
import logging
import sys
from typing import Any
from datetime import datetime
class StructuredFormatter(logging.Formatter):
"""Custom formatter for structured logging."""
def format(self, record: logging.LogRecord) -> str:
"""Format log record with structured information."""
# Get timestamp
timestamp = datetime.fromtimestamp(record.created).isoformat()
# Build structured log entry
parts = [
f"[{timestamp}]",
f"[{record.levelname}]",
f"[{record.name}]",
]
# Add extra context if available
if hasattr(record, "orchestrator"):
parts.append(f"[orchestrator={record.orchestrator}]")
if hasattr(record, "delegation_to"):
parts.append(f"[→{record.delegation_to}]")
if hasattr(record, "duration"):
parts.append(f"[{record.duration:.2f}s]")
# Add the message
parts.append(record.getMessage())
# Add exception info if present
if record.exc_info:
parts.append("\n" + self.formatException(record.exc_info))
return " ".join(parts)
def setup_logging(level: int = logging.INFO, verbose: bool = False) -> None:
"""
Setup logging configuration.
Args:
level: Logging level (DEBUG, INFO, WARNING, ERROR)
verbose: Enable verbose output with structured logging
"""
# Get root logger
root_logger = logging.getLogger()
root_logger.setLevel(level)
# Remove existing handlers
root_logger.handlers.clear()
# Create console handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(level)
# Set formatter
if verbose:
formatter = StructuredFormatter()
else:
formatter = logging.Formatter(
"%(levelname)s - %(name)s - %(message)s"
)
console_handler.setFormatter(formatter)
root_logger.addHandler(console_handler)
# Set specific log levels for dependencies
logging.getLogger("asyncio").setLevel(logging.WARNING)
logging.getLogger("urllib3").setLevel(logging.WARNING)
class DelegationLogger:
"""Logger with delegation-specific context."""
def __init__(self, name: str = "delegation_mcp"):
self.logger = logging.getLogger(name)
def delegation_start(
self,
orchestrator: str,
query: str,
delegated_to: str | None = None
) -> None:
"""Log delegation start."""
extra = {"orchestrator": orchestrator}
if delegated_to:
extra["delegation_to"] = delegated_to
msg = f"Delegating query to {delegated_to}"
else:
msg = f"Processing query with {orchestrator}"
self.logger.info(msg, extra=extra)
def delegation_success(
self,
orchestrator: str,
delegated_to: str | None,
duration: float,
) -> None:
"""Log successful delegation."""
target = delegated_to or orchestrator
extra = {
"orchestrator": orchestrator,
"duration": duration,
}
if delegated_to:
extra["delegation_to"] = delegated_to
self.logger.info(f"✓ Delegation completed successfully", extra=extra)
def delegation_failure(
self,
orchestrator: str,
delegated_to: str | None,
error: str,
duration: float,
) -> None:
"""Log failed delegation."""
target = delegated_to or orchestrator
extra = {
"orchestrator": orchestrator,
"duration": duration,
}
if delegated_to:
extra["delegation_to"] = delegated_to
self.logger.error(f"✗ Delegation failed: {error}", extra=extra)
def retry_attempt(self, attempt: int, max_retries: int, error: str) -> None:
"""Log retry attempt."""
self.logger.warning(
f"Retry attempt {attempt}/{max_retries}: {error}"
)
def timeout(self, orchestrator: str, timeout_seconds: float) -> None:
"""Log timeout."""
self.logger.error(
f"Timeout after {timeout_seconds}s",
extra={"orchestrator": orchestrator}
)
def rule_match(self, pattern: str, delegate_to: str, confidence: int = 100) -> None:
"""Log rule match."""
self.logger.info(
f"Rule matched: '{pattern}' → {delegate_to} (confidence: {confidence}%)"
)
def no_rule_match(self, query: str) -> None:
"""Log when no rule matches."""
self.logger.debug(f"No delegation rule matched for query")
# Global logger instance
delegation_logger = DelegationLogger()