Spaces:
Sleeping
Sleeping
File size: 4,749 Bytes
8b02e7c | 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 | """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()
|