Codebuddy MCP Server

""" MCP tools implementation for the Codebuddy server. Follows Clean Code and SOLID principles: - Single Responsibility: Each tool has one clear purpose - Open/Closed: Easy to add new tools without modifying existing ones - Interface Segregation: Small, focused tool interfaces - Dependency Injection: Storage dependency injected, not hardcoded Enhanced with cognitive scaffolding based on research: - Miller's 7±2 rule for working memory limits - Chain-of-thought reasoning patterns - Hierarchical task network (HTN) principles - Metacognitive self-reflection loops - Software engineering best practices """ import logging from datetime import datetime, timedelta from typing import List, Dict, Any from uuid import uuid4 from fastmcp import FastMCP from models import ( Task, TaskStatus, TaskCreateRequest, TaskUpdateRequest, LessonsSummary, CognitiveGuidance, ReflectionPrompts, TaskPlanResponse ) from storage import TaskStorage from error_handling import error_handler, ErrorType, ErrorSeverity # ===== COGNITIVE SCAFFOLDING FRAMEWORKS ===== # Research-based templates for better agentic thinking COGNITIVE_PATTERNS = { "discovery_questions": [ "🎯 What exactly am I trying to achieve? (Clear success criteria)", "🔍 What constraints do I need to work within? (Time, resources, dependencies)", "🧠 What assumptions am I making that I should validate?", "⚠️ What could go wrong and how can I mitigate it?", "📏 What's the smallest version that would still be valuable? (MVP thinking)" ], "validation_criteria_templates": [ "✅ **Success looks like**: Specific, measurable outcomes", "⚠️ **Warning signs**: Red flags that indicate problems", "🔄 **When to pivot**: Conditions that suggest changing approach", "📊 **How to measure progress**: Concrete checkpoints and metrics" ], "reflection_framework": [ "🤔 **What worked well?** (Patterns to repeat)", "🚧 **What was harder than expected?** (Learning opportunities)", "💡 **What would I do differently?** (Process improvements)", "📈 **What patterns can I reuse?** (Knowledge transfer)" ], "software_engineering_reminders": [ "🏗️ **Architecture First**: Design before coding - saves rework later", "🧪 **Test Early**: Write tests as you go, not at the end", "📝 **Document Decisions**: Record why, not just what you built", "🔄 **Iterate Quickly**: Small working versions beat big perfect plans", "👥 **Code for Humans**: Your future self will thank you for clarity", "🛡️ **Error Handling**: Plan for failures from the start", "📦 **Modular Design**: Small, focused components are easier to debug" ], "tool_usage_nudges": [ "💡 **Remember**: You have access to other tools that might help with this task", "🔍 **Search Pattern**: Use search tools to find similar past solutions", "📊 **List & Review**: Check your task list to see what's already in progress", "🎯 **Update Frequently**: Keep task status current for better coordination", "📚 **Learn from History**: Review lessons from completed tasks" ] } # Hierarchical step templates following 7±2 rule (max 6 core phases) COGNITIVE_STEP_TEMPLATES = { "generic_project": [ "### 🎯 **Clarify & Scope** | *Why: Prevents scope creep* | *Check: Clear success criteria written*", "### 🔍 **Analyze & Design** | *Why: Reduces implementation complexity* | *Check: Architecture decisions documented*", "### 🛠️ **Setup & Foundation** | *Why: Solid base prevents rework* | *Check: Environment ready and tested*", "### 📋 **Implement Core Features** | *Why: Deliver value incrementally* | *Check: Each feature fully tested*", "### ✅ **Validate & Test** | *Why: Catch issues early* | *Check: All acceptance criteria met*", "### 🔄 **Review & Iterate** | *Why: Continuous improvement* | *Check: Lessons captured for future use*" ], "debugging_workflow": [ "### 🔍 **Reproduce the Issue** | *Why: Understand the problem fully* | *Check: Consistent reproduction steps*", "### 🧪 **Isolate Variables** | *Why: Find root cause faster* | *Check: Minimum failing example created*", "### 💡 **Hypothesis Formation** | *Why: Structured investigation* | *Check: Testable theories documented*", "### 🛠️ **Test & Fix** | *Why: Verify solutions work* | *Check: Fix addresses root cause*", "### ✅ **Validate Resolution** | *Why: Ensure no regressions* | *Check: All test cases pass*" ], "research_learning": [ "### 📚 **Define Learning Goals** | *Why: Focused effort* | *Check: Clear questions to answer*", "### 🔍 **Gather Information** | *Why: Broad understanding* | *Check: Multiple sources consulted*", "### 🧠 **Synthesize Knowledge** | *Why: Connect concepts* | *Check: Can explain to others*", "### 🛠️ **Apply & Experiment** | *Why: Learning by doing* | *Check: Working examples created*", "### 📝 **Document Insights** | *Why: Knowledge retention* | *Check: Key learnings captured*" ] } logger = logging.getLogger(__name__) class TaskService: """ Service layer for task operations. Follows SRP: Handles business logic for task management. Separates concerns between MCP tools and storage operations. """ def __init__(self, storage: TaskStorage): """Inject storage dependency following Dependency Inversion Principle.""" self.storage = storage def generate_task_steps(self, problem: str) -> List[str]: """ Generate cognitively scaffolded steps for a problem. Uses research-based patterns and respects 7±2 rule for working memory. Provides structured thinking support for agentic users. """ problem_lower = problem.lower() # Select appropriate cognitive template based on problem type if any(word in problem_lower for word in ['debug', 'fix', 'bug', 'error', 'issue', 'problem']): return COGNITIVE_STEP_TEMPLATES["debugging_workflow"] elif any(word in problem_lower for word in ['learn', 'research', 'study', 'understand', 'explore']): return COGNITIVE_STEP_TEMPLATES["research_learning"] else: return COGNITIVE_STEP_TEMPLATES["generic_project"] def assess_complexity(self, problem: str, steps: List[str]) -> str: """ Assess cognitive complexity of a task for load management. Simple heuristic based on problem characteristics. """ complexity_indicators = 0 problem_lower = problem.lower() # Complexity factors if len(problem.split()) > 20: # Long descriptions often indicate complexity complexity_indicators += 1 if any(word in problem_lower for word in ['integrate', 'architecture', 'system', 'complex', 'multiple']): complexity_indicators += 2 if any(word in problem_lower for word in ['new', 'unfamiliar', 'learning', 'research']): complexity_indicators += 1 if len(steps) > 6: # More than 6 steps may overwhelm working memory complexity_indicators += 1 if complexity_indicators >= 3: return "complex" elif complexity_indicators >= 1: return "moderate" else: return "simple" def generate_cognitive_guidance(self, problem: str, complexity: str) -> CognitiveGuidance: """Generate metacognitive scaffolding based on task characteristics.""" complexity_guidance = { "simple": "🧠 **Cognitive Load**: Light - should feel straightforward. If it feels overwhelming, break down further.", "moderate": "🧠 **Cognitive Load**: Moderate - manageable in focused sessions. Take breaks between phases.", "complex": "🧠 **Cognitive Load**: High - consider splitting into multiple tasks. Use the 7±2 rule for each phase." } return CognitiveGuidance( complexity_check=complexity_guidance.get(complexity, complexity_guidance["moderate"]), progress_monitor="📊 **Self-Check**: After each step, ask: 'Am I closer to my goal?' and 'What did I learn?'", adaptation_prompt="🔄 **If Stuck**: Step back, reassess scope, or break current step into smaller pieces. Consider using other available tools.", success_pattern="✅ **Success Signals**: Each step feels achievable, you're building momentum, and can explain progress clearly.", tool_suggestions=[ "💡 Use `search_tasks` to find similar past solutions and patterns", "📋 Check `list_tasks` to see what else is in progress and avoid conflicts", "🎯 Use `update_task` frequently to track progress and insights", "📚 Try `summarize_lessons` to learn from completed tasks" ] ) def generate_reflection_prompts(self, problem: str) -> ReflectionPrompts: """Generate metacognitive reflection questions.""" return ReflectionPrompts( assumption_check="🤔 What assumptions am I making about this problem that I should validate first?", simplification_prompt="📏 What's the smallest version of this that would still deliver value? (MVP thinking)", explanation_test="👥 If I had to explain this approach to a colleague, what would I say? (Clarity check)" ) def analyze_lessons_from_tasks(self, tasks: List[Task]) -> LessonsSummary: """ Analyze patterns in completed and blocked tasks. Extracts insights for improving future task execution. """ completed_tasks = [t for t in tasks if t.status == TaskStatus.COMPLETED] blocked_tasks = [t for t in tasks if t.status == TaskStatus.BLOCKED] # Analyze success patterns success_patterns = [] if completed_tasks: # Simple keyword analysis for successful tasks common_success_words = self._extract_common_keywords( [f"{t.problem} {t.notes}" for t in completed_tasks] ) for word in common_success_words[:3]: # Top 3 patterns success_patterns.append(f"Tasks mentioning '{word}' tend to succeed") # Analyze common blockers common_blockers = [] if blocked_tasks: blocker_keywords = self._extract_common_keywords( [t.notes for t in blocked_tasks if t.notes] ) for word in blocker_keywords[:3]: # Top 3 blockers common_blockers.append(f"'{word}' is a common blocking issue") # Generate recommendations recommendations = [] if len(completed_tasks) > 0: avg_steps = sum(len(t.steps) for t in completed_tasks) / len(completed_tasks) if avg_steps < 3: recommendations.append("Consider breaking problems into more detailed steps") elif avg_steps > 8: recommendations.append("Try to simplify tasks with fewer, broader steps") if len(blocked_tasks) / max(len(tasks), 1) > 0.3: recommendations.append("High blocking rate - consider smaller, incremental tasks") if not recommendations: recommendations.append("Continue current approach - good task completion patterns") return LessonsSummary( success_patterns=success_patterns, common_blockers=common_blockers, recommendations=recommendations, tasks_analyzed=len(tasks) ) def _extract_common_keywords(self, texts: List[str]) -> List[str]: """ Simple keyword extraction from text list. Returns most frequent meaningful words. """ if not texts: return [] # Simple word frequency analysis word_counts = {} stop_words = {'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of', 'with', 'by', 'is', 'are', 'was', 'were', 'be', 'been', 'have', 'has', 'had', 'will', 'would', 'could', 'should'} for text in texts: words = text.lower().split() for word in words: word = word.strip('.,!?;:"()[]{}') if len(word) > 2 and word not in stop_words: word_counts[word] = word_counts.get(word, 0) + 1 # Return top words by frequency return sorted(word_counts.keys(), key=lambda w: word_counts[w], reverse=True) # === HELPER FUNCTIONS FOR COGNITIVE SCAFFOLDING === def _get_status_transition_guidance(old_status: TaskStatus, new_status: TaskStatus) -> str: """Provide guidance for status transitions.""" transitions = { (TaskStatus.PLANNED, TaskStatus.IN_PROGRESS): "🚀 **Starting work** - Focus on the first step, ignore the rest for now", (TaskStatus.IN_PROGRESS, TaskStatus.COMPLETED): "🎉 **Task completed** - Great work! Capture what you learned", (TaskStatus.IN_PROGRESS, TaskStatus.BLOCKED): "🚧 **Encountered blocker** - Step back, reassess, consider alternative approaches", (TaskStatus.BLOCKED, TaskStatus.IN_PROGRESS): "✅ **Blocker resolved** - Resume with fresh perspective and new insights", (TaskStatus.PLANNED, TaskStatus.BLOCKED): "⚠️ **Blocked before starting** - May need more clarity or different approach", (TaskStatus.COMPLETED, TaskStatus.IN_PROGRESS): "🔄 **Reopening task** - New requirements or issues discovered" } key = (old_status, new_status) return transitions.get(key, f"📝 Status changed from {old_status.value} to {new_status.value}") def _generate_progress_feedback(old_status: TaskStatus, new_status: TaskStatus) -> str: """Generate encouraging progress feedback.""" if new_status == TaskStatus.COMPLETED: return "🎯 **Excellent!** Task completion builds momentum and confidence" elif old_status == TaskStatus.PLANNED and new_status == TaskStatus.IN_PROGRESS: return "⚡ **Momentum started!** The hardest part is often just beginning" elif old_status == TaskStatus.BLOCKED and new_status == TaskStatus.IN_PROGRESS: return "💪 **Breakthrough achieved!** Overcoming blocks builds problem-solving skills" else: return "📈 **Progress tracked** - Every update helps build better working habits" def _get_next_action_suggestions(status: TaskStatus, task: Task) -> List[str]: """Suggest next actions based on current status.""" if status == TaskStatus.PLANNED: return [ "📋 Review the task steps and success criteria", "🎯 Mark as 'in_progress' when you start the first step", "🔍 Search for similar past tasks for insights", "📝 Consider breaking down complex steps further" ] elif status == TaskStatus.IN_PROGRESS: return [ "🎯 Focus on completing the current step fully", "📊 Update notes with progress and insights", "⚠️ Mark as 'blocked' if you encounter obstacles", "✅ Mark as 'completed' when all steps are done" ] elif status == TaskStatus.COMPLETED: return [ "📚 Use `summarize_lessons` to extract insights from completed tasks", "🔍 Look for patterns in your successful approaches", "🎯 Plan any follow-up tasks that emerged from this work", "📝 Document key learnings while they're fresh" ] elif status == TaskStatus.BLOCKED: return [ "🤔 Break down the blocker into smaller, specific issues", "🔍 Search for similar problems you've solved before", "💡 Consider alternative approaches or simpler versions", "👥 Reach out for help or fresh perspective if needed" ] else: return ["📋 Review task status and plan next steps"] def _get_reflection_for_status(status: TaskStatus) -> str: """Get appropriate reflection question for the status.""" reflections = { TaskStatus.PLANNED: "🤔 **Before starting**: Am I clear on what success looks like for this task?", TaskStatus.IN_PROGRESS: "📊 **Progress check**: What's working well, and what needs adjustment?", TaskStatus.COMPLETED: "🎯 **Looking back**: What did I learn that I can apply to future tasks?", TaskStatus.BLOCKED: "🚧 **Problem solving**: What's the real blocker here, and what are my options?" } return reflections.get(status, "🤔 What insights can I capture about this task?") def _get_se_reminders_for_status(status: TaskStatus) -> List[str]: """Get software engineering reminders based on task status.""" if status == TaskStatus.IN_PROGRESS: return [ "💾 **Commit frequently** - Save your progress with descriptive messages", "🧪 **Test as you go** - Don't leave testing until the end", "📝 **Document decisions** - Record why you chose specific approaches" ] elif status == TaskStatus.COMPLETED: return [ "🔍 **Code review** - Review your own code before marking truly complete", "📚 **Knowledge sharing** - Document lessons for your future self", "🛡️ **Security check** - Ensure no sensitive data or vulnerabilities introduced" ] elif status == TaskStatus.BLOCKED: return [ "🔬 **Scientific debugging** - Form hypotheses and test systematically", "📋 **Reproduce the issue** - Create minimal failing examples", "🤝 **Seek help effectively** - Prepare clear problem descriptions" ] else: return [ "🏗️ **Plan before coding** - Design interfaces and architecture first", "📊 **Measure twice, cut once** - Validate requirements before implementation" ] def setup_mcp_tools(mcp: FastMCP, task_service: TaskService) -> None: """ Register all MCP tools with the FastMCP server. Follows Open/Closed Principle: Easy to add new tools here. Each tool is a small, focused function following SRP. """ @mcp.tool() def plan_task(problem: str) -> dict: """ 🧠 **Cognitive Project Planning Tool** **Purpose**: Break down complex problems using proven cognitive scaffolding principles **How it helps your thinking**: - Applies the 7±2 rule to prevent cognitive overload - Uses hierarchical decomposition to manage complexity - Includes self-reflection prompts for adaptive thinking - Provides clear validation criteria for each step - Incorporates software engineering best practices **Best used when**: You have a complex goal and need structured thinking support **Cognitive pattern**: Discovery → Planning → Execution → Validation → Reflection Args: problem: The problem description to plan steps for Returns: Dictionary with enhanced cognitive scaffolding and structured guidance """ try: # Validate input if not problem or not problem.strip(): raise ValueError("Problem description cannot be empty") problem_clean = problem.strip() # === COGNITIVE SCAFFOLDING GENERATION === # Apply research-based thinking frameworks # 1. Generate cognitively structured steps (respects 7±2 rule) steps = task_service.generate_task_steps(problem_clean) # 2. Assess complexity for cognitive load management complexity = task_service.assess_complexity(problem_clean, steps) # 3. Generate metacognitive guidance cognitive_guidance = task_service.generate_cognitive_guidance(problem_clean, complexity) # 4. Create reflection prompts for self-monitoring reflection_prompts = task_service.generate_reflection_prompts(problem_clean) # 5. Select relevant software engineering practices se_practices = [] problem_lower = problem_clean.lower() if any(word in problem_lower for word in ['code', 'implement', 'build', 'develop']): se_practices.extend([ "🏗️ **Start with Architecture**: Design interfaces before implementation", "🧪 **Test-Driven**: Write tests first to clarify requirements", "📝 **Document Decisions**: Record architectural choices and trade-offs" ]) if any(word in problem_lower for word in ['debug', 'fix', 'error']): se_practices.extend([ "🔬 **Scientific Method**: Form hypotheses, test systematically", "📋 **Reproduce First**: Consistent reproduction before fixing", "🛡️ **Regression Prevention**: Ensure fix doesn't break other things" ]) if any(word in problem_lower for word in ['refactor', 'improve', 'optimize']): se_practices.extend([ "📊 **Measure First**: Baseline performance before optimizing", "🎯 **Target Bottlenecks**: Profile to find actual slow points", "✅ **Maintain Behavior**: Keep existing functionality intact" ]) # Default practices for all projects se_practices.extend([ "🔄 **Version Control**: Commit frequently with clear messages", "👥 **Code Reviews**: Fresh eyes catch issues you miss", "📚 **Learn Continuously**: Each problem teaches something new" ]) # Create enhanced task with cognitive fields task = Task( problem=problem_clean, steps=steps, status=TaskStatus.PLANNED, complexity_level=complexity, success_criteria=[ "All steps completed with validation checks passed", "Solution meets original problem requirements", "Key learnings documented for future reference" ], dependencies=[], # Can be filled in by user risk_factors=[ f"Complexity level: {complexity} - monitor cognitive load", "Scope creep - stick to defined success criteria", "Technical debt - balance speed vs. maintainability" ] ) saved_task = task_service.storage.save_task(task) logger.info(f"Created cognitively scaffolded task {saved_task.id} with complexity: {complexity}") # === ENHANCED RESPONSE WITH COGNITIVE SCAFFOLDING === return { "task_id": str(saved_task.id), "cognitive_assessment": { "complexity_level": complexity, "working_memory_load": f"{len(steps)} phases (within 7±2 limit)" if len(steps) <= 7 else f"{len(steps)} phases (consider breaking down further)", "planning_rationale": "Using hierarchical task network (HTN) principles for structured decomposition", "estimated_cognitive_effort": "Low" if complexity == "simple" else "Moderate" if complexity == "moderate" else "High" }, "hierarchical_steps": [ { "step_number": i + 1, "description": step, "cognitive_support": "Focus on this step only - don't worry about later phases yet", "validation_check": "Step complete when you can clearly explain what you accomplished" } for i, step in enumerate(steps) ], "cognitive_guidance": { "complexity_check": cognitive_guidance.complexity_check, "progress_monitor": cognitive_guidance.progress_monitor, "adaptation_prompt": cognitive_guidance.adaptation_prompt, "success_pattern": cognitive_guidance.success_pattern, "tool_suggestions": cognitive_guidance.tool_suggestions }, "reflection_prompts": { "assumption_check": reflection_prompts.assumption_check, "simplification_prompt": reflection_prompts.simplification_prompt, "explanation_test": reflection_prompts.explanation_test }, "software_engineering_guidance": se_practices, "discovery_questions": COGNITIVE_PATTERNS["discovery_questions"], "metacognitive_notes": [ f"📊 This task breakdown follows cognitive load theory - {len(steps)} steps respects working memory limits", "🔄 Each step builds on the previous one - complete sequentially for best results", "🧠 If any step feels overwhelming, break it down further using the same principles", "💡 Remember: You have other tools available - use them to support this task" ], "next_recommended_actions": [ "🎯 Review the discovery questions to clarify requirements", "📋 Use `list_tasks` to see how this fits with your other work", "🔍 Try `search_tasks` to find similar past solutions", "▶️ When ready, use `update_task` to mark as 'in_progress'" ] } except ValueError as e: error_handler.handle_error(e, ErrorType.VALIDATION, ErrorSeverity.LOW) raise except Exception as e: error_handler.handle_error(e, ErrorType.INTERNAL, ErrorSeverity.MEDIUM) raise @mcp.tool() def update_task(task_id: str, status: str, notes: str = "") -> dict: """ 🎯 **Task Progress Tracking Tool** **Purpose**: Update task status while maintaining cognitive awareness and learning capture **Cognitive Benefits**: - Tracks progress for better self-monitoring - Captures insights for pattern recognition - Provides transition guidance for status changes - Suggests next actions based on current state **Best Practice**: Update frequently to maintain momentum and capture learnings Args: task_id: The ID of the task to update status: New status (planned, in_progress, completed, blocked) notes: Optional notes about the update (encouraged for learning capture) Returns: Dictionary with updated task information and cognitive guidance """ try: # Validate inputs if not task_id or not task_id.strip(): raise ValueError("Task ID cannot be empty") # Validate status try: new_status = TaskStatus(status.lower()) except ValueError: valid_statuses = [s.value for s in TaskStatus] raise ValueError(f"Invalid status '{status}'. Must be one of: {valid_statuses}") # Get existing task existing_task = task_service.storage.get_task_by_id(task_id.strip()) if not existing_task: raise ValueError(f"Task with ID '{task_id}' not found") # Update task with enhanced notes if learning insights provided enhanced_notes = notes.strip() if enhanced_notes and new_status == TaskStatus.COMPLETED: if not any(keyword in enhanced_notes.lower() for keyword in ['learned', 'insight', 'discovered', 'found']): enhanced_notes += " | 💡 Consider adding: What did you learn? What would you do differently?" updated_task = existing_task.update_progress(new_status, enhanced_notes) saved_task = task_service.storage.save_task(updated_task) logger.info(f"Updated task {saved_task.id} to status {saved_task.status}") # === COGNITIVE GUIDANCE BASED ON STATUS TRANSITION === transition_guidance = _get_status_transition_guidance(existing_task.status, new_status) next_actions = _get_next_action_suggestions(new_status, saved_task) return { "id": str(saved_task.id), "problem": saved_task.problem, "status": saved_task.status.value, "previous_status": existing_task.status.value, "notes": saved_task.notes, "updated_at": saved_task.updated_at.isoformat(), "cognitive_feedback": { "status_transition": transition_guidance, "progress_acknowledgment": _generate_progress_feedback(existing_task.status, new_status), "momentum_check": "🔄 Regular updates build momentum and help track patterns", "learning_capture": "📝 Document insights while they're fresh - your future self will thank you" }, "next_recommended_actions": next_actions, "reflection_prompt": _get_reflection_for_status(new_status), "tool_suggestions": [ "📋 Use `list_tasks` to see your overall progress across all tasks", "🔍 Try `search_tasks` to find related work or similar past solutions", "📚 Use `summarize_lessons` to extract patterns from your completed tasks", "🎯 Consider `plan_task` for any new work that emerged from this update" ], "software_engineering_reminders": _get_se_reminders_for_status(new_status) } except ValueError as e: error_handler.handle_error(e, ErrorType.VALIDATION, ErrorSeverity.LOW) raise except Exception as e: error_handler.handle_error(e, ErrorType.INTERNAL, ErrorSeverity.MEDIUM) raise @mcp.tool() def list_tasks(limit: int = 10) -> dict: """ 📋 **Task Overview & Coordination Tool** **Purpose**: Get strategic overview of your work to maintain cognitive coordination **Cognitive Benefits**: - Prevents task conflicts and duplicate effort - Shows progress patterns across multiple projects - Helps prioritize and manage cognitive load - Identifies completion opportunities **Best Practice**: Check regularly to maintain big-picture awareness Args: limit: Maximum number of tasks to return (default: 10, max: 100) Returns: Dictionary with tasks list, cognitive insights, and strategic guidance """ try: # Validate limit if limit < 1: limit = 10 elif limit > 100: limit = 100 # Prevent excessive memory usage tasks = task_service.storage.list_tasks(limit) total_count = task_service.storage.get_total_count() # === COGNITIVE ANALYSIS OF TASK PORTFOLIO === status_distribution = {} complexity_distribution = {} recent_activity = 0 tasks_data = [] for task in tasks: # Count status distribution status = task.status.value status_distribution[status] = status_distribution.get(status, 0) + 1 # Count complexity distribution if task.complexity_level: complexity_distribution[task.complexity_level] = complexity_distribution.get(task.complexity_level, 0) + 1 # Check for recent activity (updated in last 7 days) if (datetime.utcnow() - task.updated_at) < timedelta(days=7): recent_activity += 1 tasks_data.append({ "id": str(task.id), "problem": task.problem, "steps": task.steps, "status": task.status.value, "notes": task.notes, "complexity_level": task.complexity_level, "created_at": task.created_at.isoformat(), "updated_at": task.updated_at.isoformat(), "step_progress": f"{len([s for s in task.steps if '✅' in s or 'done' in s.lower()])}/{len(task.steps)}" if task.steps else "0/0" }) # === COGNITIVE PORTFOLIO INSIGHTS === portfolio_insights = [] in_progress_count = status_distribution.get('in_progress', 0) if in_progress_count > 3: portfolio_insights.append("⚠️ **High parallel work**: Consider focusing on fewer tasks to reduce cognitive switching costs") elif in_progress_count == 0 and status_distribution.get('planned', 0) > 0: portfolio_insights.append("🚀 **Ready to start**: Pick your highest priority planned task and begin") if status_distribution.get('blocked', 0) > 0: portfolio_insights.append("🚧 **Blocked tasks detected**: Address blockers before starting new work") completion_rate = status_distribution.get('completed', 0) / max(len(tasks_data), 1) if completion_rate > 0.7: portfolio_insights.append("🎯 **High completion rate**: Great execution pattern - keep it up!") elif completion_rate < 0.3: portfolio_insights.append("💡 **Low completion rate**: Consider breaking tasks smaller or addressing common blockers") if recent_activity < len(tasks_data) * 0.3: portfolio_insights.append("📊 **Stale tasks detected**: Update task statuses to maintain momentum and accurate tracking") return { "tasks": tasks_data, "returned_count": len(tasks_data), "total_count": total_count, "cognitive_portfolio_analysis": { "status_distribution": status_distribution, "complexity_distribution": complexity_distribution, "recent_activity_count": recent_activity, "cognitive_load_assessment": "Light" if in_progress_count <= 2 else "Moderate" if in_progress_count <= 4 else "High", "completion_rate_percentage": round(completion_rate * 100, 1) }, "strategic_insights": portfolio_insights, "recommended_actions": [ "🎯 Focus on completing in-progress tasks before starting new ones", "🚧 Address any blocked tasks - they consume mental energy even when ignored", "📊 Update stale tasks to maintain accurate progress tracking", "🔍 Use `search_tasks` to find related work and avoid duplicates", "📚 Consider `summarize_lessons` to extract patterns from completed tasks" ], "cognitive_guidance": { "focus_recommendation": "Work on 1-2 tasks at a time for optimal cognitive efficiency", "prioritization_tip": "Choose tasks that build momentum - mix quick wins with important work", "completion_strategy": "Small, frequent completions are better than large, delayed ones", "learning_opportunity": "Each completed task teaches something - capture those insights" }, "tool_integration_suggestions": [ "💡 Use `plan_task` for any new work that needs to be added", "🎯 Use `update_task` to keep progress current and capture insights", "🔍 Use `search_tasks` to find connections between tasks", "📚 Use `summarize_lessons` after completing several tasks" ] } except Exception as e: error_handler.handle_error(e, ErrorType.INTERNAL, ErrorSeverity.MEDIUM) raise @mcp.tool() def search_tasks(query: str) -> dict: """ 🔍 **Pattern Recognition & Knowledge Discovery Tool** **Purpose**: Find connections between past and current work for better decision-making **Cognitive Benefits**: - Leverages past solutions to solve current problems - Identifies successful patterns and approaches - Prevents reinventing the wheel - Builds on accumulated knowledge and experience **Best Practice**: Search before starting new work - you may have solved similar problems Args: query: Search query string (keywords related to your current problem) Returns: Dictionary with matching tasks, insights, and pattern analysis """ try: if not query or not query.strip(): raise ValueError("Search query cannot be empty") tasks = task_service.storage.search_tasks(query.strip()) # === COGNITIVE PATTERN ANALYSIS === completed_matches = [t for t in tasks if t.status == TaskStatus.COMPLETED] blocked_matches = [t for t in tasks if t.status == TaskStatus.BLOCKED] in_progress_matches = [t for t in tasks if t.status == TaskStatus.IN_PROGRESS] tasks_data = [] success_patterns = [] warning_patterns = [] for task in tasks: task_data = { "id": str(task.id), "problem": task.problem, "steps": task.steps, "status": task.status.value, "notes": task.notes, "complexity_level": task.complexity_level, "created_at": task.created_at.isoformat(), "updated_at": task.updated_at.isoformat(), "relevance_score": "high" if query.lower() in task.problem.lower() else "medium" } # Extract insights from completed tasks if task.status == TaskStatus.COMPLETED and task.notes: if any(word in task.notes.lower() for word in ['worked', 'success', 'solved', 'effective']): success_patterns.append(f"✅ From '{task.problem[:50]}...': {task.notes[:100]}...") # Extract warnings from blocked tasks if task.status == TaskStatus.BLOCKED and task.notes: warning_patterns.append(f"⚠️ Blocker in '{task.problem[:50]}...': {task.notes[:100]}...") tasks_data.append(task_data) # === STRATEGIC INSIGHTS GENERATION === cognitive_insights = [] if completed_matches: cognitive_insights.append(f"🎯 **{len(completed_matches)} similar completed tasks found** - review their approaches") if blocked_matches: cognitive_insights.append(f"🚧 **{len(blocked_matches)} similar blocked tasks found** - learn from their challenges") if in_progress_matches: cognitive_insights.append(f"🔄 **{len(in_progress_matches)} similar ongoing tasks** - consider coordination or consolidation") if not tasks: cognitive_insights.append("💡 **No similar tasks found** - you're exploring new territory! Document your approach well.") return { "query": query.strip(), "tasks": tasks_data, "results_count": len(tasks_data), "cognitive_insights": cognitive_insights, "pattern_analysis": { "completed_similar": len(completed_matches), "blocked_similar": len(blocked_matches), "in_progress_similar": len(in_progress_matches), "success_rate": round(len(completed_matches) / max(len(tasks), 1) * 100, 1), "novelty_indicator": "High" if len(tasks) == 0 else "Medium" if len(tasks) < 3 else "Low" }, "extracted_success_patterns": success_patterns[:3], # Top 3 "extracted_warning_patterns": warning_patterns[:3], # Top 3 "strategic_recommendations": [ "📚 **Learn from successes**: Study approaches that worked in completed similar tasks", "🚧 **Avoid known pitfalls**: Review blockers from similar past work", "🔄 **Consider coordination**: If similar work is in progress, explore collaboration", "📝 **Document your approach**: Help future searches by capturing your methods and insights" ], "next_actions": [ "📋 Review the most relevant matches for applicable strategies", "🎯 Use `plan_task` to create a new approach incorporating these insights", "📊 Use `update_task` to capture learnings as you apply these patterns", "👥 Consider reaching out if similar work is in progress elsewhere" ], "cognitive_guidance": { "pattern_recognition": "Look for recurring themes in successful approaches", "failure_analysis": "Understand why similar tasks got blocked - different context or systemic issues?", "knowledge_building": "Each search builds your problem-solving knowledge base", "transfer_learning": "Adapt successful patterns to your current context rather than copying exactly" } } except ValueError as e: error_handler.handle_error(e, ErrorType.VALIDATION, ErrorSeverity.LOW) raise except Exception as e: error_handler.handle_error(e, ErrorType.INTERNAL, ErrorSeverity.MEDIUM) raise @mcp.tool() def summarize_lessons() -> dict: """ 📚 **Meta-Learning & Pattern Extraction Tool** **Purpose**: Transform your task history into actionable wisdom for better future performance **Cognitive Benefits**: - Builds metacognitive awareness of your working patterns - Identifies successful strategies to repeat - Reveals systemic issues before they become major problems - Accelerates learning through pattern recognition **Best Practice**: Review lessons periodically to compound your problem-solving effectiveness Returns: Dictionary with deep insights, patterns, and strategic recommendations for improvement """ try: # Analyze recent tasks (last 50 for meaningful patterns) recent_tasks = task_service.storage.list_tasks(limit=50) if not recent_tasks: return { "message": "No tasks found to analyze - start by creating and completing some tasks!", "next_steps": [ "🎯 Use `plan_task` to create your first structured task", "📊 Use `update_task` to track progress and capture insights", "🔄 Return here after completing a few tasks for pattern analysis" ], "tasks_analyzed": 0 } # === DEEP COGNITIVE ANALYSIS === completed_tasks = [t for t in recent_tasks if t.status == TaskStatus.COMPLETED] blocked_tasks = [t for t in recent_tasks if t.status == TaskStatus.BLOCKED] in_progress_tasks = [t for t in recent_tasks if t.status == TaskStatus.IN_PROGRESS] # Analyze completion patterns completion_insights = [] success_strategies = [] failure_patterns = [] cognitive_patterns = [] # Success pattern analysis if completed_tasks: avg_completion_time = sum((t.updated_at - t.created_at).days for t in completed_tasks) / len(completed_tasks) completion_insights.append(f"⏱️ **Average completion time**: {avg_completion_time:.1f} days") # Analyze successful complexity management complex_completed = [t for t in completed_tasks if t.complexity_level == 'complex'] if complex_completed: success_strategies.append(f"🧠 **Complex task success**: You've completed {len(complex_completed)} complex tasks - you can handle challenging work!") # Look for successful step patterns successful_step_counts = [len(t.steps) for t in completed_tasks if t.steps] if successful_step_counts: avg_successful_steps = sum(successful_step_counts) / len(successful_step_counts) success_strategies.append(f"📋 **Optimal breakdown**: Tasks with ~{avg_successful_steps:.0f} steps tend to get completed") # Failure pattern analysis if blocked_tasks: # Analyze what causes blocks blocked_with_notes = [t for t in blocked_tasks if t.notes] if blocked_with_notes: common_block_words = [] for task in blocked_with_notes: words = task.notes.lower().split() common_block_words.extend([w for w in words if len(w) > 4]) if common_block_words: # Simple frequency analysis word_freq = {} for word in common_block_words: word_freq[word] = word_freq.get(word, 0) + 1 top_block_causes = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:3] for word, freq in top_block_causes: if freq > 1: failure_patterns.append(f"🚧 **Recurring blocker**: '{word}' appears in {freq} blocked tasks") # Analyze complexity vs blocking complex_blocked = [t for t in blocked_tasks if t.complexity_level == 'complex'] if len(complex_blocked) / max(len(blocked_tasks), 1) > 0.7: failure_patterns.append("⚠️ **Complexity correlation**: Most blocks occur in complex tasks - consider smaller breakdowns") # Progress pattern analysis stalled_tasks = [t for t in in_progress_tasks if (datetime.utcnow() - t.updated_at).days > 7] if stalled_tasks: cognitive_patterns.append(f"📊 **Attention pattern**: {len(stalled_tasks)} tasks haven't been updated in >7 days - may need attention") # Cognitive load analysis current_cognitive_load = len(in_progress_tasks) if current_cognitive_load > 5: cognitive_patterns.append("🧠 **High cognitive load**: Multiple active tasks may be reducing effectiveness") elif current_cognitive_load == 0 and len(recent_tasks) > 5: cognitive_patterns.append("🎯 **Ready for action**: No active tasks - good time to start something new") # Generate strategic recommendations strategic_recommendations = [] if completed_tasks and blocked_tasks: completion_rate = len(completed_tasks) / (len(completed_tasks) + len(blocked_tasks)) if completion_rate > 0.8: strategic_recommendations.append("🎯 **Excellent execution**: High completion rate - maintain current approach") elif completion_rate < 0.5: strategic_recommendations.append("💡 **Focus on completion**: Consider smaller tasks or better blocker management") if failure_patterns: strategic_recommendations.append("🔍 **Pattern detected**: Address recurring blockers systematically - they're costing you momentum") if not success_strategies: strategic_recommendations.append("📝 **Capture insights**: Add more detailed notes to completed tasks to identify success patterns") strategic_recommendations.extend([ "🔄 **Continuous improvement**: Apply these insights to your next task planning", "📊 **Track progress**: Regular updates help identify patterns and maintain momentum", "🧠 **Cognitive awareness**: Balance challenging and routine work to optimize performance" ]) return { "analysis_summary": { "tasks_analyzed": len(recent_tasks), "completion_rate_percentage": round(len(completed_tasks) / max(len(recent_tasks), 1) * 100, 1), "current_active_tasks": len(in_progress_tasks), "cognitive_load_assessment": "Light" if current_cognitive_load <= 2 else "Moderate" if current_cognitive_load <= 4 else "High" }, "success_patterns": success_strategies or ["🎯 Complete more tasks with detailed notes to identify success patterns"], "failure_patterns": failure_patterns or ["✅ No significant blocking patterns detected - good execution!"], "cognitive_insights": cognitive_patterns or ["📊 Healthy task management patterns - keep it up!"], "strategic_recommendations": strategic_recommendations, "performance_metrics": { "completed_tasks": len(completed_tasks), "blocked_tasks": len(blocked_tasks), "in_progress_tasks": len(in_progress_tasks), "average_completion_days": round(sum((t.updated_at - t.created_at).days for t in completed_tasks) / max(len(completed_tasks), 1), 1) if completed_tasks else None, "stalled_task_count": len(stalled_tasks) if 'stalled_tasks' in locals() else 0 }, "next_optimization_actions": [ "🎯 Apply these insights when using `plan_task` for new work", "📊 Use `update_task` more frequently to capture real-time insights", "🔍 Use `search_tasks` to find and replicate successful approaches", "📋 Check `list_tasks` regularly to maintain cognitive coordination" ], "meta_learning_guidance": { "pattern_recognition": "Look for themes across multiple tasks, not just individual successes", "feedback_loops": "The more you capture insights, the better these analyses become", "iterative_improvement": "Small consistent improvements compound over time", "cognitive_scaffolding": "Use these insights to structure your future problem-solving approaches" } } except Exception as e: error_handler.handle_error(e, ErrorType.INTERNAL, ErrorSeverity.MEDIUM) raise