Initial deployment of customer support chatbot

README.md

@@ -1,13 +1,73 @@
 ---
 title: Customer Support Chatbot
-emoji: 📊
-colorFrom: red
-colorTo: red
+emoji: 🤖
+colorFrom: blue
+colorTo: purple
 sdk: gradio
 sdk_version: 6.1.0
 app_file: app.py
 pinned: false
-short_description: Customer Support Chatbot
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Customer Support Chatbot
+
+A prototype customer support chatbot for a computer products company, built with Gradio, OpenAI GPT-4o-mini, and MCP server integration.
+
+## Features
+
+- **Product Inquiries**: Browse, search, and get details about products (no authentication required)
+- **Order Management**: View and manage orders (requires authentication)
+- **Session Memory**: Maintains conversation context within each session
+- **Authentication**: Simple email/PIN authentication for order access
+
+## How to Use
+
+### Product Queries (No Authentication)
+- "Show me monitors"
+- "What products do you have?"
+- "Tell me about product COM-0001"
+- "Search for printers"
+
+### Order Queries (Requires Authentication)
+- "Show my orders"
+- "What's the status of my order?"
+- "I want to place an order"
+
+### Authentication
+To authenticate, include in your message:
+```
+email: your@email.com, pin: 1234
+```
+
+## Test Customers
+
+The following test customers are available:
+
+- donaldgarcia@example.net / 7912
+- michellejames@example.com / 1520
+- laurahenderson@example.org / 1488
+- spenceamanda@example.org / 2535
+- glee@example.net / 4582
+- williamsthomas@example.net / 4811
+- justin78@example.net / 9279
+- jason31@example.com / 1434
+- samuel81@example.com / 4257
+- williamleon@example.net / 9928
+
+## Architecture
+
+- **app.py**: Gradio UI and session management
+- **agent.py**: LLM agent with tool calling
+- **mcp_client.py**: MCP server JSON-RPC client
+- **auth.py**: Authentication handler
+- **memory.py**: Session-based conversation memory
+- **config.py**: Configuration management
+
+## Environment Variables
+
+Set these in your HuggingFace Space settings:
+
+- `OPENAI_API_KEY`: Your OpenAI API key
+- `MCP_SERVER_URL`: MCP server URL (default: https://vipfapwm3x.us-east-1.awsapprunner.com/mcp)
+

agent.py

@@ -0,0 +1,309 @@
+"""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):
+        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."""
+        # 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'"
+        
+        # Build system message with authentication status
+        is_authenticated = self.auth_handler.is_authenticated(session_id)
+        auth_status = "authenticated" if is_authenticated else "not authenticated"
+        customer_email = self.auth_handler.get_email(session_id) if is_authenticated else None
+        
+        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 += """
+
+When a customer asks about orders and is not authenticated, ask them to provide their email and PIN.
+If the customer is already authenticated, you can directly help with their orders.
+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})
+        
+        # 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 self.auth_handler.is_authenticated(session_id):
+            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'"
+        
+        try:
+            # Call OpenAI with tool calling
+            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 not provided
+                    if self._requires_auth(tool_name):
+                        customer_id = self._get_customer_id(session_id)
+                        if customer_id:
+                            # Add customer_id to tool args if not present and tool needs it
+                            if tool_name in ["list_orders", "get_customer", "create_order"]:
+                                if "customer_id" not in tool_args or not tool_args.get("customer_id"):
+                                    tool_args["customer_id"] = customer_id
+                        elif tool_name == "list_orders":
+                            # For list_orders, if no customer_id, we can still call it without filter
+                            # The MCP server will handle it
+                            pass
+                    
+                    # 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 = 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."
+

app.py

@@ -0,0 +1,135 @@
+"""Gradio UI for customer support chatbot."""
+import gradio as gr
+import re
+from typing import Optional
+from agent import SupportAgent
+from mcp_client import MCPClient
+from auth import AuthHandler
+from memory import SessionMemory
+
+
+# Initialize components
+mcp_client = MCPClient()
+auth_handler = AuthHandler(mcp_client)
+memory = SessionMemory()
+agent = SupportAgent(mcp_client, auth_handler)
+
+
+def parse_auth(message: str) -> tuple[Optional[str], Optional[str]]:
+    """Parse email and PIN from message."""
+    email_pattern = r'email:\s*([^\s,]+)'
+    pin_pattern = r'pin:\s*(\d{4})'
+    
+    email_match = re.search(email_pattern, message, re.IGNORECASE)
+    pin_match = re.search(pin_pattern, message, re.IGNORECASE)
+    
+    email = email_match.group(1) if email_match else None
+    pin = pin_match.group(1) if pin_match else None
+    
+    return email, pin
+
+
+def chat_response(message, history, session_id):
+    """Handle chat message and return response."""
+    if not message:
+        return history, ""
+    
+    # Check if message contains authentication
+    email, pin = parse_auth(message)
+    if email and pin:
+        # Attempt authentication
+        success, msg = auth_handler.authenticate(session_id, email, pin)
+        if success:
+            response = "✅ Authentication successful! How can I help you today?"
+        else:
+            response = f"❌ Authentication failed: {msg}. Please check your email and PIN and try again."
+        
+        # Gradio 6.x format: list of dicts with role and content
+        history.append({"role": "user", "content": message})
+        history.append({"role": "assistant", "content": response})
+        memory.add_message(session_id, "user", message)
+        memory.add_message(session_id, "assistant", response)
+        return history, ""
+    
+    # Get conversation history
+    conv_history = memory.get_conversation_context(session_id)
+    
+    # Process message with agent
+    response = agent.process_message(session_id, message, conv_history)
+    
+    # Update history - Gradio 6.x format
+    history.append({"role": "user", "content": message})
+    history.append({"role": "assistant", "content": response})
+    memory.add_message(session_id, "user", message)
+    memory.add_message(session_id, "assistant", response)
+    
+    return history, ""
+
+
+def create_interface():
+    """Create Gradio interface."""
+    with gr.Blocks() as demo:
+        gr.Markdown("# 🛒 Customer Support Chatbot")
+        gr.Markdown("Welcome! I can help you with product inquiries and order management.")
+        gr.Markdown("**Note:** For order-related queries, you'll need to authenticate with your email and PIN.")
+        
+        chatbot = gr.Chatbot(
+            label="Chat",
+            height=500
+        )
+        
+        msg = gr.Textbox(
+            label="Your Message",
+            placeholder="Type your message here...",
+            lines=2
+        )
+        
+        with gr.Row():
+            submit_btn = gr.Button("Send", variant="primary")
+            clear_btn = gr.Button("Clear Chat")
+        
+        gr.Markdown("### Authentication")
+        gr.Markdown("To authenticate, include in your message: `email: your@email.com, pin: 1234`")
+        
+        # Session state
+        session_id = gr.State(value="default_session")
+        
+        # Event handlers
+        def submit_message(message, history, session):
+            history, _ = chat_response(message, history, session)
+            return history, ""
+        
+        def clear_chat(session):
+            memory.clear(session)
+            auth_handler.clear_auth(session)
+            return []  # Return empty list for Gradio 6.x
+        
+        submit_btn.click(
+            submit_message,
+            inputs=[msg, chatbot, session_id],
+            outputs=[chatbot, msg]
+        )
+        
+        msg.submit(
+            submit_message,
+            inputs=[msg, chatbot, session_id],
+            outputs=[chatbot, msg]
+        )
+        
+        clear_btn.click(
+            clear_chat,
+            inputs=[session_id],
+            outputs=[chatbot]
+        )
+    
+    return demo
+
+
+if __name__ == "__main__":
+    demo = create_interface()
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=False
+    )
+

auth.py

@@ -0,0 +1,67 @@
+"""Authentication handler for customer sessions."""
+import re
+from typing import Dict, Optional, Any
+from mcp_client import MCPClient
+
+
+class AuthHandler:
+    """Manages customer authentication state per session."""
+    
+    def __init__(self, mcp_client: MCPClient):
+        self.mcp_client = mcp_client
+        self.auth_state: Dict[str, Dict[str, Any]] = {}
+    
+    def authenticate(self, session_id: str, email: str, pin: str) -> tuple[bool, str]:
+        """Authenticate customer and store session state."""
+        try:
+            result = self.mcp_client.verify_customer(email, pin)
+            
+            # Extract customer_id from result
+            # The result contains formatted text with customer details
+            # We'll extract customer_id from the text or structured content
+            customer_id = None
+            customer_info_text = ""
+            
+            if "content" in result and len(result["content"]) > 0:
+                customer_info_text = result["content"][0].get("text", "")
+            elif "structuredContent" in result:
+                customer_info_text = result["structuredContent"].get("result", "")
+            
+            # Try to extract customer_id from text (format: "Customer ID: <uuid>" or similar)
+            uuid_pattern = r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'
+            matches = re.findall(uuid_pattern, customer_info_text, re.IGNORECASE)
+            if matches:
+                customer_id = matches[0]
+            
+            self.auth_state[session_id] = {
+                "email": email,
+                "authenticated": True,
+                "customer_id": customer_id,
+                "customer_info": result,
+                "customer_info_text": customer_info_text
+            }
+            return True, "Authentication successful"
+        except Exception as e:
+            return False, str(e)
+    
+    def is_authenticated(self, session_id: str) -> bool:
+        """Check if session is authenticated."""
+        return self.auth_state.get(session_id, {}).get("authenticated", False)
+    
+    def get_email(self, session_id: str) -> Optional[str]:
+        """Get authenticated email for session."""
+        return self.auth_state.get(session_id, {}).get("email")
+    
+    def get_customer_info(self, session_id: str) -> Optional[Dict[str, Any]]:
+        """Get customer info for authenticated session."""
+        return self.auth_state.get(session_id, {}).get("customer_info")
+    
+    def get_customer_id(self, session_id: str) -> Optional[str]:
+        """Get customer ID for authenticated session."""
+        return self.auth_state.get(session_id, {}).get("customer_id")
+    
+    def clear_auth(self, session_id: str):
+        """Clear authentication for session."""
+        if session_id in self.auth_state:
+            del self.auth_state[session_id]
+

config.py

@@ -0,0 +1,20 @@
+"""Configuration management for the customer support chatbot."""
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# OpenAI Configuration
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+if not OPENAI_API_KEY:
+    raise ValueError("OPENAI_API_KEY not found in environment variables")
+
+# MCP Server Configuration
+MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "https://vipfapwm3x.us-east-1.awsapprunner.com/mcp")
+
+# HuggingFace Configuration (for deployment)
+HF_TOKEN = os.getenv("HF_TOKEN", "")
+
+# OpenAI Model
+OPENAI_MODEL = "gpt-4o-mini"
+

mcp_client.py

@@ -0,0 +1,81 @@
+"""MCP server client for JSON-RPC communication."""
+import requests
+import json
+from typing import Dict, Any, Optional
+from config import MCP_SERVER_URL
+
+
+class MCPClient:
+    """Client for communicating with MCP server via JSON-RPC 2.0."""
+    
+    def __init__(self):
+        self.url = MCP_SERVER_URL
+        self.request_id = 0
+        self._initialized = False
+    
+    def _get_next_id(self) -> int:
+        """Get next request ID."""
+        self.request_id += 1
+        return self.request_id
+    
+    def _call(self, method: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        """Make JSON-RPC call to MCP server."""
+        payload = {
+            "jsonrpc": "2.0",
+            "id": self._get_next_id(),
+            "method": method
+        }
+        if params:
+            payload["params"] = params
+        
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json"
+        }
+        
+        try:
+            response = requests.post(self.url, json=payload, headers=headers, timeout=10)
+            response.raise_for_status()
+            result = response.json()
+            
+            if "error" in result:
+                raise Exception(f"MCP Error: {result['error'].get('message', 'Unknown error')}")
+            
+            return result.get("result", {})
+        except requests.exceptions.RequestException as e:
+            raise Exception(f"Failed to communicate with MCP server: {str(e)}")
+    
+    def initialize(self):
+        """Initialize MCP connection."""
+        if self._initialized:
+            return
+        
+        result = self._call("initialize", {
+            "protocolVersion": "2024-11-05",
+            "capabilities": {},
+            "clientInfo": {
+                "name": "customer-support-chatbot",
+                "version": "1.0.0"
+            }
+        })
+        self._initialized = True
+        return result
+    
+    def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+        """Call an MCP tool."""
+        if not self._initialized:
+            self.initialize()
+        
+        result = self._call("tools/call", {
+            "name": tool_name,
+            "arguments": arguments
+        })
+        return result
+    
+    def verify_customer(self, email: str, pin: str) -> Dict[str, Any]:
+        """Verify customer with email and PIN."""
+        return self.call_tool("verify_customer_pin", {
+            "email": email,
+            "pin": pin
+        })
+

memory.py

@@ -0,0 +1,34 @@
+"""Session-based conversation memory manager."""
+from typing import List, Dict, Any
+
+
+class SessionMemory:
+    """Manages conversation memory per session."""
+    
+    def __init__(self):
+        self.memories: Dict[str, List[Dict[str, str]]] = {}
+    
+    def get_messages(self, session_id: str) -> List[Dict[str, str]]:
+        """Get conversation history for a session."""
+        return self.memories.get(session_id, [])
+    
+    def add_message(self, session_id: str, role: str, content: str):
+        """Add a message to session memory."""
+        if session_id not in self.memories:
+            self.memories[session_id] = []
+        
+        self.memories[session_id].append({
+            "role": role,
+            "content": content
+        })
+    
+    def clear(self, session_id: str):
+        """Clear memory for a session."""
+        if session_id in self.memories:
+            del self.memories[session_id]
+    
+    def get_conversation_context(self, session_id: str, max_messages: int = 10) -> List[Dict[str, str]]:
+        """Get recent conversation context."""
+        messages = self.get_messages(session_id)
+        return messages[-max_messages:] if len(messages) > max_messages else messages
+

requirements.txt

@@ -0,0 +1,5 @@
+gradio>=4.0.0
+openai>=1.0.0
+python-dotenv>=1.0.0
+requests>=2.31.0
+