align loggers

Author: Zvi Fried

src/mcp_as_a_judge/coding_task_manager.py

@@ -6,15 +6,14 @@
 """
 
 import json
-import logging
 import time
-from typing import Optional
 
 from mcp_as_a_judge.db.conversation_history_service import ConversationHistoryService
+from mcp_as_a_judge.logging_config import get_logger
 from mcp_as_a_judge.models.task_metadata import TaskMetadata, TaskState
 
-# Set up logger directly to avoid circular imports
-logger = logging.getLogger(__name__)
+# Set up logger using custom get_logger function
+logger = get_logger(__name__)
 
 
 async def create_new_coding_task(
@@ -40,7 +39,7 @@ async def create_new_coding_task(
         New TaskMetadata instance
     """
     logger.info(f"📝 Creating new coding task: {task_title}")
-    
+
     # Create new TaskMetadata with auto-generated UUID
     task_metadata = TaskMetadata(
         title=task_title,
@@ -49,11 +48,11 @@ async def create_new_coding_task(
         state=TaskState.CREATED,  # Default state for new tasks
         tags=tags,
     )
-    
+
     # Add initial requirements to history if provided
     if user_requirements:
         task_metadata.update_requirements(user_requirements, source="initial")
-    
+
     logger.info(f"✅ Created new task metadata: {task_metadata.task_id}")
     return task_metadata
 
@@ -63,8 +62,8 @@ async def update_existing_coding_task(
     user_request: str,
     task_title: str,
     task_description: str,
-    user_requirements: Optional[str],
-    state: Optional[TaskState],
+    user_requirements: str | None,
+    state: TaskState | None,
     tags: list[str],
     conversation_service: ConversationHistoryService,
 ) -> TaskMetadata:
@@ -88,46 +87,46 @@ async def update_existing_coding_task(
         ValueError: If task not found or invalid state transition
     """
     logger.info(f"📝 Updating existing coding task: {task_id}")
-    
+
     # Load existing task metadata from conversation history
     existing_metadata = await load_task_metadata_from_history(
         task_id=task_id,
         conversation_service=conversation_service,
     )
-    
+
     if not existing_metadata:
         raise ValueError(f"Task not found: {task_id}")
-    
+
     # Update mutable fields
     existing_metadata.title = task_title
     existing_metadata.description = task_description
     existing_metadata.tags = tags
     existing_metadata.updated_at = int(time.time())
-    
+
     # Update requirements if provided
     if user_requirements is not None:
         existing_metadata.update_requirements(user_requirements, source="update")
-    
+
     # Update state if provided (with validation)
     if state is not None:
         validate_state_transition(existing_metadata.state, state)
         existing_metadata.update_state(state)
-    
+
     logger.info(f"✅ Updated task metadata: {task_id}")
     return existing_metadata
 
 
 async def load_task_metadata_from_history(
     task_id: str,
     conversation_service: ConversationHistoryService,
-) -> Optional[TaskMetadata]:
+) -> TaskMetadata | None:
     """
     Load TaskMetadata from conversation history using task_id as primary key.
-    
+
     Args:
         task_id: Task ID to load
         conversation_service: Conversation service
-        
+
     Returns:
         TaskMetadata if found, None otherwise
     """
@@ -136,7 +135,7 @@ async def load_task_metadata_from_history(
         conversation_history = await conversation_service.get_conversation_history(
             session_id=task_id
         )
-        
+
         # Look for the most recent task metadata record
         for record in reversed(conversation_history):
             if record.source == "set_coding_task" and "task_metadata" in record.output:
@@ -145,9 +144,9 @@ async def load_task_metadata_from_history(
                 if "current_task_metadata" in output_data:
                     metadata_dict = output_data["current_task_metadata"]
                     return TaskMetadata.model_validate(metadata_dict)
-        
+
         return None
-        
+
     except Exception as e:
         logger.warning(f"⚠️ Failed to load task metadata from history: {e}")
         return None
@@ -161,7 +160,7 @@ async def save_task_metadata_to_history(
 ) -> None:
     """
     Save TaskMetadata to conversation history using task_id as primary key.
-    
+
     Args:
         task_metadata: Task metadata to save
         user_request: Original user request
@@ -180,9 +179,9 @@ async def save_task_metadata_to_history(
                 "timestamp": int(time.time()),
             }),
         )
-        
+
         logger.info(f"💾 Saved task metadata to conversation history: {task_metadata.task_id}")
-        
+
     except Exception as e:
         logger.error(f"❌ Failed to save task metadata to history: {e}")
         # Don't raise - this is not critical for tool operation
@@ -191,11 +190,11 @@ async def save_task_metadata_to_history(
 def validate_state_transition(current_state: TaskState, new_state: TaskState) -> None:
     """
     Validate that the state transition is allowed.
-    
+
     Args:
         current_state: Current TaskState
         new_state: Requested new TaskState
-        
+
     Raises:
         ValueError: If transition is not allowed
     """
@@ -210,7 +209,7 @@ def validate_state_transition(current_state: TaskState, new_state: TaskState) ->
         TaskState.BLOCKED: [TaskState.CREATED, TaskState.PLANNING, TaskState.PLAN_APPROVED, TaskState.IMPLEMENTING, TaskState.REVIEW_READY, TaskState.CANCELLED],
         TaskState.CANCELLED: [],  # No transitions from cancelled state
     }
-    
+
     if new_state not in valid_transitions.get(current_state, []):
         raise ValueError(
             f"Invalid state transition: {current_state.value} → {new_state.value}. "

src/mcp_as_a_judge/logging_config.py

@@ -88,13 +88,16 @@ def configure_application_loggers(level: int = logging.INFO) -> None:
     # List of application-specific loggers to configure
     app_loggers = [
         "mcp_as_a_judge.server",
+        "mcp_as_a_judge.server_helpers",
         "mcp_as_a_judge.conversation_history_service",
         "mcp_as_a_judge.db.conversation_history_service",
         "mcp_as_a_judge.db.providers.in_memory",
         "mcp_as_a_judge.db.providers.sqlite_provider",
         "mcp_as_a_judge.messaging",
         "mcp_as_a_judge.llm_client",
         "mcp_as_a_judge.config",
+        "mcp_as_a_judge.workflow.workflow_guidance",
+        "mcp_as_a_judge.coding_task_manager",
     ]
 
     # Set level for each application logger

src/mcp_as_a_judge/server_helpers.py

@@ -13,6 +13,7 @@
 from mcp_as_a_judge.constants import MAX_TOKENS
 from mcp_as_a_judge.llm_client import llm_manager
 from mcp_as_a_judge.llm_integration import load_llm_config_from_env
+from mcp_as_a_judge.logging_config import get_logger
 from mcp_as_a_judge.messaging.llm_provider import llm_provider
 from mcp_as_a_judge.prompt_loader import create_separate_messages
 
@@ -27,17 +28,18 @@ def initialize_llm_configuration() -> None:
 
     This function loads LLM configuration from environment variables and
     configures the LLM manager if a valid configuration is found.
-    Prints status messages to inform users about the configuration state.
+    Logs status messages to inform users about the configuration state.
     """
+    logger = get_logger(__name__)
     llm_config = load_llm_config_from_env()
     if llm_config:
         llm_manager.configure(llm_config)
         vendor_name = llm_config.vendor.value if llm_config.vendor else "unknown"
-        print(
-            f"LLM fallback configured: {vendor_name} with model {llm_config.model_name}"
+        logger.info(
+            f"🔧 LLM fallback configured: {vendor_name} with model {llm_config.model_name}"
         )
     else:
-        print("No LLM API key found in environment. MCP sampling will be required.")
+        logger.info("🔧 No LLM API key found in environment. MCP sampling will be required.")
 
 
 def extract_json_from_response(response_text: str) -> str:

src/mcp_as_a_judge/workflow/workflow_guidance.py

@@ -7,19 +7,18 @@
 """
 
 import json
-from typing import Any, Dict, Optional
+from typing import Any
 
 from pydantic import BaseModel, Field
 
-import logging
-
 from mcp_as_a_judge.constants import MAX_TOKENS
 from mcp_as_a_judge.db.conversation_history_service import ConversationHistoryService
+from mcp_as_a_judge.logging_config import get_logger
 from mcp_as_a_judge.messaging.llm_provider import llm_provider
 from mcp_as_a_judge.models.task_metadata import TaskMetadata, TaskState
 
-# Set up logger directly to avoid circular imports
-logger = logging.getLogger(__name__)
+# Set up logger using custom get_logger function
+logger = get_logger(__name__)
 
 
 
@@ -34,7 +33,7 @@ class WorkflowGuidance(BaseModel):
 
     Compatible with the original WorkflowGuidance model from models.py.
     """
-    next_tool: Optional[str] = Field(
+    next_tool: str | None = Field(
         description="Next tool to call, or None if workflow complete"
     )
     reasoning: str = Field(
@@ -84,10 +83,10 @@ async def calculate_next_stage(
     task_metadata: TaskMetadata,
     current_operation: str,
     conversation_service: ConversationHistoryService,
-    ctx: Optional[Any] = None,  # MCP Context for llm_provider
-    validation_result: Optional[Any] = None,
-    completion_result: Optional[Any] = None,
-    accumulated_changes: Optional[Dict] = None,
+    ctx: Any | None = None,  # MCP Context for llm_provider
+    validation_result: Any | None = None,
+    completion_result: Any | None = None,
+    accumulated_changes: dict | None = None,
 ) -> WorkflowGuidance:
     """
     SHARED METHOD used by all tools to calculate next_tool and instructions.
@@ -110,14 +109,14 @@ async def calculate_next_stage(
         Exception: If LLM fails to generate valid navigation
     """
     logger.info(f"🧠 Calculating next stage for task {task_metadata.task_id}")
-    
+
     try:
         # Load conversation history using task_id as primary key
         # Note: For now we'll use task_id as session_id until we update the DB schema
         conversation_history = await conversation_service.get_conversation_history(
             session_id=task_metadata.task_id
         )
-        
+
         # Format conversation history for LLM context
         conversation_context = _format_conversation_for_llm(conversation_history)
 
@@ -174,9 +173,10 @@ async def calculate_next_stage(
         logger.info(f"📤 Sending navigation request to LLM for task {task_metadata.task_id}")
 
         # Use the same messaging pattern as other tools
-        from mcp_as_a_judge.prompt_loader import create_separate_messages
         from mcp.types import SamplingMessage
 
+        from mcp_as_a_judge.prompt_loader import create_separate_messages
+
         # Create system and user variables for the workflow guidance
         system_vars = WorkflowGuidanceSystemVars(
             response_schema=json.dumps(WorkflowGuidance.model_json_schema())
@@ -227,7 +227,7 @@ async def calculate_next_stage(
         except (ValueError, json.JSONDecodeError) as e:
             logger.error(f"❌ Failed to parse LLM response: {e}")
             logger.error(f"❌ Raw response: {response[:500]}...")
-            raise ValueError(f"Failed to parse workflow guidance response: {e}")
+            raise ValueError(f"Failed to parse workflow guidance response: {e}") from e
 
         # Validate required fields
         required_fields = ["next_tool", "reasoning", "preparation_needed", "guidance"]
@@ -278,31 +278,31 @@ async def calculate_next_stage(
             next_tool=None,
             reasoning="Error occurred during workflow calculation",
             preparation_needed=["Review the error and task state"],
-            guidance=f"Error calculating next stage: {str(e)}. Please review task manually."
+            guidance=f"Error calculating next stage: {e!s}. Please review task manually."
         )
 
 
 def _format_conversation_for_llm(conversation_history) -> str:
     """
     Format conversation history for LLM context.
-    
+
     Args:
         conversation_history: List of conversation records
-        
+
     Returns:
         Formatted string for LLM prompt
     """
     if not conversation_history:
         return "No previous conversation history."
-    
+
     formatted_lines = []
     for record in conversation_history[-10:]:  # Last 10 records
         formatted_lines.append(
             f"[{record.timestamp}] {record.source}:\n"
             f"Input: {record.input}\n"
             f"Output: {record.output}\n"
         )
-    
+
     return "\n".join(formatted_lines)