🍕 Restaurant WhatsApp Bot

AI-powered WhatsApp bot for food ordering using FastAPI, Google Gemini, and MCP (Model Context Protocol) server integration.

🏗️ Architecture Overview

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│   WhatsApp      │────│  FastAPI Bot     │────│   Google AI     │
│   Database      │    │  (MCP Client)    │    │   (Gemini)      │
└─────────────────┘    └──────────────────┘    └─────────────────┘
                              │
                       ┌──────────────────┐
                       │   MCP Server     │
                       │  (Claude/Tools)  │
                       └──────────────────┘

🚀 Setup Instructions

1. Clone the Repository

git clone https://github.com/yourusername/restaurant-whatsapp-bot.git
cd restaurant-whatsapp-bot

2. Create Virtual Environment

python -m venv fastapi_env
fastapi_env\Scripts\activate  # Windows
# or
source fastapi_env/bin/activate  # Linux/Mac

3. Install Dependencies

pip install -r requirements.txt

4. Environment Setup

cp .env.example .env
# Edit .env with your actual values

🖥️ MCP Server Integration

What is MCP Server?

The Model Context Protocol (MCP) server enables seamless integration between your bot and Claude/other AI models, providing enhanced capabilities for:

• 🔍 Advanced context management
• 🛠️ Tool integration (database queries, API calls)
• 📊 Real-time data processing
• 🤖 Multi-model AI orchestration

MCP Server Setup

1. Install MCP Server

# Install MCP server dependencies
npm install -g @modelcontextprotocol/server-claude
# or
pip install mcp-server-claude

2. Configure MCP Server

Create mcp-config.json:

{
  "mcpServers": {
    "restaurant-bot": {
      "command": "python",
      "args": ["mcp_server.py"],
      "env": {
        "MONGODB_URL": "mongodb://localhost:27017",
        "DATABASE_NAME": "restaurant_db"
      }
    }
  }
}

3. Create MCP Server Script

# Create mcp_server.py in your project root
touch mcp_server.py

4. MCP Server Implementation

Add this to mcp_server.py:

#!/usr/bin/env python3
import asyncio
import json
import sys
from typing import Any, Dict, List
from motor.motor_asyncio import AsyncIOMotorClient
from datetime import datetime

class RestaurantMCPServer:
    def __init__(self):
        self.mongodb_client = None
        self.db = None
        
    async def initialize(self):
        """Initialize database connections"""
        self.mongodb_client = AsyncIOMotorClient("mongodb://localhost:27017")
        self.db = self.mongodb_client.restaurant_db
        
    async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Handle MCP requests"""
        method = request.get("method")
        params = request.get("params", {})
        
        if method == "tools/list":
            return await self.list_tools()
        elif method == "tools/call":
            return await self.call_tool(params)
        elif method == "resources/list":
            return await self.list_resources()
        elif method == "resources/read":
            return await self.read_resource(params)
        else:
            return {"error": f"Unknown method: {method}"}
    
    async def list_tools(self) -> Dict[str, Any]:
        """List available tools for the restaurant bot"""
        return {
            "tools": [
                {
                    "name": "get_menu",
                    "description": "Get current restaurant menu items",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "category": {
                                "type": "string",
                                "description": "Menu category (optional)"
                            }
                        }
                    }
                },
                {
                    "name": "create_order",
                    "description": "Create a new food order",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "customer_phone": {"type": "string"},
                            "items": {"type": "array"},
                            "total_amount": {"type": "number"}
                        },
                        "required": ["customer_phone", "items"]
                    }
                },
                {
                    "name": "get_customer_history",
                    "description": "Get customer order history",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "phone_number": {"type": "string"}
                        },
                        "required": ["phone_number"]
                    }
                },
                {
                    "name": "update_order_status",
                    "description": "Update order status",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "order_id": {"type": "string"},
                            "status": {"type": "string"}
                        },
                        "required": ["order_id", "status"]
                    }
                }
            ]
        }
    
    async def call_tool(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Execute tool calls"""
        tool_name = params.get("name")
        arguments = params.get("arguments", {})
        
        try:
            if tool_name == "get_menu":
                return await self.get_menu(arguments)
            elif tool_name == "create_order":
                return await self.create_order(arguments)
            elif tool_name == "get_customer_history":
                return await self.get_customer_history(arguments)
            elif tool_name == "update_order_status":
                return await self.update_order_status(arguments)
            else:
                return {"error": f"Unknown tool: {tool_name}"}
        except Exception as e:
            return {"error": f"Tool execution failed: {str(e)}"}
    
    async def get_menu(self, args: Dict[str, Any]) -> Dict[str, Any]:
        """Get menu items from database"""
        category = args.get("category")
        
        query = {}
        if category:
            query["category"] = category
            
        menu_items = await self.db.menu.find(query).to_list(length=None)
        
        return {
            "content": [
                {
                    "type": "text",
                    "text": f"Found {len(menu_items)} menu items:\n" + 
                           "\n".join([f"• {item['name']} - Rs.{item['price']}" 
                                    for item in menu_items])
                }
            ]
        }
    
    async def create_order(self, args: Dict[str, Any]) -> Dict[str, Any]:
        """Create new order"""
        order_data = {
            "customer_phone": args["customer_phone"],
            "items": args["items"],
            "total_amount": args.get("total_amount", 0),
            "status": "pending",
            "created_at": datetime.utcnow(),
            "updated_at": datetime.utcnow()
        }
        
        result = await self.db.orders.insert_one(order_data)
        
        return {
            "content": [
                {
                    "type": "text", 
                    "text": f"Order created successfully! Order ID: {str(result.inserted_id)}"
                }
            ]
        }
    
    async def get_customer_history(self, args: Dict[str, Any]) -> Dict[str, Any]:
        """Get customer order history"""
        phone_number = args["phone_number"]
        
        orders = await self.db.orders.find(
            {"customer_phone": phone_number}
        ).sort("created_at", -1).limit(10).to_list(length=None)
        
        history_text = f"Customer {phone_number} order history:\n"
        for order in orders:
            history_text += f"• Order {str(order['_id'])[:8]} - Rs.{order.get('total_amount', 0)} - {order.get('status', 'unknown')}\n"
        
        return {
            "content": [
                {
                    "type": "text",
                    "text": history_text
                }
            ]
        }
    
    async def list_resources(self) -> Dict[str, Any]:
        """List available resources"""
        return {
            "resources": [
                {
                    "uri": "restaurant://menu",
                    "name": "Restaurant Menu",
                    "description": "Current menu items and prices"
                },
                {
                    "uri": "restaurant://orders",
                    "name": "Order Management", 
                    "description": "Order tracking and management"
                },
                {
                    "uri": "restaurant://customers",
                    "name": "Customer Database",
                    "description": "Customer information and history"
                }
            ]
        }
    
    async def read_resource(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Read resource content"""
        uri = params.get("uri")
        
        if uri == "restaurant://menu":
            menu_items = await self.db.menu.find({}).to_list(length=None)
            return {
                "contents": [
                    {
                        "uri": uri,
                        "mimeType": "application/json",
                        "text": json.dumps(menu_items, default=str, indent=2)
                    }
                ]
            }
        else:
            return {"error": f"Unknown resource: {uri}"}

async def main():
    """Main MCP server loop"""
    server = RestaurantMCPServer()
    await server.initialize()
    
    print("🚀 Restaurant MCP Server started", file=sys.stderr)
    
    async for line in sys.stdin:
        try:
            request = json.loads(line.strip())
            response = await server.handle_request(request)
            print(json.dumps(response))
            sys.stdout.flush()
        except Exception as e:
            error_response = {"error": f"Request processing failed: {str(e)}"}
            print(json.dumps(error_response))
            sys.stdout.flush()

if __name__ == "__main__":
    asyncio.run(main())

5. Start MCP Server

# In a separate terminal
python mcp_server.py

6. Configure Claude Desktop (Optional)

Add to Claude Desktop config (%APPDATA%\Claude\claude_desktop_config.json):

{
  "mcpServers": {
    "restaurant-bot": {
      "command": "python",
      "args": ["C:\\path\\to\\your\\project\\mcp_server.py"],
      "env": {
        "MONGODB_URL": "mongodb://localhost:27017"
      }
    }
  }
}

7. Integration with FastAPI Bot

Update your llm_processor.py to use MCP:

# Add MCP client integration
import json
import subprocess

class LLMProcessor:
    def __init__(self, mongodb, conversation_manager):
        # ... existing code ...
        self.mcp_client = MCPClient()
    
    async def process_with_mcp(self, message: str, context: dict):
        """Process message using MCP server"""
        # Call MCP server for enhanced processing
        mcp_response = await self.mcp_client.call_tool(
            "get_customer_history", 
            {"phone_number": context.get("phone_number")}
        )
        
        # Combine with regular LLM processing
        enhanced_context = f"{context}\nMCP Data: {mcp_response}"
        return await self.process_message_as_chatbot(message, enhanced_context)

🛠️ Required Services

Prerequisites

• Python 3.8+
• Node.js 16+ (for MCP server)
• MongoDB running on localhost:27017
• WhatsApp Desktop installed and logged in
• Google AI API key from https://ai.google.dev/

Service Dependencies

# MongoDB
# Download from: https://www.mongodb.com/try/download/community

# WhatsApp Desktop  
# Download from: https://www.whatsapp.com/download

# Google AI Studio
# Get API key: https://ai.google.dev/

🔧 Configuration Files

.env Configuration

# Google AI API Key
GOOGLE_API_KEY=your_google_api_key_here

# MongoDB Connection
MONGODB_URL=mongodb://localhost:27017
DATABASE_NAME=restaurant_db

# WhatsApp Database Path
WHATSAPP_SQLITE_DB_PATH=C:\Users\YourUser\AppData\Roaming\WhatsApp\Databases\msgstore.db

# WhatsApp API
WHATSAPP_API_URL=http://localhost:3000

# MCP Server Settings
MCP_SERVER_HOST=localhost
MCP_SERVER_PORT=3001

# AI Model Settings
MODEL_NAME=models/gemini-2.0-flash
TEMPERATURE=0.7

🚀 Running the Complete System

1. Start Core Services

# Terminal 1: MongoDB
mongod

# Terminal 2: MCP Server  
python mcp_server.py

# Terminal 3: FastAPI Bot
uvicorn main:app --reload --host 0.0.0.0 --port 8000

2. Verify Setup

# Check FastAPI
curl http://localhost:8000/health

# Check MCP Server
echo '{"method": "tools/list", "params": {}}' | python mcp_server.py

📱 Features

Core Bot Features

• 🤖 AI-powered responses in Roman Urdu
• 📋 Order management and tracking
• 🍽️ Dynamic menu handling
• 💬 Customer conversation history
• 📞 WhatsApp integration

MCP Enhanced Features

• 🔍 Advanced context awareness
• 🛠️ Real-time database operations
• 📊 Customer analytics
• 🔄 Multi-model AI coordination
• 🚀 Extensible tool ecosystem

🔌 API Endpoints

GET  /                     # Health check
GET  /health              # System status
POST /api/whatsapp/send   # Send WhatsApp message
GET  /api/menu           # Get menu items
POST /api/orders         # Create order
GET  /api/orders/{id}    # Get order details

🐛 Troubleshooting

Common Issues

1. WhatsApp DB not found: Update WHATSAPP_SQLITE_DB_PATH in .env
2. MongoDB connection failed: Ensure MongoDB is running
3. MCP server not responding: Check MCP server logs
4. Google API errors: Verify API key and quotas

Debug Commands

# Check WhatsApp DB
sqlite3 "path/to/msgstore.db" "SELECT COUNT(*) FROM messages;"

# Test MongoDB connection
python -c "from motor.motor_asyncio import AsyncIOMotorClient; print('MongoDB OK')"

# Test MCP server
echo '{"method": "tools/list"}' | python mcp_server.py

📚 Documentation

• FastAPI Docs
• MCP Protocol
• Google AI API
• MongoDB Motor

🤝 Contributing

1. Fork the repository
2. Create feature branch (git checkout -b feature/amazing-feature)
3. Commit changes (git commit -m 'Add amazing feature')
4. Push to branch (git push origin feature/amazing-feature)
5. Open Pull Request

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

💡 Quick Start (TL;DR)

# Clone and setup
git clone <repo-url> && cd restaurant-whatsapp-bot
python -m venv fastapi_env && fastapi_env\Scripts\activate
pip install -r requirements.txt
cp .env.example .env  # Edit with your values

# Start services
mongod  # Terminal 1
python mcp_server.py  # Terminal 2  
uvicorn main:app --reload  # Terminal 3

# Test
curl http://localhost:8000/health

🎉 Your Restaurant WhatsApp Bot with MCP integration is ready!