File size: 4,209 Bytes
310260a | 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 | """
Create Task MCP Tool
This tool enables the AI agent to create new tasks for the authenticated user.
Enforces user_id scoping and validates input parameters.
"""
from typing import Dict, Any
from sqlmodel import Session
from datetime import datetime
from ..models.task import Task
from .mcp_server import MCPContext
async def create_task_internal(
ctx: MCPContext,
title: str,
description: str = ""
) -> Dict[str, Any]:
"""
Internal MCP tool for creating a new task.
This function is called by the AgentService with user_id pre-bound.
It creates a task in the database and returns structured result.
Args:
ctx: MCP context containing db_engine and user_id
title: Task title (required, 1-200 chars)
description: Task description (optional, max 2000 chars)
Returns:
Dict containing:
- task: Created task details (id, title, description, completed, timestamps)
- status: "success" or "error"
- error: Error message (only if status is "error")
Error Cases:
- Empty title: Returns error "Title is required"
- Title too long: Returns error "Title exceeds 200 characters"
- Database error: Returns error with message
"""
try:
# Validate title
if not title or title.strip() == "":
return {
"status": "error",
"error": "Title is required"
}
title = title.strip()
if len(title) > 200:
return {
"status": "error",
"error": "Title exceeds 200 characters"
}
# Validate description length
if description and len(description) > 2000:
return {
"status": "error",
"error": "Description exceeds 2000 characters"
}
# Create task in database
with ctx.get_session() as session:
task = Task(
title=title,
description=description if description else None,
completed=False,
user_id=ctx.user_id,
created_at=datetime.utcnow(),
updated_at=datetime.utcnow()
)
session.add(task)
session.commit()
session.refresh(task)
# Return structured result
return {
"status": "success",
"task": {
"id": task.id,
"title": task.title,
"description": task.description,
"completed": task.completed,
"created_at": task.created_at.isoformat(),
"updated_at": task.updated_at.isoformat()
}
}
except Exception as e:
# Log error and return structured error response
print(f"Error creating task: {str(e)}")
return {
"status": "error",
"error": f"Database error: {str(e)}"
}
def get_tool_definition() -> Dict[str, Any]:
"""
Get OpenAI function calling definition for create_task tool.
Returns:
Tool definition in OpenAI function calling format
"""
return {
"type": "function",
"function": {
"name": "create_task",
"description": (
"Create a new task for the user. "
"Use this when the user wants to add a task, reminder, or todo item. "
"Examples: 'remind me to X', 'add task X', 'I need to X', 'create a task for X'."
),
"parameters": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "The task title or main description (required, 1-200 characters)"
},
"description": {
"type": "string",
"description": "Additional details or notes about the task (optional, max 2000 characters)"
}
},
"required": ["title"]
}
}
}
|