handlers.py•16.7 kB
# Aidderall MCP Server - Hierarchical task management for AI assistants
# Copyright (C) 2024 Briam R. <briamr@gmail.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from typing import Any, Dict, Optional
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import EmbeddedResource, ImageContent, TextContent, Tool
from .models import Task, TaskStatus
from .task_manager import TaskManager
class AidderallHandlers:
def __init__(self, task_manager: TaskManager) -> None:
self.task_manager = task_manager
async def handle_create_new_task(self, title: str, body: str) -> Dict[str, Any]:
try:
# Get previous task info before creating new one
previous_task = self.task_manager.current_task
task = self.task_manager.create_new_task(title, body)
response = {
"task_id": task.id,
"message": f"Created independent task: {task.title}",
"focus_path": self.task_manager.get_focus_path(),
"is_current": True,
}
if previous_task:
response["note"] = (
f"Previous task '{previous_task.title}' is now pending"
)
return response
except ValueError as e:
return {"error": str(e)}
async def handle_extend_current_task(self, title: str, body: str) -> Dict[str, Any]:
try:
task = self.task_manager.extend_current_task(title, body)
response = {
"task_id": task.id,
"message": f"Added subtask: {task.title}",
"focus_path": self.task_manager.get_focus_path(),
"is_current": True,
"hint": "Use switch_focus to work on any task in any order",
}
# Add warning if getting deep
stack_depth = self.task_manager.get_stack_depth()
if stack_depth >= 4:
response["info"] = (
f"Stack is {stack_depth} levels deep. Consider using create_new_task for unrelated work."
)
return response
except ValueError as e:
return {"error": str(e)}
async def handle_get_current_task(self) -> Dict[str, Any]:
current = self.task_manager.current_task
if self.task_manager.is_zen_state:
if not current:
return {"message": "No tasks (zen state)", "zen_state": True}
else:
return {"message": "All tasks completed (zen state)", "zen_state": True}
if not current:
return {"error": "No current task"}
siblings_left = self.task_manager.get_siblings_to_left()
breadcrumb = self.task_manager.get_breadcrumb_trail()
response = {
"task_id": current.id,
"title": current.title,
"body": current.body,
"focus_path": self.task_manager.get_focus_path(),
"stack_depth": self.task_manager.get_stack_depth(),
"siblings_to_left": [{"id": s.id, "title": s.title} for s in siblings_left],
"breadcrumb_trail": [{"id": t.id, "title": t.title} for t in breadcrumb],
}
# Add navigation options
response["navigation_hint"] = (
"Use switch_focus to work on any task, or complete_current_task when done"
)
return response
async def handle_get_big_picture(self, format: str = "text") -> Dict[str, Any]:
result = self.task_manager.get_big_picture(format)
if format == "json":
return result # type: ignore
else:
return {"structure": result}
async def handle_complete_current_task(self) -> Dict[str, Any]:
completed = self.task_manager.complete_current_task()
if not completed:
return {"error": "No current task to complete"}
new_current = self.task_manager.current_task
response = {
"completed_task_id": completed.id,
"completed_task_title": completed.title,
"new_current_task_id": new_current.id if new_current else None,
"new_current_task_title": (
new_current.title if new_current else "No active tasks (zen state)"
),
}
# Add focus path if there's a new current task
if new_current:
response["focus_path"] = self.task_manager.get_focus_path()
response["navigation_hint"] = (
f"Focus moved to '{new_current.title}'. Use switch_focus to work on any other task."
)
return response
async def handle_get_completed_tasks(
self, order: str = "chronological"
) -> Dict[str, Any]:
try:
tasks = self.task_manager.get_completed_tasks(order)
return {
"count": len(tasks),
"tasks": [
{
"id": t.id,
"title": t.title,
"body": t.body,
"completed_at": (
t.completed_at.isoformat() if t.completed_at else None
),
}
for t in tasks
],
}
except ValueError as e:
return {"error": str(e)}
async def handle_update_current_task(self, body: str) -> Dict[str, Any]:
try:
task = self.task_manager.update_current_task(body)
return {
"task_id": task.id,
"title": task.title,
"message": "Task body updated successfully",
}
except ValueError as e:
return {"error": str(e)}
async def handle_get_stack_overview(self) -> Dict[str, Any]:
return self.task_manager.get_stack_overview()
async def handle_peek_context(self, include_body: bool = False) -> Dict[str, Any]:
parent, immediate = self.task_manager.peek_context()
def format_task(task: Optional[Task]) -> Optional[Dict[str, Any]]:
if not task:
return None
result = {
"id": task.id,
"title": task.title,
"status": task.status.value,
"created_at": task.created_at.isoformat(),
}
if include_body:
result["body"] = task.body
if task.completed_at:
result["completed_at"] = task.completed_at.isoformat()
return result
return {
"parent_context": format_task(parent),
"immediate_context": format_task(immediate),
}
async def handle_list_siblings(self, include_body: bool = False) -> Dict[str, Any]:
siblings = self.task_manager.list_siblings()
return {
"count": len(siblings),
"siblings": [
{
"id": s.id,
"title": s.title,
"status": s.status.value,
"created_at": s.created_at.isoformat(),
**({"body": s.body} if include_body else {}),
**(
{"completed_at": s.completed_at.isoformat()}
if s.completed_at
else {}
),
}
for s in siblings
],
}
async def handle_switch_focus(self, task_id: str) -> Dict[str, Any]:
try:
result = self.task_manager.switch_focus(task_id)
return result
except ValueError as e:
return {"error": str(e)}
async def handle_remove_task(self, task_id: str) -> Dict[str, Any]:
try:
result = self.task_manager.remove_task(task_id)
return result
except ValueError as e:
return {"error": str(e)}
def get_tool_definitions(self) -> list[Tool]:
return [
Tool(
name="create_new_task",
description="Create an INDEPENDENT task for unrelated work. Use for: new topics, context switches, or parallel workstreams. Adds a new top-level task to your workspace. Previous task keeps its status. Example: working on Feature A, need to research Topic B → create_new_task for Topic B. For breaking down current work, use extend_current_task.",
inputSchema={
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Brief task description (max 256 chars)",
},
"body": {
"type": "string",
"description": "Full task context, notes, and details",
},
},
"required": ["title", "body"],
},
),
Tool(
name="extend_current_task",
description="Add a subtask to organize and decompose work. Creates hierarchical structure for complex tasks. Subtasks help break down work into manageable pieces. You can work on tasks in any order using switch_focus. Example: Task A → Task B → Task C creates a nested structure, but you can jump between them freely. For unrelated work, use create_new_task.",
inputSchema={
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Brief task description (max 256 chars)",
},
"body": {
"type": "string",
"description": "Full task context, notes, and details",
},
},
"required": ["title", "body"],
},
),
Tool(
name="get_current_task",
description="Get the task with CURRENT status (your active focus). May be manually set via switch_focus, or automatically determined. In zen state (no tasks OR all tasks completed), returns appropriate message.",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="get_big_picture",
description="See ALL tasks in your task stack (including completed ones). Shows full work context as a living document. Marks current task with 'YOU ARE HERE'. Indicates zen state when no tasks exist OR all tasks are completed.",
inputSchema={
"type": "object",
"properties": {
"format": {
"type": "string",
"enum": ["text", "json"],
"description": "Output format",
"default": "text",
}
},
},
),
Tool(
name="complete_current_task",
description="Mark current task as COMPLETED. Task remains visible in structure (living document approach). Focus automatically moves to a nearby incomplete task. Creates permanent record in history. You can use switch_focus to work on any specific task instead. Use remove_task to clean up workspace later.",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="get_completed_tasks",
description="View chronological history of ALL completed tasks. This is a permanent archive separate from the visible structure. Tasks remain here even after being removed from the workspace. Useful for reviewing what you've accomplished over time.",
inputSchema={
"type": "object",
"properties": {
"order": {
"type": "string",
"enum": ["chronological", "logical"],
"description": "Order of completed tasks",
"default": "chronological",
}
},
},
),
Tool(
name="update_current_task",
description="Update notes/content of the task you're currently focused on (current task only)",
inputSchema={
"type": "object",
"properties": {
"body": {
"type": "string",
"description": "New body content for the task",
}
},
"required": ["body"],
},
),
Tool(
name="get_stack_overview",
description="Get structured data of your entire task stack (JSON format with all task details and relationships)",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="peek_context",
description="Look at parent task and previous sibling without changing focus (understand WHY you're doing current task)",
inputSchema={
"type": "object",
"properties": {
"include_body": {
"type": "boolean",
"description": "Include task body content",
"default": False,
}
},
},
),
Tool(
name="list_siblings",
description="See all sibling tasks to the left of current focus. May include both pending and completed tasks. Helpful for understanding your position in the current task sequence.",
inputSchema={
"type": "object",
"properties": {
"include_body": {
"type": "boolean",
"description": "Include task body content",
"default": False,
}
},
},
),
Tool(
name="switch_focus",
description="Switch focus to ANY task by ID. The primary way to navigate your task workspace - jump between tasks in any order, revisit completed work, or change priorities on the fly. Current task retains its status, target task becomes current. Use get_big_picture or get_stack_overview to see task IDs.",
inputSchema={
"type": "object",
"properties": {
"task_id": {
"type": "string",
"description": "The ID of the task to switch focus to",
}
},
"required": ["task_id"],
},
),
Tool(
name="remove_task",
description="Remove a task from the structure (cleanup your workspace). The task remains in completed_tasks history if it was completed. Can remove any task (completed or not). Removing a parent task removes all its subtasks. Use get_big_picture or get_stack_overview to see task IDs.",
inputSchema={
"type": "object",
"properties": {
"task_id": {
"type": "string",
"description": "The ID of the task to remove from the structure",
}
},
"required": ["task_id"],
},
),
]