mypy

Author: mcp-release-bot

src/mcp_as_a_judge/server.py

@@ -10,6 +10,7 @@
 import json
 import re
 import time
+from typing import Any
 
 from mcp.server.fastmcp import Context, FastMCP
 from pydantic import ValidationError
@@ -2787,7 +2788,7 @@ class UserFeedbackSchema(BaseModel):
         feedback_context = {
             "repository_analysis": repository_analysis,
             "task_description": task_metadata.description,
-            "task_size": task_metadata.size.value if task_metadata.size else "unknown",
+            "task_size": task_metadata.task_size.value if task_metadata.task_size else "unknown",
             "workflow_state": task_metadata.state.value,
             "elicitation_success": elicitation_result.success,
             "questions_asked": specific_questions[:5],  # Limit to avoid token bloat
@@ -3150,7 +3151,7 @@ async def get_user_approve_requirement(
     plan: str,
     design: str,
     research: str,
-    technical_decisions: list[dict],
+    technical_decisions: list[dict | Any],
     implementation_scope: dict,
     language_specific_practices: list[str],
     task_id: str,
@@ -3357,7 +3358,7 @@ class PlanApprovalSchema(BaseModel):
         # Save plan approval interaction to database for better LLM context
         approval_context = {
             "task_description": task_metadata.description,
-            "task_size": task_metadata.size.value if task_metadata.size else "unknown",
+            "task_size": task_metadata.task_size.value if task_metadata.task_size else "unknown",
             "workflow_state": task_metadata.state.value,
             "plan_content": formatted_plan[:1000],  # Limit to avoid token bloat
             "user_approved": approved,

src/mcp_as_a_judge/utils/repository_analyzer.py

@@ -121,8 +121,8 @@ def analyze(self) -> dict:
 
     def _detect_languages(self) -> dict:
         """Detect programming languages used in the repository."""
-        language_scores = {}
-        file_counts = Counter()
+        language_scores: dict[str, float] = {}
+        file_counts: Counter[str] = Counter()
 
         # Walk through repository
         for root, dirs, files in os.walk(self.repo_path):
@@ -177,7 +177,7 @@ def _detect_languages(self) -> dict:
 
     def _detect_frameworks(self) -> dict:
         """Detect frameworks and libraries used."""
-        framework_scores = {}
+        framework_scores: dict[str, float] = {}
 
         for _root, dirs, files in os.walk(self.repo_path):
             dirs[:] = [
@@ -212,7 +212,7 @@ def _detect_frameworks(self) -> dict:
 
     def _analyze_structure(self) -> dict:
         """Analyze project structure and organization patterns."""
-        structure = {"directories": [], "key_files": [], "patterns": []}
+        structure: dict[str, list[str]] = {"directories": [], "key_files": [], "patterns": []}
 
         # Get top-level directories and files
         if self.repo_path.exists():
@@ -383,36 +383,37 @@ def _generate_recommendations(self) -> dict:
         languages = self._detect_languages()
         frameworks = self._detect_frameworks()
 
+        reasoning_list: list[str] = []
         recommendations = {
             "technology_stack": "unclear",
             "suggested_approach": "ask_user",
-            "reasoning": [],
+            "reasoning": reasoning_list,
         }
 
         if languages["confidence"] == "high" and languages["primary"]:
             recommendations["technology_stack"] = "detected"
             recommendations["suggested_approach"] = "confirm_or_choose_different"
-            recommendations["reasoning"].append(
+            reasoning_list.append(
                 f"Clear {languages['primary']} project detected - recommend continuing with {languages['primary']} for consistency"
             )
-            recommendations["reasoning"].append(
+            reasoning_list.append(
                 "However, user may choose different language for valid reasons (microservices, tooling, team expertise, etc.)"
             )
 
             if frameworks["likely"]:
-                recommendations["reasoning"].append(
+                reasoning_list.append(
                     f"Likely frameworks: {', '.join(frameworks['likely'])}"
                 )
         elif languages["confidence"] == "medium":
             recommendations["technology_stack"] = "mixed_or_unclear"
             recommendations["suggested_approach"] = "clarify_with_user"
-            recommendations["reasoning"].append(
+            reasoning_list.append(
                 "Multiple languages detected or unclear primary language"
             )
         else:
             recommendations["technology_stack"] = "empty_or_new"
             recommendations["suggested_approach"] = "ask_user_preferences"
-            recommendations["reasoning"].append("No clear technology stack detected")
+            reasoning_list.append("No clear technology stack detected")
 
         return recommendations
 

src/mcp_as_a_judge/workflow/workflow_guidance.py

@@ -565,33 +565,19 @@ async def calculate_next_stage(
         )
 
         # Fallback: if next_tool missing/None and not completed, route to get_current_coding_task
-        if (
-            workflow_guidance.next_tool is None
-            and task_metadata.state != TaskState.COMPLETED
-        ):
+        if workflow_guidance.next_tool is None and task_metadata.state != TaskState.COMPLETED:
             if "get_current_coding_task" in available_name_set:
                 workflow_guidance.next_tool = "get_current_coding_task"
             else:
                 # As a last resort, pick appropriate tool based on state
-                if (
-                    task_metadata.state == TaskState.CREATED
-                    or task_metadata.state == TaskState.REQUIREMENTS_FEEDBACK
-                ):
-                    workflow_guidance.next_tool = "get_user_feedback"
-                elif task_metadata.state == TaskState.USER_APPROVE_REQUIREMENTS:
-                    workflow_guidance.next_tool = (
-                        None  # Let AI assistant create plan first
-                    )
-                elif task_metadata.state == TaskState.PLANNING:
+                # Note: CREATED, REQUIREMENTS_FEEDBACK, USER_APPROVE_REQUIREMENTS, PLAN_APPROVED,
+                # IMPLEMENTING, REVIEW_READY, TESTING, and COMPLETED are handled by early returns
+                current_state = task_metadata.state
+                if current_state == TaskState.PLANNING:
                     workflow_guidance.next_tool = "judge_coding_plan"
-                elif task_metadata.state in (
-                    TaskState.PLAN_APPROVED,
-                    TaskState.IMPLEMENTING,
-                    TaskState.REVIEW_READY,
-                ):
-                    workflow_guidance.next_tool = "judge_code_change"
-                elif task_metadata.state == TaskState.TESTING:
-                    workflow_guidance.next_tool = "judge_testing_implementation"
+                elif current_state in (TaskState.BLOCKED, TaskState.CANCELLED):
+                    # For blocked/cancelled tasks, no specific tool recommendation
+                    workflow_guidance.next_tool = None
 
         logger.info(
             f"Calculated next stage: next_tool={workflow_guidance.next_tool}, "

tests/conftest.py

@@ -9,6 +9,7 @@
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from pydantic import BaseModel
 
 from mcp_as_a_judge.models import JudgeResponse
 
@@ -140,6 +141,15 @@ def mock_no_sampling_context():
     return mock_context
 
 
+class MockTextContent(BaseModel):
+    type: str = "text"
+    text: str
+
+
+class MockCreateMessageResult(BaseModel):
+    content: MockTextContent
+
+
 class MockServerSession:
     """Mock server session for testing."""
 
@@ -148,25 +158,18 @@ def __init__(self, has_sampling: bool = True):
         self.has_sampling = has_sampling
 
     async def create_message(self, **kwargs):
-        """Mock create_message method."""
+        """Mock create_message method returning Pydantic-like response objects."""
         if not self.has_sampling:
             raise RuntimeError("Context is not available outside of a request")
 
         # Return proper JSON response for workflow guidance
         if "workflow" in str(kwargs).lower() or "guidance" in str(kwargs).lower():
             json_response = '{"next_tool": "judge_coding_plan", "reasoning": "Need to validate the coding plan", "preparation_needed": ["Gather requirements", "Research best practices"], "guidance": "Start by analyzing the requirements and creating a comprehensive plan"}'
-            # Create a mock that mimics the MCP response structure
-            mock_content = MagicMock()
-            mock_content.type = "text"
-            mock_content.text = json_response
-            return MagicMock(content=mock_content)
+            return MockCreateMessageResult(content=MockTextContent(text=json_response))
 
         # Return proper JSON response for judge responses
         json_response = '{"approved": true, "feedback": "Mocked evaluation response"}'
-        mock_content = MagicMock()
-        mock_content.type = "text"
-        mock_content.text = json_response
-        return MagicMock(content=mock_content)
+        return MockCreateMessageResult(content=MockTextContent(text=json_response))
 
 
 class MockContext: