Pathfinder MCP Server

"""Structured logging for Pathfinder MCP Server.""" import json import logging import sys from datetime import datetime, timezone from typing import Any class JSONFormatter(logging.Formatter): """JSON formatter for structured logging.""" def format(self, record: logging.LogRecord) -> str: """Format log record as JSON.""" log_obj: dict[str, Any] = { "timestamp": datetime.now(timezone.utc).isoformat(), "level": record.levelname, "message": record.getMessage(), "logger": record.name, } # Add extra fields if present if hasattr(record, "session_id"): log_obj["session_id"] = record.session_id if hasattr(record, "phase"): log_obj["phase"] = record.phase if hasattr(record, "tool"): log_obj["tool"] = record.tool if hasattr(record, "event"): log_obj["event"] = record.event # Add exception info if present if record.exc_info: log_obj["exception"] = self.formatException(record.exc_info) return json.dumps(log_obj) def get_logger(name: str = "pathfinder") -> logging.Logger: """Get configured logger instance. Args: name: Logger name Returns: Configured logger """ logger = logging.getLogger(name) # Only configure if not already configured if not logger.handlers: logger.setLevel(logging.DEBUG) # Output to stderr (stdio transport compatible) handler = logging.StreamHandler(sys.stderr) handler.setFormatter(JSONFormatter()) logger.addHandler(handler) return logger # Module-level logger logger = get_logger() def log_phase_transition( session_id: str, from_phase: str, to_phase: str, **extra: Any ) -> None: """Log a phase transition event.""" logger.info( f"Phase transition: {from_phase} -> {to_phase}", extra={ "session_id": session_id, "phase": to_phase, "event": "phase_transition", "from_phase": from_phase, **extra, }, ) def log_tool_call(tool_name: str, session_id: str | None = None, **extra: Any) -> None: """Log a tool invocation.""" logger.debug( f"Tool called: {tool_name}", extra={ "tool": tool_name, "session_id": session_id, "event": "tool_call", **extra, }, ) def log_tool_result( tool_name: str, success: bool, session_id: str | None = None, error: str | None = None, **extra: Any, ) -> None: """Log a tool result.""" level = logging.INFO if success else logging.ERROR msg = f"Tool completed: {tool_name}" if success else f"Tool failed: {tool_name}" logger.log( level, msg, extra={ "tool": tool_name, "session_id": session_id, "event": "tool_result", "success": success, "error": error, **extra, }, ) def log_compaction( session_id: str, pre_tokens: int, post_tokens: int, **extra: Any ) -> None: """Log a context compaction event.""" logger.info( f"Context compacted: {pre_tokens} -> {post_tokens} tokens", extra={ "session_id": session_id, "event": "compaction", "pre_tokens": pre_tokens, "post_tokens": post_tokens, **extra, }, ) def log_error( message: str, session_id: str | None = None, tool: str | None = None, **extra: Any, ) -> None: """Log an error.""" logger.error( message, extra={ "session_id": session_id, "tool": tool, "event": "error", **extra, }, )