Spaces:
Runtime error
Runtime error
File size: 12,762 Bytes
dd1b74d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 | """
MCP Server implementation for the AI Chatbot with Reusable Intelligence
Using the Python SDK to wrap existing Phase II logic as tools
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
import uuid
from datetime import datetime
# Import database and models to connect to real persistence layer
from database import SessionLocal, engine
from models.todo import Todo
from schemas.todo import TodoCreate, TodoUpdate
from services.todo_service import TodoService
def get_db():
"""Dependency to get database session"""
db = SessionLocal()
try:
yield db
finally:
db.close()
# Mock MCP SDK implementation - in a real scenario, this would import the actual MCP SDK
class MCPSDK:
"""Mock MCP SDK for demonstration purposes"""
@staticmethod
def execute_tool(tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
"""Execute an MCP tool with the given parameters"""
print(f"MCP SDK executing tool: {tool_name} with params: {params}")
return {"result": "success", "data": params}
class MCPTool(ABC):
"""Abstract base class for MCP tools"""
@abstractmethod
def execute(self, **kwargs) -> Dict[str, Any]:
"""Execute the tool with given parameters"""
pass
class AddTaskTool(MCPTool):
"""MCP tool for adding tasks via MCP SDK"""
def execute(self, title: str, description: Optional[str] = None,
status: str = "pending", priority: str = "medium",
due_date: Optional[str] = None,
conversation_id: Optional[str] = None, user_id: Optional[int] = None) -> Dict[str, Any]:
"""
Add a new task via MCP tools as required by FR-006
"""
params = {
"title": title,
"description": description,
"status": status,
"priority": priority,
"due_date": due_date,
"conversation_id": conversation_id,
"user_id": user_id,
"created_at": datetime.utcnow().isoformat()
}
# Execute through MCP SDK to ensure compliance with constitution
result = MCPSDK.execute_tool("add_task", params)
# Connect to real database and create the task
db = SessionLocal()
try:
# If user_id is None (not authenticated), we can't create a task since user_id is required
if user_id is None:
raise Exception("User must be authenticated to create tasks")
# Parse due_date if provided
parsed_due_date = None
if due_date:
try:
parsed_due_date = datetime.fromisoformat(due_date.replace('Z', '+00:00'))
except ValueError:
# If ISO format fails, try parsing as a standard datetime string
try:
parsed_due_date = datetime.strptime(due_date.split('.')[0], '%Y-%m-%dT%H:%M:%S')
except ValueError:
# If all parsing fails, ignore the due date
parsed_due_date = None
# Create TodoCreate schema object
todo_data = TodoCreate(
title=title,
description=description,
priority=priority,
due_date=parsed_due_date,
recurrence="none" # Default for chatbot-created tasks
)
# Use TodoService to create the task in the database
todo_service = TodoService(db, user_id=user_id)
created_todo = todo_service.create(todo_data)
# Return the created task data
return {
"id": created_todo.id,
"title": created_todo.title,
"description": created_todo.description,
"completed": created_todo.completed,
"priority": created_todo.priority,
"due_date": created_todo.due_date.isoformat() if created_todo.due_date else None,
"recurrence": created_todo.recurrence,
"created_at": created_todo.created_at.isoformat(),
"updated_at": created_todo.updated_at.isoformat(),
"user_id": created_todo.user_id,
"conversation_id": conversation_id
}
finally:
db.close()
class ListTasksTool(MCPTool):
"""MCP tool for listing tasks via MCP SDK"""
def execute(self, conversation_id: Optional[str] = None,
status: Optional[str] = None,
priority: Optional[str] = None, user_id: Optional[int] = None) -> List[Dict[str, Any]]:
"""
List tasks via MCP tools as required by FR-006
"""
params = {
"conversation_id": conversation_id,
"status": status,
"priority": priority,
"user_id": user_id
}
# Execute through MCP SDK to ensure compliance with constitution
result = MCPSDK.execute_tool("list_tasks", params)
# Connect to real database and fetch tasks
db = SessionLocal()
try:
# If user_id is None (not authenticated), we can't fetch tasks since user_id is required
if user_id is None:
raise Exception("User must be authenticated to list tasks")
# Use TodoService to fetch tasks from the database
todo_service = TodoService(db, user_id=user_id)
# Map status parameter for TodoService (TodoService uses completed=True/False)
status_param = None
if status:
if status.lower() == "completed":
status_param = "completed"
elif status.lower() == "pending":
status_param = "pending"
# Fetch tasks from database
todos = todo_service.get_all(status=status_param, priority=priority)
# Convert Todo objects to dictionary format
tasks = []
for todo in todos:
tasks.append({
"id": todo.id,
"title": todo.title,
"description": todo.description,
"completed": todo.completed,
"priority": todo.priority or "medium",
"due_date": todo.due_date.isoformat() if todo.due_date else None,
"recurrence": todo.recurrence,
"created_at": todo.created_at.isoformat(),
"updated_at": todo.updated_at.isoformat(),
"user_id": todo.user_id,
"conversation_id": conversation_id
})
return tasks
finally:
db.close()
class UpdateTaskTool(MCPTool):
"""MCP tool for updating tasks via MCP SDK"""
def execute(self, task_id: str, user_id: Optional[int] = None, **updates) -> Dict[str, Any]:
"""
Update a task via MCP tools as required by FR-006
"""
params = {
"task_id": task_id,
"user_id": user_id,
**updates
}
# Execute through MCP SDK to ensure compliance with constitution
result = MCPSDK.execute_tool("update_task", params)
# Connect to real database and update the task
db = SessionLocal()
try:
# If user_id is None (not authenticated), we can't update tasks since user_id is required
if user_id is None:
raise Exception("User must be authenticated to update tasks")
# Use TodoService to update the task in the database
todo_service = TodoService(db, user_id=user_id)
# Prepare update data in TodoUpdate format
todo_update_data = TodoUpdate(**{k: v for k, v in updates.items() if v is not None})
updated_todo = todo_service.update(int(task_id), todo_update_data)
if updated_todo:
return {
"id": updated_todo.id,
"title": updated_todo.title,
"description": updated_todo.description,
"completed": updated_todo.completed,
"priority": updated_todo.priority or "medium",
"due_date": updated_todo.due_date.isoformat() if updated_todo.due_date else None,
"recurrence": updated_todo.recurrence,
"created_at": updated_todo.created_at.isoformat(),
"updated_at": updated_todo.updated_at.isoformat(),
"user_id": updated_todo.user_id
}
else:
raise Exception(f"Task with id {task_id} not found or user not authorized")
finally:
db.close()
class DeleteTaskTool(MCPTool):
"""MCP tool for deleting tasks via MCP SDK"""
def execute(self, task_id: str, user_id: Optional[int] = None) -> Dict[str, Any]:
"""
Delete a task via MCP tools as required by FR-006
"""
params = {
"task_id": task_id,
"user_id": user_id
}
# Execute through MCP SDK to ensure compliance with constitution
result = MCPSDK.execute_tool("delete_task", params)
# Connect to real database and delete the task
db = SessionLocal()
try:
# If user_id is None (not authenticated), we can't delete tasks since user_id is required
if user_id is None:
raise Exception("User must be authenticated to delete tasks")
# Use TodoService to delete the task from the database
todo_service = TodoService(db, user_id=user_id)
success = todo_service.delete(int(task_id))
if success:
return {
"message": f"Task {task_id} deleted successfully",
"task_id": task_id,
"success": True
}
else:
raise Exception(f"Task with id {task_id} not found or user not authorized")
finally:
db.close()
class MCPToolRegistry:
"""Registry for managing MCP tools"""
def __init__(self):
self.tools = {
"add_task": AddTaskTool(),
"list_tasks": ListTasksTool(),
"update_task": UpdateTaskTool(),
"delete_task": DeleteTaskTool()
}
def get_tool(self, tool_name: str) -> MCPTool:
"""Get a tool by name"""
return self.tools.get(tool_name)
def execute_tool(self, tool_name: str, **kwargs) -> Dict[str, Any]:
"""Execute a tool by name with given parameters"""
tool = self.get_tool(tool_name)
if not tool:
raise ValueError(f"Unknown tool: {tool_name}")
return tool.execute(**kwargs)
# Global MCP tool registry instance
mcp_registry = MCPToolRegistry()
def get_mcp_tool(tool_name: str) -> MCPTool:
"""Get an MCP tool by name"""
return mcp_registry.get_tool(tool_name)
def execute_mcp_tool(tool_name: str, **kwargs) -> Dict[str, Any]:
"""Execute an MCP tool with given parameters"""
return mcp_registry.execute_tool(tool_name, **kwargs)
# Implementation of MCP server tools for Neon DB interaction (T040)
def add_task_mcp(title: str, description: Optional[str] = None,
status: str = "pending", priority: str = "medium",
due_date: Optional[str] = None,
conversation_id: Optional[str] = None, user_id: Optional[int] = None) -> Dict[str, Any]:
"""MCP server tool for adding tasks to Neon DB"""
return execute_mcp_tool("add_task", title=title, description=description,
status=status, priority=priority, due_date=due_date,
conversation_id=conversation_id, user_id=user_id)
def list_tasks_mcp(conversation_id: Optional[str] = None,
status: Optional[str] = None,
priority: Optional[str] = None, user_id: Optional[int] = None) -> List[Dict[str, Any]]:
"""MCP server tool for listing tasks from Neon DB"""
return execute_mcp_tool("list_tasks", conversation_id=conversation_id,
status=status, priority=priority, user_id=user_id)
def update_task_mcp(task_id: str, user_id: Optional[int] = None, **updates) -> Dict[str, Any]:
"""MCP server tool for updating tasks in Neon DB"""
return execute_mcp_tool("update_task", task_id=task_id, user_id=user_id, **updates)
def delete_task_mcp(task_id: str, user_id: Optional[int] = None) -> Dict[str, Any]:
"""MCP server tool for deleting tasks from Neon DB"""
return execute_mcp_tool("delete_task", task_id=task_id, user_id=user_id) |