creativesar/taskflow
0
1"""2MCP Server Implementation3Phase III: AI Chatbot - MCP Tools for Task Management4 5This module implements the MCP server with 5 tools:6- add_task: Create a new task7- list_tasks: Retrieve tasks from user's list8- complete_task: Mark a task as complete9- delete_task: Remove a task from the list10- update_task: Modify task title or description11"""12 13from typing import Optional, List, Dict, Any14from mcp.server import Server15from mcp.types import Tool, TextContent16import json17from sqlmodel.ext.asyncio.session import AsyncSession18 19# Import database and service functions20from db import get_session, engine21from services import task_service22 23# Initialize MCP server24mcp_server = Server("todo-server")25 26 27def get_mcp_tools() -> List[Tool]:28 """29 Returns the list of all MCP tools available to the agent.30 31 Returns:32 List of Tool objects that can be used by OpenAI Agents SDK33 """34 tools = [35 Tool(36 name="add_task",37 description="Create a new task for the user",38 inputSchema={39 "type": "object",40 "properties": {41 "user_id": {42 "type": "string",43 "description": "The user ID who owns the task"44 },45 "title": {46 "type": "string",47 "description": "Task title (1-200 characters)"48 },49 "description": {50 "type": "string",51 "description": "Optional task description (0-1000 characters)"52 }53 },54 "required": ["user_id", "title"]55 }56 ),57 Tool(58 name="list_tasks",59 description="Retrieve tasks from the user's list",60 inputSchema={61 "type": "object",62 "properties": {63 "user_id": {64 "type": "string",65 "description": "The user ID whose tasks to retrieve"66 },67 "status": {68 "type": "string",69 "enum": ["all", "pending", "completed"],70 "description": "Filter tasks by status (default: all)"71 }72 },73 "required": ["user_id"]74 }75 ),76 Tool(77 name="complete_task",78 description="Mark a task as complete or toggle completion status",79 inputSchema={80 "type": "object",81 "properties": {82 "user_id": {83 "type": "string",84 "description": "The user ID who owns the task"85 },86 "task_id": {87 "type": "integer",88 "description": "The ID of the task to complete"89 }90 },91 "required": ["user_id", "task_id"]92 }93 ),94 Tool(95 name="delete_task",96 description="Remove a task from the list",97 inputSchema={98 "type": "object",99 "properties": {100 "user_id": {101 "type": "string",102 "description": "The user ID who owns the task"103 },104 "task_id": {105 "type": "integer",106 "description": "The ID of the task to delete"107 }108 },109 "required": ["user_id", "task_id"]110 }111 ),112 Tool(113 name="update_task",114 description="Modify task title or description",115 inputSchema={116 "type": "object",117 "properties": {118 "user_id": {119 "type": "string",120 "description": "The user ID who owns the task"121 },122 "task_id": {123 "type": "integer",124 "description": "The ID of the task to update"125 },126 "title": {127 "type": "string",128 "description": "New task title (1-200 characters)"129 },130 "description": {131 "type": "string",132 "description": "New task description (0-1000 characters)"133 }134 },135 "required": ["user_id", "task_id"]136 }137 )138 ]139 return tools140 141 142# MCP Tool Implementations (T-306 to T-310)143 144async def add_task(user_id: str, title: str, description: Optional[str] = None) -> Dict[str, Any]:145 """146 Create a new task for the user.147 148 Args:149 user_id: The user ID who owns the task150 title: Task title (1-200 characters)151 description: Optional task description (0-1000 characters)152 153 Returns:154 Dict with task_id, status, and title155 156 Raises:157 ValueError: If validation fails158 """159 # Validate title160 if not title or len(title) < 1 or len(title) > 200:161 raise ValueError("Title must be 1-200 characters")162 163 # Validate description164 if description and len(description) > 1000:165 raise ValueError("Description cannot exceed 1000 characters")166 167 # Create task in database168 async with AsyncSession(engine) as session:169 task = await task_service.create_task(session, user_id, title, description)170 171 return {172 "task_id": task.id,173 "status": "created",174 "title": task.title175 }176 177 178async def list_tasks(user_id: str, status: str = "all") -> List[Dict[str, Any]]:179 """180 Retrieve tasks from the user's list.181 182 Args:183 user_id: The user ID whose tasks to retrieve184 status: Filter by status ("all", "pending", "completed")185 186 Returns:187 List of task dictionaries188 189 Raises:190 ValueError: If status is invalid191 """192 # Validate status193 if status not in ["all", "pending", "completed"]:194 raise ValueError("Invalid status. Use 'all', 'pending', or 'completed'")195 196 # Query tasks197 async with AsyncSession(engine) as session:198 tasks = await task_service.list_tasks(session, user_id)199 200 # Filter by status201 if status == "pending":202 tasks = [t for t in tasks if not t.completed]203 elif status == "completed":204 tasks = [t for t in tasks if t.completed]205 206 # Convert to dict format207 return [208 {209 "id": task.id,210 "title": task.title,211 "description": task.description,212 "completed": task.completed,213 "created_at": task.created_at.isoformat()214 }215 for task in tasks216 ]217 218 219async def complete_task(user_id: str, task_id: int) -> Dict[str, Any]:220 """221 Mark a task as complete or toggle completion status.222 223 Args:224 user_id: The user ID who owns the task225 task_id: The ID of the task to complete226 227 Returns:228 Dict with task_id, status, and title229 230 Raises:231 ValueError: If task not found or access denied232 """233 async with AsyncSession(engine) as session:234 task = await task_service.toggle_completion(session, task_id, user_id)235 236 if not task:237 raise ValueError("Task not found or access denied")238 239 return {240 "task_id": task.id,241 "status": "completed" if task.completed else "pending",242 "title": task.title243 }244 245 246async def delete_task(user_id: str, task_id: int) -> Dict[str, Any]:247 """248 Remove a task from the list.249 250 Args:251 user_id: The user ID who owns the task252 task_id: The ID of the task to delete253 254 Returns:255 Dict with task_id, status, and title256 257 Raises:258 ValueError: If task not found or access denied259 """260 async with AsyncSession(engine) as session:261 # Get task first to return its title262 task = await task_service.get_task(session, task_id, user_id)263 264 if not task:265 raise ValueError("Task not found or access denied")266 267 title = task.title268 269 # Delete task270 deleted = await task_service.delete_task(session, task_id, user_id)271 272 if not deleted:273 raise ValueError("Failed to delete task")274 275 return {276 "task_id": task_id,277 "status": "deleted",278 "title": title279 }280 281 282async def update_task(283 user_id: str,284 task_id: int,285 title: Optional[str] = None,286 description: Optional[str] = None287) -> Dict[str, Any]:288 """289 Modify task title or description.290 291 Args:292 user_id: The user ID who owns the task293 task_id: The ID of the task to update294 title: New task title (1-200 characters)295 description: New task description (0-1000 characters)296 297 Returns:298 Dict with task_id, status, and title299 300 Raises:301 ValueError: If validation fails or task not found302 """303 # Validate title if provided304 if title and (len(title) < 1 or len(title) > 200):305 raise ValueError("Title must be 1-200 characters")306 307 # Validate description if provided308 if description and len(description) > 1000:309 raise ValueError("Description cannot exceed 1000 characters")310 311 async with AsyncSession(engine) as session:312 # Get existing task313 task = await task_service.get_task(session, task_id, user_id)314 315 if not task:316 raise ValueError("Task not found or access denied")317 318 # Use existing values if not provided319 new_title = title if title else task.title320 new_description = description if description is not None else task.description321 322 # Update task323 updated_task = await task_service.update_task(324 session, task_id, user_id, new_title, new_description325 )326 327 if not updated_task:328 raise ValueError("Failed to update task")329 330 return {331 "task_id": updated_task.id,332 "status": "updated",333 "title": updated_task.title334 }335 336 337# Verify imports work correctly338if __name__ == "__main__":339 print("[OK] MCP SDK imported successfully")340 print(f"[OK] MCP server '{mcp_server.name}' initialized")341 tools = get_mcp_tools()342 print(f"[OK] {len(tools)} MCP tools defined:")343 for tool in tools:344 print(f" - {tool.name}: {tool.description}")345 print("\n[OK] All 5 MCP tool implementations completed:")346 print(" - add_task: Create new tasks")347 print(" - list_tasks: Retrieve tasks with filtering")348 print(" - complete_task: Toggle task completion")349 print(" - delete_task: Remove tasks")350 print(" - update_task: Modify task details")351 print("\nMCP server is ready for integration with OpenAI Agents SDK.")352