"""LLM agent with tool calling capabilities.""" import json from typing import Dict, Any, List, Optional from openai import OpenAI from config import OPENAI_API_KEY, OPENAI_MODEL from mcp_client import MCPClient from auth import AuthHandler class SupportAgent: """Customer support agent with MCP tool integration.""" def __init__(self, mcp_client: MCPClient, auth_handler: AuthHandler): # Initialize OpenAI client self.client = OpenAI(api_key=OPENAI_API_KEY) self.model = OPENAI_MODEL self.mcp_client = mcp_client self.auth_handler = auth_handler # Initialize MCP connection self.mcp_client.initialize() # Define available tools self.tools = self._define_tools() def _define_tools(self) -> List[Dict[str, Any]]: """Define tool schemas for OpenAI function calling.""" return [ { "type": "function", "function": { "name": "list_products", "description": "List products with optional filters by category or active status", "parameters": { "type": "object", "properties": { "category": { "type": "string", "description": "Filter by category (e.g., 'Computers', 'Monitors', 'Printers')" }, "is_active": { "type": "boolean", "description": "Filter by active status" } } } } }, { "type": "function", "function": { "name": "get_product", "description": "Get detailed product information by SKU", "parameters": { "type": "object", "properties": { "sku": { "type": "string", "description": "Product SKU (e.g., 'COM-0001', 'MON-0054')" } }, "required": ["sku"] } } }, { "type": "function", "function": { "name": "search_products", "description": "Search products by name or description keyword", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search term (case-insensitive, partial match)" } }, "required": ["query"] } } }, { "type": "function", "function": { "name": "get_customer", "description": "Get customer information by customer ID. Requires authentication.", "parameters": { "type": "object", "properties": { "customer_id": { "type": "string", "description": "Customer UUID" } }, "required": ["customer_id"] } } }, { "type": "function", "function": { "name": "list_orders", "description": "List orders with optional filters. Requires authentication.", "parameters": { "type": "object", "properties": { "customer_id": { "type": "string", "description": "Filter by customer UUID" }, "status": { "type": "string", "description": "Filter by status: draft, submitted, approved, fulfilled, cancelled" } } } } }, { "type": "function", "function": { "name": "get_order", "description": "Get detailed order information including items. Requires authentication.", "parameters": { "type": "object", "properties": { "order_id": { "type": "string", "description": "Order UUID" } }, "required": ["order_id"] } } }, { "type": "function", "function": { "name": "create_order", "description": "Create a new order with items. Requires authentication.", "parameters": { "type": "object", "properties": { "customer_id": { "type": "string", "description": "Customer UUID" }, "items": { "type": "array", "description": "List of order items", "items": { "type": "object", "properties": { "sku": {"type": "string"}, "quantity": {"type": "integer"}, "unit_price": {"type": "string"}, "currency": {"type": "string", "default": "USD"} }, "required": ["sku", "quantity", "unit_price"] } } }, "required": ["customer_id", "items"] } } } ] def _requires_auth(self, tool_name: str) -> bool: """Check if tool requires authentication.""" auth_required_tools = ["get_customer", "list_orders", "get_order", "create_order"] return tool_name in auth_required_tools def _get_customer_id(self, session_id: str) -> Optional[str]: """Get customer_id from authenticated session.""" if not self.auth_handler.is_authenticated(session_id): return None return self.auth_handler.get_customer_id(session_id) def process_message(self, session_id: str, user_message: str, conversation_history: List[Dict[str, str]]) -> str: """Process user message and return response.""" # Get authentication status is_authenticated = self.auth_handler.is_authenticated(session_id) customer_email = self.auth_handler.get_email(session_id) if is_authenticated else None # Check if message is about authentication if "email" in user_message.lower() and "pin" in user_message.lower(): # Try to extract email and PIN from message # This is a simple approach - in production, use structured input return "To authenticate, please provide your email and PIN in the format: 'email: your@email.com, pin: 1234'" # Check if query might need authentication order_keywords = ["order", "purchase", "buy", "my orders", "order history", "track order", "place order"] needs_auth = any(keyword in user_message.lower() for keyword in order_keywords) if needs_auth and not is_authenticated: return "To access your orders, I need to verify your identity. Please provide your email and PIN in this format: 'email: your@email.com, pin: 1234'" # Process with LLM (all API calls are automatically logged in OpenAI Platform under Logs → Completions) response_text = self._process_with_llm(session_id, user_message, conversation_history, is_authenticated, customer_email) return response_text def _process_with_llm(self, session_id: str, user_message: str, conversation_history: List[Dict[str, str]], is_authenticated: bool, customer_email: Optional[str]) -> str: """Internal method to process message with LLM.""" # Build system message with authentication status auth_status = "authenticated" if is_authenticated else "not authenticated" system_content = """You are a helpful customer support agent for a computer products company. You can help customers with: - Product inquiries (browsing, searching, getting details) - no authentication needed - Order management (viewing orders, order status, placing orders) - requires authentication Current session status: """ + auth_status if customer_email: system_content += f"\nAuthenticated customer: {customer_email}" system_content += """ IMPORTANT INSTRUCTIONS: - When a customer asks to see/list/show their orders, use the list_orders tool directly - When a customer asks about a specific order, use the get_order tool - The customer_id is already set for authenticated sessions - you don't need to provide it - Be friendly, professional, and helpful. Provide clear, concise answers.""" messages = [ { "role": "system", "content": system_content } ] # Add conversation history messages.extend(conversation_history) # Add current user message messages.append({"role": "user", "content": user_message}) try: # Call OpenAI with tool calling # Note: For standard OpenAI Python SDK, API calls appear in Logs -> Completions # The Traces tab is for OpenAI Agents SDK (JavaScript/TypeScript) response = self.client.chat.completions.create( model=self.model, messages=messages, tools=self.tools, tool_choice="auto" ) message = response.choices[0].message # Handle tool calls if message.tool_calls: tool_results = [] for tool_call in message.tool_calls: tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) # Check authentication for order-related tools if self._requires_auth(tool_name) and not self.auth_handler.is_authenticated(session_id): tool_results.append({ "role": "tool", "tool_call_id": tool_call.id, "name": tool_name, "content": "Authentication required. Please provide your email and PIN." }) continue # Inject customer_id for order-related tools if self._requires_auth(tool_name): customer_id = self._get_customer_id(session_id) if customer_id: # ALWAYS replace customer_id with the authenticated UUID # Don't trust what the LLM provides - it may provide email instead if tool_name in ["list_orders", "get_customer", "create_order"]: tool_args["customer_id"] = customer_id else: # If customer_id is not available, don't call the tool # This prevents using email as customer_id tool_results.append({ "role": "tool", "tool_call_id": tool_call.id, "name": tool_name, "content": "Error: Customer ID not found. Please re-authenticate." }) continue # Call MCP tool try: result = self.mcp_client.call_tool(tool_name, tool_args) # Extract text content from result if "content" in result and len(result["content"]) > 0: content = result["content"][0].get("text", str(result)) elif "structuredContent" in result: content = result["structuredContent"].get("result", str(result)) else: content = str(result) tool_results.append({ "role": "tool", "tool_call_id": tool_call.id, "name": tool_name, "content": content }) except Exception as e: tool_results.append({ "role": "tool", "tool_call_id": tool_call.id, "name": tool_name, "content": f"Error: {str(e)}" }) # Get final response with tool results messages.append(message) messages.extend(tool_results) # Final response - automatically traced final_response = self.client.chat.completions.create( model=self.model, messages=messages ) return final_response.choices[0].message.content else: return message.content except Exception as e: return f"I apologize, but I encountered an error: {str(e)}. Please try again."