File size: 3,553 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 | """
Get Task MCP Tool
This tool enables the AI agent to retrieve details of a specific task by ID.
Useful for multi-step operations and context-aware task references.
"""
from typing import Dict, Any
from sqlmodel import Session, select
from ..models.task import Task
from .mcp_server import MCPContext
async def get_task_internal(
ctx: MCPContext,
task_id: int
) -> Dict[str, Any]:
"""
Internal MCP tool for retrieving a specific task by ID.
This function is called by the AgentService with user_id pre-bound.
It retrieves detailed information about a single task.
Args:
ctx: MCP context containing db_engine and user_id
task_id: ID of the task to retrieve
Returns:
Dict containing:
- task: Task details (id, title, description, completed, timestamps)
- status: "success" or "error"
- error: Error message (only if status is "error")
Error Cases:
- Task not found: Returns error "Task not found"
- Task belongs to different user: Returns error "Task not found" (security)
- Invalid task_id: Returns error "Invalid task ID"
- Database error: Returns error with message
"""
try:
# Validate task_id
if not isinstance(task_id, int) or task_id <= 0:
return {
"status": "error",
"error": "Invalid task ID"
}
# Query task with user_id scoping for security
with ctx.get_session() as session:
statement = select(Task).where(
Task.id == task_id,
Task.user_id == ctx.user_id
)
task = session.exec(statement).first()
if not task:
return {
"status": "error",
"error": "Task not found"
}
# 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 getting 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 get_task tool.
Returns:
Tool definition in OpenAI function calling format
"""
return {
"type": "function",
"function": {
"name": "get_task",
"description": (
"Retrieve detailed information about a specific task by its ID. "
"Use this when the user asks for details about a particular task. "
"Examples: 'show me task 5', 'what's in task 3', 'details of the first task', "
"'tell me about task 10'."
),
"parameters": {
"type": "object",
"properties": {
"task_id": {
"type": "integer",
"description": "The ID of the task to retrieve"
}
},
"required": ["task_id"]
}
}
}
|