File size: 15,025 Bytes
67f8819 b471009 67f8819 | 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 335 336 337 338 339 340 341 342 343 344 345 346 347 348 | """
TodoAgent - OpenAI Agent orchestration with MCP tools.
Per @specs/001-chatbot-mcp/plan.md Section VIII - AI Chatbot Architecture
OpenAI Agents SDK for orchestration with MCP tool calling.
"""
from openai import OpenAI
from typing import List, Dict, Any, Optional, AsyncGenerator
from uuid import UUID
import logging
import json
from config import settings
from mcp.server import get_mcp_server
logger = logging.getLogger(__name__)
class TodoAgent:
"""
OpenAI Agent for Todo task management with MCP tool integration.
This agent:
- Uses OpenAI Chat Completions API with tool calling
- Integrates with MCP tools for task operations
- Maintains conversation context for multi-turn dialogs
- Enforces user isolation by passing user_id to all tool calls
"""
# System prompt for friendly, concise, task-oriented behavior
SYSTEM_PROMPT = """You are a helpful task management assistant for TaskFlow. Your role is to help users manage their todo tasks through natural language.
Key behaviors:
- Be friendly and concise in your responses
- Focus on helping users create, view, complete, and delete tasks
- When users ask to add tasks, extract the task title and any description
- When users ask to see tasks, list their tasks clearly with numbers for easy reference
- When users ask to complete/delete/update tasks, ALWAYS call list_tasks FIRST to identify the task
- Users may refer to tasks by number (e.g., "first task", "task 2") or by title/content
- Use the available tools to perform all task operations
- Never make up task information - always use the tools
- The user_id parameter is automatically injected - do NOT ask the user for it
Task identification workflow:
1. When user wants to update/delete/complete a task, FIRST call list_tasks
2. Show the user the tasks with numbers: "1. Task title", "2. Task title", etc.
3. If user specified a number or title, match it to get the task_id
4. Then call the appropriate tool (update_task, delete_task, complete_task) with the task_id
Available tools:
- add_task: Create a new task with title and optional description
- list_tasks: Show all tasks with their IDs (call this FIRST before update/delete/complete)
- complete_task: Mark a task as completed (requires task_id from list_tasks)
- delete_task: Remove a task permanently (requires task_id from list_tasks)
- update_task: Change a task's title or description (requires task_id from list_tasks)
Note: user_id is automatically provided for all tool calls. Never ask the user for their user ID."""
def __init__(self, user_id: UUID):
"""
Initialize the TodoAgent for a specific user.
Args:
user_id: The user's UUID for data isolation
"""
self.user_id = user_id
# Log API key info for debugging (don't log full key)
key_preview = settings.openai_api_key[:10] if settings.openai_api_key else "None"
key_length = len(settings.openai_api_key) if settings.openai_api_key else 0
logger.info(f"TodoAgent initializing with OpenAI API key: {key_preview}... (length: {key_length})")
self.client = OpenAI(api_key=settings.openai_api_key)
self.model = settings.openai_model
# Get MCP server and register tools as OpenAI functions
self.mcp_server = get_mcp_server()
self.tools = self._get_openai_tools()
logger.info(f"TodoAgent initialized for user {user_id}")
def _get_openai_tools(self) -> List[Dict[str, Any]]:
"""
Convert MCP tools to OpenAI function format.
Returns:
List of tool definitions in OpenAI format
Per @specs/001-chatbot-mcp/contracts/mcp-tools.json
"""
tools = []
# Get all registered MCP tools from SimpleMCPRegistry
mcp_tools = self.mcp_server.list_tools()
for tool in mcp_tools:
# Convert MCP tool schema to OpenAI function format
function_def = {
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters.copy() # Copy to avoid mutation
}
}
# Add user_id parameter to all tools for data isolation
# Note: user_id is auto-filled and should NOT be in required parameters
if "properties" not in function_def["function"]["parameters"]:
function_def["function"]["parameters"]["properties"] = {}
function_def["function"]["parameters"]["properties"]["user_id"] = {
"type": "string",
"description": "User ID for data isolation (auto-filled, do not ask user)",
}
# Initialize required array if not present
if "required" not in function_def["function"]["parameters"]:
function_def["function"]["parameters"]["required"] = []
# IMPORTANT: Do NOT add user_id to required - it's auto-injected
# This prevents the model from asking the user for their user_id
tools.append(function_def)
logger.info(f"Registered {len(tools)} tools with OpenAI agent")
return tools
async def process_message(
self,
user_message: str,
conversation_history: Optional[List[Dict[str, str]]] = None
) -> AsyncGenerator[str, None]:
"""
Process a user message through the agent with tool execution.
Args:
user_message: The user's message text
conversation_history: Previous messages in the conversation
Yields:
Response text chunks as they're generated
Per @specs/001-chatbot-mcp/plan.md - MCP First with OpenAI Agents
"""
# Build messages array
messages = [
{"role": "system", "content": self.SYSTEM_PROMPT}
]
# Add conversation history if provided
if conversation_history:
messages.extend(conversation_history)
# Add current user message
messages.append({"role": "user", "content": user_message})
logger.info(f"Processing message for user {self.user_id}: {user_message[:50]}...")
# Check if this is an update/delete/complete request
# If so, pre-load tasks to provide context
update_delete_keywords = ["update", "delete", "remove", "change", "modify", "complete", "finish", "mark"]
needs_task_list = any(keyword in user_message.lower() for keyword in update_delete_keywords)
if needs_task_list:
# Get tasks first and add to context
list_tool = self.mcp_server.get_tool("list_tasks")
if list_tool:
tasks_result = await list_tool.handler(user_id=str(self.user_id), include_completed=True)
if tasks_result.get("success") and tasks_result.get("tasks"):
# Add task context to system prompt
task_list = "\n".join([
f"Task {i+1}: ID={t['id']}, Title='{t['title']}', Completed={t['completed']}"
for i, t in enumerate(tasks_result["tasks"])
])
enhanced_prompt = self.SYSTEM_PROMPT + f"\n\nCURRENT USER TASKS:\n{task_list}\n\nWhen user refers to a task by number or title, use the corresponding ID from this list."
messages[0] = {"role": "system", "content": enhanced_prompt}
logger.info(f"Pre-loaded {len(tasks_result['tasks'])} tasks for context")
try:
# Make chat completion request with tools
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.tools,
tool_choice="auto", # Let model decide when to use tools
temperature=0.7, # Slightly creative but focused
max_tokens=1000, # Reasonable response length
)
# Process response
choice = response.choices[0]
message = choice.message
# Check if model wants to call tools
if message.tool_calls:
# Execute tool calls and collect results
tool_messages = []
for tool_call in message.tool_calls:
result = await self._execute_tool_call(tool_call)
# Add tool result as a tool message
tool_messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# Add assistant message with tool calls
messages.append({
"role": "assistant",
"content": message.content or "",
"tool_calls": message.tool_calls
})
# Add tool result messages
messages.extend(tool_messages)
# Get follow-up response with tool results
follow_up = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.7,
max_tokens=1000,
)
if follow_up.choices[0].message.content:
yield follow_up.choices[0].message.content
# Direct text response (no tools needed)
elif message.content:
yield message.content
else:
yield "I understand. How can I help you with your tasks?"
except Exception as e:
error_type = type(e).__name__
error_msg = str(e)
logger.error(f"Error processing message: {error_type}: {error_msg}", exc_info=True)
# Provide more helpful error messages
if "Connection" in error_msg or "connect" in error_msg.lower():
yield "I'm having trouble connecting to my AI service. Please check if the OpenAI API key is configured correctly in Railway environment variables."
elif "401" in error_msg or "Unauthorized" in error_msg or "authentication" in error_msg.lower():
yield "My AI service credentials are invalid. Please check the OpenAI API key in Railway environment variables."
elif "rate" in error_msg.lower() or "limit" in error_msg.lower():
yield "I've reached my rate limit. Please try again in a moment."
else:
yield f"I encountered an error ({error_type}): {error_msg}"
async def _execute_tool_call(self, tool_call) -> str:
"""
Execute a single tool call from OpenAI.
Args:
tool_call: The OpenAI tool call object
Returns:
Result message to display to user
Per @specs/001-chatbot-mcp/plan.md - MCP First architecture
"""
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
# Inject user_id for data isolation
function_args["user_id"] = str(self.user_id)
logger.info(f"Executing tool: {function_name} with args: {function_args}")
try:
# Get the tool from SimpleMCPRegistry
tool = self.mcp_server.get_tool(function_name)
if not tool:
return f"Error: Tool '{function_name}' not found"
# For update_task and delete_task, validate task_id exists first
if function_name in ["update_task", "delete_task", "complete_task"]:
task_id = function_args.get("task_id")
if task_id:
# Verify the task exists and belongs to user before proceeding
list_tool = self.mcp_server.get_tool("list_tasks")
tasks_result = await list_tool.handler(user_id=str(self.user_id), include_completed=True)
valid_task_ids = [t["id"] for t in tasks_result.get("tasks", [])]
if task_id not in valid_task_ids:
# Task doesn't exist or doesn't belong to user
# Provide helpful error with current tasks
if tasks_result.get("tasks"):
task_list = "\n".join([
f"Task {i+1}: {t['title']}"
for i, t in enumerate(tasks_result["tasks"])
])
return f"Error: Task not found. Here are your current tasks:\n{task_list}\n\nPlease specify which task you'd like to {function_name.replace('_', ' ')}."
else:
return "Error: You don't have any tasks yet. Create some tasks first!"
# Execute the tool via MCP
result = await tool.handler(**function_args)
# Parse result
if isinstance(result, dict):
if result.get("success"):
# Format success message based on tool
if function_name == "add_task":
return f"✓ Task '{result.get('title')}' created successfully!"
elif function_name == "complete_task":
return f"✓ Task '{result.get('title')}' marked as complete!"
elif function_name == "delete_task":
return f"✓ Task '{result.get('title')}' deleted."
elif function_name == "update_task":
return f"✓ Task updated successfully!"
elif function_name == "list_tasks":
tasks = result.get("tasks", [])
count = result.get("count", 0)
if count == 0:
return "You don't have any tasks yet."
# Number tasks for easy reference
task_list = "\n".join([
f"{i+1}. {t['title']}" + (" ✓" if t['completed'] else "")
for i, t in enumerate(tasks)
])
return f"You have {count} task(s):\n{task_list}\n\nYou can refer to tasks by number (e.g., \"complete task 1\")"
else:
return "Operation completed successfully!"
else:
return f"Error: {result.get('error', 'Unknown error')}"
return str(result)
except Exception as e:
logger.error(f"Error executing tool {function_name}: {e}")
return f"Error executing {function_name}: {str(e)}"
def create_todo_agent(user_id: UUID) -> TodoAgent:
"""
Factory function to create a TodoAgent instance.
Args:
user_id: The user's UUID
Returns:
Initialized TodoAgent instance
"""
return TodoAgent(user_id)
|