Customer Flow: User sends natural language message → Bot extracts structured order → Saves to database Admin Flow: Admin receives formatted order → Checks payment reference → Adds waybill → Marks for Pudo drop-off
Example User Message:
"Hi, I want 1 case for Ntombi Witbooi, phone 0662997702, MetLife mall kwt, KWT, Eastern Cape"
System Generates for Admin Dashboard:
Most simple order form for team SA
*Order form*
Names: Ntombi Witbooi
Order (ID): OR 33914904395
Stock to collect: 1 case
Cell nr: 0662997702
*Pudo Locker name:* MetLife mall kwt
*Your suburb* KWT
Eastern Cape
Available Stock Options:
- 24 bottles R180
- 36 bottles R220
- 24 bottles plus 20 sprays R180
- 20 sprays R150
- 48 bottles R360
Admin Actions: Verify payment reference → Add waybill number → Drop at Pudo
Multi-Channel → Channel Adapters → LLM Provider → MCP Tools → Database → Admin Dashboard
Pluggable Components:
- Channels: Telegram (MVP), WhatsApp, Website, Facebook (future)
- LLM Providers: Gemini (default), OpenAI, DeepSeek, Ollama (swappable)
- MCP Tools: Product lookup, validation, fallback extraction
- Schema Versions: v1 (current), v2+ (future evolution)
order-chatbot-mvp/
├── backend/ # FastAPI app
│ ├── main.py # Main API with Telegram webhook
│ ├── models.py # Database models
│ ├── telegram_handler.py # Telegram message processing
│ ├── llm_service.py # Gemini integration with MCP fallback
│ ├── admin_api.py # Admin endpoints
│ └── database.py # DB connection
├── admin-dashboard/ # Simple React admin
│ ├── pages/
│ │ ├── orders.tsx # Order list/management
│ │ └── dashboard.tsx # Simple overview
│ └── components/
├── mcp-tools/ # Simple MCP server for reliability
│ └── order_tools.py # Product lookup and fallback extraction
├── docker-compose.yml # All services
├── .env.example
└── README.md
- FastAPI: Simple API with Telegram webhook
- SQLAlchemy: Basic ORM
- PostgreSQL: Database
- Gemini 1.5: LLM with MCP fallback tools
- Pydantic: Data validation
- Next.js: Simple admin dashboard
- Tailwind CSS: Basic styling
- React Query: Data fetching
- Docker Compose: Local development
- Redis: Simple caching (optional)
-- Just 3 tables for MVP
CREATE TABLE customers (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name VARCHAR(255) NOT NULL,
phone VARCHAR(20),
telegram_user_id VARCHAR(50) UNIQUE,
created_at TIMESTAMP DEFAULT NOW()
);
CREATE TABLE orders (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
customer_id UUID REFERENCES customers(id),
order_display_id VARCHAR(20) UNIQUE NOT NULL, -- OR33914904395 format
status VARCHAR(50) NOT NULL DEFAULT 'pending', -- pending|paid|packed|delivered
-- Order details
customer_name VARCHAR(255) NOT NULL,
phone VARCHAR(20),
items_text TEXT NOT NULL, -- JSON of items
locker_name VARCHAR(255) NOT NULL,
locker_address TEXT NOT NULL, -- "suburb, province"
-- Pricing
total_amount DECIMAL(10,2) NOT NULL,
-- Payment (admin fills this)
payment_reference VARCHAR(255),
payment_verified BOOLEAN DEFAULT FALSE,
payment_verified_at TIMESTAMP,
payment_verified_by VARCHAR(255),
-- Fulfillment (admin fills this)
waybill_number VARCHAR(255),
waybill_added_at TIMESTAMP,
waybill_added_by VARCHAR(255),
pudo_drop_date DATE,
-- Generated order form for admin
formatted_order_form TEXT,
-- Meta
raw_telegram_message JSONB,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
CREATE TABLE products (
sku VARCHAR(100) PRIMARY KEY,
name VARCHAR(255) NOT NULL,
price DECIMAL(10,2) NOT NULL,
aliases TEXT[] DEFAULT '{}' -- ["1 case", "24 bottles"]
);INSERT INTO products (sku, name, price, aliases) VALUES
('24-bottles', '24 Bottles Case', 180.00, ARRAY['1 case', '24 bottles', 'one case']),
('36-bottles', '36 Bottles', 220.00, ARRAY['36 bottles']),
('20-sprays', '20 Spray Bottles', 150.00, ARRAY['20 sprays', '20 spray bottles']),
('48-bottles', '48 Bottles', 360.00, ARRAY['48 bottles', '2 cases']);# mcp-tools/order_tools.py
from mcp import Server
import json
import re
class SimpleOrderMCP:
def __init__(self, db):
self.db = db
self.server = Server("simple-order-tools")
@self.server.call_tool()
async def get_products(self, arguments: dict):
"""Get current product list with prices"""
products = await self.db.fetch_all("SELECT * FROM products WHERE sku = ANY($1)",
arguments.get('skus', []))
return [{
"sku": p["sku"],
"name": p["name"],
"price": float(p["price"]),
"aliases": p["aliases"]
} for p in products]
@self.server.call_tool()
async def extract_order_simple(self, arguments: dict):
"""Simple regex-based order extraction for fallback"""
text = arguments["text"]
# Map common phrases to SKUs
items = []
text_lower = text.lower()
if any(phrase in text_lower for phrase in ['1 case', '24 bottles', 'case']):
items.append({"sku": "24-bottles", "qty": 1})
if '36 bottles' in text_lower:
items.append({"sku": "36-bottles", "qty": 1})
if any(phrase in text_lower for phrase in ['20 sprays', '20 spray']):
items.append({"sku": "20-sprays", "qty": 1})
if any(phrase in text_lower for phrase in ['24 bottles plus 20 sprays', '24 bottles + 20 sprays']):
items.append({"sku": "24-bottles-20-sprays", "qty": 1})
if any(phrase in text_lower for phrase in ['48 bottles', '2 cases', 'two cases']):
items.append({"sku": "48-bottles", "qty": 1})
# Extract name (look for capitalized words)
name_matches = re.findall(r'\b([A-Z][a-z]+(?: [A-Z][a-z]+)*)', text)
name = name_matches[0] if name_matches else "Unknown Customer"
# Extract phone (SA format)
phone_match = re.search(r'(\+?27\d{9}|0\d{9})', text.replace(' ', ''))
phone = phone_match.group(1) if phone_match else None
# Extract location (look for common Pudo locations)
location_patterns = [
r'([\w\s]+(?:mall|shopping|centre|pick n pay|checkers|woolworths)[\w\s]*)',
r'([\w\s]+)\s*,?\s*(johannesburg|cape town|durban|pretoria|port elizabeth)',
]
locker_name = "Not specified"
suburb = "Not specified"
province = "Not specified"
for pattern in location_patterns:
match = re.search(pattern, text, re.IGNORECASE)
if match:
locker_name = match.group(1).strip()
if len(match.groups()) > 1:
suburb = match.group(2).strip().title()
break
# Detect province from common city names
if any(city in text_lower for city in ['johannesburg', 'pretoria', 'sandton']):
province = "Gauteng"
elif any(city in text_lower for city in ['cape town', 'stellenbosch']):
province = "Western Cape"
elif 'durban' in text_lower:
province = "KwaZulu-Natal"
elif any(city in text_lower for city in ['port elizabeth', 'east london']):
province = "Eastern Cape"
return {
"customer_name": name,
"phone": phone,
"items": items,
"locker": {
"name": locker_name,
"suburb": suburb,
"province": province
},
"confidence": 0.7 if items else 0.2
}# backend/telegram_handler.py
from telegram import Update
from telegram.ext import Application, MessageHandler, filters
import json
class SimpleTelegramHandler:
def __init__(self, llm_service, db):
self.llm_service = llm_service
self.db = db
async def handle_message(self, update: Update, context):
"""Process incoming Telegram message"""
message = update.message
user_id = str(message.from_user.id)
text = message.text
try:
# Extract order using LLM + MCP fallback
order_data = await self.llm_service.extract_order(text, user_id)
if not order_data.get("items"):
await message.reply_text(
"I couldn't understand your order. Please include:\n"
"- Your name\n"
"- Items (1 case, 36 bottles, 20 sprays, etc.)\n"
"- Phone number\n"
"- Pudo locker location\n\n"
"Example: 'Hi, John Smith here, I want 1 case, phone 0821234567, Pick n Pay Rosebank, Johannesburg'"
)
return
# Generate order ID (OR + timestamp)
order_id_short = f"OR{int(time.time())}"
# Calculate total
total = await self._calculate_total(order_data["items"])
# Save to database
full_order_id = await self._save_order(order_data, user_id, text, total, order_id_short)
# Generate formatted order form for admin
order_form = self._generate_order_form(order_data, order_id_short, total)
# Send confirmation to user
await message.reply_text(
f"✅ Order received!\n\n"
f"{order_form}\n\n"
f"Please transfer R{total} to our banking details and send us the reference number."
)
except Exception as e:
await message.reply_text(
"Sorry, something went wrong. Please try again or contact support."
)
print(f"Error processing message: {e}")
def _generate_order_form(self, order_data, order_id, total):
"""Generate the formatted order form for admin processing"""
items_text = ", ".join([f"{item['qty']}x {item['sku'].replace('-', ' ')}" for item in order_data['items']])
return f"""*Order form*
Names: {order_data['customer_name']}
Order (ID): {order_id}
Stock to collect: {items_text}
Cell nr: {order_data.get('phone', 'Not provided')}
*Pudo Locker name:* {order_data.get('locker', {}).get('name', 'Not specified')}
*Your suburb:* {order_data.get('locker', {}).get('suburb', 'Not specified')}
{order_data.get('locker', {}).get('province', 'Not specified')}
Total: R{total}"""
async def _calculate_total(self, items):
total = 0
for item in items:
product = await self.db.fetch_one(
"SELECT price FROM products WHERE sku = $1", item["sku"]
)
if product:
total += float(product["price"]) * item["qty"]
return total
async def _save_order(self, order_data, user_id, raw_text, total, order_id_short):
# Upsert customer
customer_id = await self.db.fetch_val(
"""INSERT INTO customers (telegram_user_id, name, phone)
VALUES ($1, $2, $3)
ON CONFLICT (telegram_user_id) DO UPDATE SET
name = EXCLUDED.name, phone = EXCLUDED.phone
RETURNING id""",
user_id, order_data["customer_name"], order_data.get("phone")
)
# Insert order with formatted display
order_id = await self.db.fetch_val(
"""INSERT INTO orders (customer_id, order_display_id, customer_name, phone, items_text,
locker_name, locker_address, total_amount, raw_telegram_message,
formatted_order_form)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING id""",
customer_id,
order_id_short,
order_data["customer_name"],
order_data.get("phone"),
json.dumps(order_data["items"]),
order_data.get("locker", {}).get("name", "TBD"),
f"{order_data.get('locker', {}).get('suburb', 'TBD')}, {order_data.get('locker', {}).get('province', 'TBD')}",
total,
{"text": raw_text, "user_id": user_id},
self._generate_order_form(order_data, order_id_short, total)
)
return order_id# backend/llm_service.py
import google.generativeai as genai
from mcp_client import MCPClient
class SimpleLLMService:
def __init__(self):
self.gemini = genai.GenerativeModel("gemini-1.5-pro")
self.mcp_client = MCPClient("http://localhost:8080")
async def extract_order(self, text: str, user_id: str):
"""Extract order with Gemini + MCP fallback"""
# Get current products from MCP
try:
products = await self.mcp_client.call_tool("get_products", {})
except:
products = [] # Use empty if MCP fails
prompt = f"""
Extract order details from this message: "{text}"
Available products with prices:
- 24 bottles: R180 (aliases: "1 case", "24 bottles", "case")
- 36 bottles: R220 (aliases: "36 bottles")
- 24 bottles plus 20 sprays: R180 (aliases: "24 bottles plus 20 sprays", "24 bottles + 20 sprays")
- 20 sprays: R150 (aliases: "20 sprays", "20 spray bottles")
- 48 bottles: R360 (aliases: "48 bottles", "2 cases")
Extract these fields:
- customer_name: Full name of person
- phone: Phone number (convert to standard format)
- items: Array of {{sku, qty}} matching products above
- locker: {{name, suburb, province}} for Pudo locker location
Example input: "Hi, John Smith here, I want 1 case, phone 0821234567, Pick n Pay Rosebank, Johannesburg, Gauteng"
Example output: {{
"customer_name": "John Smith",
"phone": "0821234567",
"items": [{{"sku": "24-bottles", "qty": 1}}],
"locker": {{"name": "Pick n Pay Rosebank", "suburb": "Johannesburg", "province": "Gauteng"}}
}}
Return ONLY valid JSON, no other text.
"""
try:
# Try Gemini first
response = await self.gemini.generate_content_async(prompt)
result = json.loads(response.text.strip())
result["confidence"] = 0.9
return result
except Exception as e:
print(f"Gemini failed: {e}, using MCP fallback")
# Fallback to MCP regex extraction
try:
result = await self.mcp_client.call_tool("extract_order_simple", {"text": text})
return result
except:
# Last resort
return {
"customer_name": "Unknown",
"phone": None,
"items": [],
"locker": {"name": "TBD", "address": "TBD"},
"confidence": 0.1
}// admin-dashboard/pages/orders.tsx
import { useState, useEffect } from 'react'
export default function Orders() {
const [orders, setOrders] = useState([])
const [selectedOrder, setSelectedOrder] = useState(null)
useEffect(() => {
fetchOrders()
}, [])
const fetchOrders = async () => {
const response = await fetch('/api/admin/orders')
const data = await response.json()
setOrders(data)
}
const verifyPayment = async (orderId, paymentRef) => {
await fetch(`/api/admin/orders/${orderId}/verify-payment`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ payment_reference: paymentRef })
})
fetchOrders()
}
const addWaybill = async (orderId, waybillNumber) => {
await fetch(`/api/admin/orders/${orderId}/waybill`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ waybill_number: waybillNumber })
})
fetchOrders()
}
return (
<div className="p-6">
<h1 className="text-2xl font-bold mb-6">Order Management Dashboard</h1>
{/* Stock Reference */}
<div className="mb-6 p-4 bg-blue-50 rounded-lg">
<h2 className="font-semibold mb-2">Available Stock:</h2>
<div className="text-sm space-y-1">
<div>• 24 bottles - R180</div>
<div>• 36 bottles - R220</div>
<div>• 24 bottles plus 20 sprays - R180</div>
<div>• 20 sprays - R150</div>
<div>• 48 bottles - R360</div>
</div>
</div>
<div className="grid gap-6">
{orders.map(order => (
<div key={order.id} className="border-2 p-6 rounded-lg bg-white shadow-sm">
{/* Display the formatted order form */}
<div className="bg-gray-50 p-4 rounded mb-4 font-mono text-sm whitespace-pre-line">
{order.formatted_order_form}
</div>
<div className="flex justify-between items-center">
<div className="flex items-center gap-4">
<span className={`px-3 py-1 rounded-full text-sm font-medium ${
order.status === 'pending' ? 'bg-yellow-100 text-yellow-800' :
order.status === 'paid' ? 'bg-green-100 text-green-800' :
order.status === 'packed' ? 'bg-blue-100 text-blue-800' :
'bg-purple-100 text-purple-800'
}`}>
{order.status.toUpperCase()}
</span>
<span className="text-sm text-gray-500">
{new Date(order.created_at).toLocaleString()}
</span>
</div>
<div className="flex gap-3">
{order.status === 'pending' && (
<button
onClick={() => {
const ref = prompt('Enter bank payment reference:')
if (ref) verifyPayment(order.id, ref)
}}
className="bg-green-600 hover:bg-green-700 text-white px-4 py-2 rounded font-medium"
>
✓ Verify Payment
</button>
)}
{order.status === 'paid' && !order.waybill_number && (
<button
onClick={() => {
const waybill = prompt('Enter Pudo waybill number:')
if (waybill) addWaybill(order.id, waybill)
}}
className="bg-blue-600 hover:bg-blue-700 text-white px-4 py-2 rounded font-medium"
>
📦 Add Waybill
</button>
)}
{order.waybill_number && (
<div className="text-sm bg-gray-100 px-3 py-2 rounded">
<div className="font-medium">Waybill: {order.waybill_number}</div>
<div className="text-gray-600">Ready for Pudo drop-off</div>
</div>
)}
</div>
</div>
</div>
))}
{orders.length === 0 && (
<div className="text-center py-12 text-gray-500">
No orders yet. Orders will appear here when customers place them via Telegram.
</div>
)}
</div>
</div>
) waybill)
}}
className="bg-blue-500 text-white px-3 py-1 rounded text-sm"
>
Add Waybill
</button>
)}
{order.waybill_number && (
<div className="text-sm">
<p>Waybill: {order.waybill_number}</p>
</div>
)}
</div>
</div>
</div>
))}
</div>
</div>
)
}# backend/admin_api.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
router = APIRouter(prefix="/admin")
class PaymentVerification(BaseModel):
payment_reference: str
class WaybillUpdate(BaseModel):
waybill_number: str
@router.get("/orders")
async def get_orders():
"""Get all orders for admin dashboard"""
orders = await db.fetch_all("""
SELECT id, order_display_id, customer_name, phone, items_text, total_amount,
status, payment_reference, payment_verified, waybill_number,
formatted_order_form, created_at, updated_at
FROM orders
ORDER BY created_at DESC
""")
return [dict(order) for order in orders]
@router.post("/orders/{order_id}/verify-payment")
async def verify_payment(order_id: str, payment: PaymentVerification):
"""Admin verifies payment and updates status"""
result = await db.execute("""
UPDATE orders
SET payment_reference = $1, payment_verified = TRUE, status = 'paid',
payment_verified_at = NOW(), payment_verified_by = 'admin',
updated_at = NOW()
WHERE id = $2
""", payment.payment_reference, order_id)
if result == 0:
raise HTTPException(404, "Order not found")
return {"message": "Payment verified", "status": "paid"}
@router.post("/orders/{order_id}/waybill")
async def add_waybill(order_id: str, waybill: WaybillUpdate):
"""Admin adds waybill for Pudo drop-off"""
result = await db.execute("""
UPDATE orders
SET waybill_number = $1, status = 'packed',
waybill_added_at = NOW(), waybill_added_by = 'admin',
updated_at = NOW()
WHERE id = $2 AND payment_verified = TRUE
""", waybill.waybill_number, order_id)
if result == 0:
raise HTTPException(404, "Order not found or payment not verified")
return {"message": "Waybill added", "status": "packed"}
@router.post("/orders/{order_id}/mark-delivered")
async def mark_delivered(order_id: str):
"""Mark order as delivered after Pudo drop-off"""
result = await db.execute("""
UPDATE orders
SET status = 'delivered', pudo_drop_date = CURRENT_DATE, updated_at = NOW()
WHERE id = $1 AND waybill_number IS NOT NULL
""", order_id)
if result == 0:
raise HTTPException(404, "Order not found or no waybill")
return {"message": "Order marked as delivered"}version: '3.8'
services:
postgres:
image: postgres:15
environment:
POSTGRES_DB: orders
POSTGRES_USER: user
POSTGRES_PASSWORD: pass
ports:
- "5432:5432"
backend:
build: ./backend
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://user:pass@postgres/orders
- TELEGRAM_TOKEN=${TELEGRAM_TOKEN}
- GEMINI_API_KEY=${GEMINI_API_KEY}
depends_on:
- postgres
- mcp-server
mcp-server:
build: ./mcp-tools
ports:
- "8080:8080"
depends_on:
- postgres
admin:
build: ./admin-dashboard
ports:
- "3000:3000"
environment:
- NEXT_PUBLIC_API_URL=http://localhost:8000- User sends natural language Telegram message → Structured order extracted
- Bot generates formatted order form like the Ntombi Witbooi example
- Clear confirmation with order ID and total amount
- Intelligent error messages when extraction fails
- View all orders with formatted order forms displayed
- Stock reference panel showing all available products and prices
- One-click payment verification with reference entry
- One-click waybill addition for Pudo processing
- Clear status progression: pending → paid → packed → delivered
- Order search and filtering capabilities
- Multi-Channel Ready: Telegram (active), WhatsApp/Web (pluggable)
- LLM Provider Swappable: Gemini (default), OpenAI/DeepSeek/Ollama ready
- MCP Fallback System: Reliable extraction even when Gemini fails
- Schema Versioning: Support order format evolution
- Event-Driven Architecture: Redis Streams for scalability
- Full Observability: Metrics, logging, tracing
- Admin Security: JWT authentication and role-based access
- 99%+ Order Success Rate: MCP ensures no orders lost
- <2s Response Time: Fast webhook processing
- Automatic Fallbacks: Multiple extraction methods
- Audit Trail: Complete order lifecycle tracking
- Horizontal Scaling: Redis Streams + worker architecture
# 1. Clone and setup
git clone <repository>
cd nlp-order-chatbot
cp .env.example .env
# Add TELEGRAM_TOKEN, GEMINI_API_KEY, and other config
# 2. Start all services (API, Workers, MCP Server, Admin UI, DB, Redis)
make dev-up
# or: docker-compose up -d
# 3. Run migrations and seed data
make migrate
make seed
# 4. Set up Telegram webhook with MCP verification
python scripts/setup_telegram_webhook.py
python scripts/verify_mcp_tools.py
# 5. Test the system
python scripts/test_order_flow.py
# 6. Access services
# - API with Swagger docs: http://localhost:8000/docs
# - Admin Dashboard: http://localhost:3000
# - MCP Server status: http://localhost:8080/health
# - Grafana monitoring: http://localhost:3001
# - Prometheus metrics: http://localhost:90901. Customer sends message:
"Hi there! I'm Ntombi Witbooi, I need 1 case please. My number is 0662997702.
I want to collect at MetLife mall kwt, KWT, Eastern Cape"
2. System extracts and generates:
Most simple order form for team SA
*Order form*
Names: Ntombi Witbooi
Order (ID): OR 33914904395
Stock to collect: 1 case (24 bottles)
Cell nr: 0662997702
*Pudo Locker name:* MetLife mall kwt
*Your suburb* KWT
Eastern Cape
Total: R180
3. Admin dashboard shows:
- Formatted order form (exactly as above)
- Stock reference panel (24 bottles R180, etc.)
- "✓ Verify Payment" button → enters bank reference
- "📦 Add Waybill" button → enters Pudo waybill number
- Status tracking through delivery
Key Features Maintained:
- Full pluggability for channels, LLMs, schemas
- MCP reliability layer preventing order loss
- Production-ready architecture with monitoring
- Simple admin workflow focused on your exact needs
Build this as a complete production system with the simple MVP admin workflow you specified, but keep all the advanced pluggable architecture for future scaling.