AI-powered WhatsApp bot for food ordering using FastAPI, Google Gemini, and MCP (Model Context Protocol) server integration.
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ WhatsApp │────│ FastAPI Bot │────│ Google AI │
│ Database │ │ (MCP Client) │ │ (Gemini) │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│
┌──────────────────┐
│ MCP Server │
│ (Claude/Tools) │
└──────────────────┘
git clone https://github.com/yourusername/restaurant-whatsapp-bot.git
cd restaurant-whatsapp-botpython -m venv fastapi_env
fastapi_env\Scripts\activate # Windows
# or
source fastapi_env/bin/activate # Linux/Macpip install -r requirements.txtcp .env.example .env
# Edit .env with your actual valuesThe 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
# Install MCP server dependencies
npm install -g @modelcontextprotocol/server-claude
# or
pip install mcp-server-claudeCreate mcp-config.json:
{
"mcpServers": {
"restaurant-bot": {
"command": "python",
"args": ["mcp_server.py"],
"env": {
"MONGODB_URL": "mongodb://localhost:27017",
"DATABASE_NAME": "restaurant_db"
}
}
}
}# Create mcp_server.py in your project root
touch mcp_server.pyAdd 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())# In a separate terminal
python mcp_server.pyAdd 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"
}
}
}
}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)- 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/
# 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/# 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# 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# Check FastAPI
curl http://localhost:8000/health
# Check MCP Server
echo '{"method": "tools/list", "params": {}}' | python mcp_server.py- 🤖 AI-powered responses in Roman Urdu
- 📋 Order management and tracking
- 🍽️ Dynamic menu handling
- 💬 Customer conversation history
- 📞 WhatsApp integration
- 🔍 Advanced context awareness
- 🛠️ Real-time database operations
- 📊 Customer analytics
- 🔄 Multi-model AI coordination
- 🚀 Extensible tool ecosystem
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
- WhatsApp DB not found: Update
WHATSAPP_SQLITE_DB_PATHin.env - MongoDB connection failed: Ensure MongoDB is running
- MCP server not responding: Check MCP server logs
- Google API errors: Verify API key and quotas
# 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- Fork the repository
- Create feature branch (
git checkout -b feature/amazing-feature) - Commit changes (
git commit -m 'Add amazing feature') - Push to branch (
git push origin feature/amazing-feature) - Open Pull Request
This project is licensed under the MIT License - see the LICENSE file for details.
# 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!