"""
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