Spaces:
Sleeping
Sleeping
| """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() | |