"""
Absher Voice Assistant MCP Server
Provides government service tools for Saudi Arabia (Absher platform)
Tools:
1. verify_national_id - Check if national ID exists
2. get_security_questions - Fetch security questions for verification
3. verify_security_answers - Verify user's security answers
4. get_user_data - Load user profile information
5. get_service_requirements - Get required fields for a service
6. renew_passport - Submit passport renewal request
7. get_traffic_fines - Query traffic violations
8. book_appointment - Schedule government office appointment
"""
import json
import sys
import logging
from typing import Any, Dict, List
from datetime import datetime
from fastmcp import FastMCP
# Import mock data
from mock_data import (
verify_national_id as check_national_id,
get_security_questions as fetch_security_questions,
verify_security_answers as check_security_answers,
get_user_by_national_id,
get_traffic_fines as fetch_traffic_fines,
get_unpaid_fines,
get_total_unpaid_amount,
get_service_requirements as fetch_service_requirements,
AVAILABLE_APPOINTMENT_SLOTS
)
# ============================================================================
# Configure Logging
# ============================================================================
# Set up logging to stderr (so it doesn't interfere with MCP protocol on stdout)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
stream=sys.stderr
)
logger = logging.getLogger(__name__)
# Initialize FastMCP server
mcp = FastMCP("Absher Voice Assistant")
# ============================================================================
# Authentication & Identity Tools
# ============================================================================
@mcp.tool()
def verify_national_id(national_id: str) -> str:
"""
التحقق من وجود رقم الهوية الوطنية في النظام
Check if a national ID exists in the system
Args:
national_id: رقم الهوية الوطنية (10 digits)
Returns:
JSON string with verification result
"""
logger.info(f"🔍 [verify_national_id] Called with national_id={national_id}")
exists = check_national_id(national_id)
result = {
"exists": exists,
"message": "تم العثور على رقم الهوية الوطنية في النظام" if exists else "رقم الهوية الوطنية غير موجود في النظام"
}
logger.info(f"✅ [verify_national_id] Result: exists={exists}")
return json.dumps(result, ensure_ascii=False)
@mcp.tool()
def get_security_questions(national_id: str) -> str:
"""
جلب الأسئلة الأمنية للمستخدم
Fetch security questions for user verification
Args:
national_id: رقم الهوية الوطنية
Returns:
JSON string with list of security questions
"""
logger.info(f"🔐 [get_security_questions] Called with national_id={national_id}")
questions = fetch_security_questions(national_id)
if not questions:
logger.warning(f"⚠️ [get_security_questions] No questions found for national_id={national_id}")
return json.dumps({
"success": False,
"message": "لم يتم العثور على أسئلة أمنية لهذا المستخدم"
}, ensure_ascii=False)
questions_list = [
{
"question_id": q.question_id,
"question": q.question
}
for q in questions
]
logger.info(f"✅ [get_security_questions] Retrieved {len(questions_list)} questions")
return json.dumps({
"success": True,
"questions": questions_list
}, ensure_ascii=False)
@mcp.tool()
def verify_security_answers(national_id: str, answers: List[Dict[str, str]]) -> str:
"""
التحقق من إجابات الأسئلة الأمنية
Verify user's answers to security questions
Args:
national_id: رقم الهوية الوطنية
answers: List of answer objects [{"question_id": "q1", "answer": "الرياض"}]
Returns:
JSON string with verification result
"""
logger.info(f"🔐 [verify_security_answers] Called with national_id={national_id}, answers_count={len(answers)}")
is_verified = check_security_answers(national_id, answers)
result = {
"verified": is_verified,
"message": "تم التحقق من هويتك بنجاح" if is_verified else "الإجابات غير صحيحة. يرجى المحاولة مرة أخرى"
}
logger.info(f"{'✅' if is_verified else '❌'} [verify_security_answers] Verification {'succeeded' if is_verified else 'failed'}")
return json.dumps(result, ensure_ascii=False)
# ============================================================================
# User Data Tool
# ============================================================================
@mcp.tool()
def get_user_data(national_id: str) -> str:
"""
جلب بيانات المستخدم الشخصية
Load user profile information
Args:
national_id: رقم الهوية الوطنية
Returns:
JSON string with complete user profile
"""
logger.info(f"👤 [get_user_data] Called with national_id={national_id}")
user = get_user_by_national_id(national_id)
if not user:
logger.warning(f"⚠️ [get_user_data] User not found for national_id={national_id}")
return json.dumps({
"success": False,
"message": "لم يتم العثور على بيانات المستخدم"
}, ensure_ascii=False)
user_data = {
"success": True,
"user": {
"national_id": user.national_id,
"first_name": user.personal_data.first_name,
"last_name": user.personal_data.last_name,
"date_of_birth": user.personal_data.date_of_birth,
"phone": user.personal_data.phone,
"email": user.personal_data.email,
"address": user.personal_data.address,
"passport_number": user.personal_data.passport_number,
"passport_issue_date": user.personal_data.passport_issue_date,
"passport_expiry": user.personal_data.passport_expiry
}
}
logger.info(f"✅ [get_user_data] Retrieved data for: {user.personal_data.first_name} {user.personal_data.last_name}")
return json.dumps(user_data, ensure_ascii=False)
# ============================================================================
# Service Discovery Tool
# ============================================================================
@mcp.tool()
def get_service_requirements(service_name: str) -> str:
"""
جلب المتطلبات المطلوبة لخدمة معينة
Get required fields for a specific service
Args:
service_name: اسم الخدمة (passport_renewal, traffic_fines, book_appointment)
Returns:
JSON string with service requirements
"""
logger.info(f"📋 [get_service_requirements] Called with service_name={service_name}")
requirements = fetch_service_requirements(service_name)
if not requirements:
logger.warning(f"⚠️ [get_service_requirements] Service not found: {service_name}")
return json.dumps({
"success": False,
"message": f"الخدمة '{service_name}' غير موجودة"
}, ensure_ascii=False)
logger.info(f"✅ [get_service_requirements] Retrieved requirements for: {requirements['service_name_ar']} ({len(requirements['required_fields'])} fields)")
return json.dumps({
"success": True,
"service_name": service_name,
"service_name_ar": requirements["service_name_ar"],
"required_fields": requirements["required_fields"]
}, ensure_ascii=False)
# ============================================================================
# Service Execution Tools
# ============================================================================
@mcp.tool()
def renew_passport(
national_id: str,
renewal_reason: str,
delivery_method: str
) -> str:
"""
تقديم طلب تجديد جواز السفر
Submit passport renewal request
Args:
national_id: رقم الهوية الوطنية
renewal_reason: سبب التجديد (انتهاء الصلاحية، تلف، ضياع، امتلاء الصفحات)
delivery_method: طريقة الاستلام (استلام من المكتب، توصيل بالبريد)
Returns:
JSON string with submission result
"""
logger.info(f"🛂 [renew_passport] Called with national_id={national_id}, renewal_reason={renewal_reason}, delivery_method={delivery_method}")
# Verify user exists
user = get_user_by_national_id(national_id)
if not user:
logger.warning(f"⚠️ [renew_passport] User not found: {national_id}")
return json.dumps({
"success": False,
"message": "المستخدم غير موجود"
}, ensure_ascii=False)
# Validate renewal_reason
valid_reasons = ["انتهاء الصلاحية", "تلف", "ضياع", "امتلاء الصفحات"]
if renewal_reason not in valid_reasons:
logger.warning(f"⚠️ [renew_passport] Invalid renewal reason: {renewal_reason}")
return json.dumps({
"success": False,
"message": f"سبب التجديد غير صحيح. الخيارات المتاحة: {', '.join(valid_reasons)}"
}, ensure_ascii=False)
# Validate delivery_method
valid_methods = ["استلام من المكتب", "توصيل بالبريد"]
if delivery_method not in valid_methods:
logger.warning(f"⚠️ [renew_passport] Invalid delivery method: {delivery_method}")
return json.dumps({
"success": False,
"message": f"طريقة الاستلام غير صحيحة. الخيارات المتاحة: {', '.join(valid_methods)}"
}, ensure_ascii=False)
# Generate request ID
request_id = f"PR{datetime.now().strftime('%Y%m%d%H%M%S')}"
logger.info(f"✅ [renew_passport] Successfully created request {request_id} for: {user.personal_data.first_name} {user.personal_data.last_name}")
return json.dumps({
"success": True,
"request_id": request_id,
"message": f"تم تقديم طلب تجديد جواز السفر بنجاح. رقم الطلب: {request_id}. سيتم معالجة الطلب خلال 5 أيام عمل.",
"details": {
"national_id": national_id,
"name": f"{user.personal_data.first_name} {user.personal_data.last_name}",
"passport_number": user.personal_data.passport_number,
"renewal_reason": renewal_reason,
"delivery_method": delivery_method,
"processing_time": "5 أيام عمل"
}
}, ensure_ascii=False)
@mcp.tool()
def get_traffic_fines(national_id: str) -> str:
"""
الاستعلام عن المخالفات المرورية
Query traffic violations for a user
Args:
national_id: رقم الهوية الوطنية
Returns:
JSON string with list of traffic fines
"""
logger.info(f"🚗 [get_traffic_fines] Called with national_id={national_id}")
# Verify user exists
user = get_user_by_national_id(national_id)
if not user:
logger.warning(f"⚠️ [get_traffic_fines] User not found: {national_id}")
return json.dumps({
"success": False,
"message": "المستخدم غير موجود"
}, ensure_ascii=False)
# Get all fines
all_fines = fetch_traffic_fines(national_id)
unpaid = get_unpaid_fines(national_id)
total_unpaid = get_total_unpaid_amount(national_id)
if not all_fines:
logger.info(f"✅ [get_traffic_fines] No fines found for: {user.personal_data.first_name} {user.personal_data.last_name}")
return json.dumps({
"success": True,
"message": "لا توجد مخالفات مرورية مسجلة",
"fines": [],
"summary": {
"total_fines": 0,
"unpaid_fines": 0,
"total_unpaid_amount": 0
}
}, ensure_ascii=False)
logger.info(f"✅ [get_traffic_fines] Found {len(all_fines)} fines ({len(unpaid)} unpaid, {total_unpaid} SAR) for: {user.personal_data.first_name} {user.personal_data.last_name}")
return json.dumps({
"success": True,
"message": f"تم العثور على {len(all_fines)} مخالفة مرورية",
"fines": all_fines,
"summary": {
"total_fines": len(all_fines),
"unpaid_fines": len(unpaid),
"total_unpaid_amount": total_unpaid
}
}, ensure_ascii=False)
@mcp.tool()
def book_appointment(
national_id: str,
service_type: str,
location: str,
preferred_date: str
) -> str:
"""
حجز موعد في مكتب حكومي
Schedule appointment at government office
Args:
national_id: رقم الهوية الوطنية
service_type: نوع الخدمة (الجوازات، الأحوال المدنية، المرور)
location: الموقع (الرياض، جدة، الدمام)
preferred_date: التاريخ المفضل (YYYY-MM-DD format)
Returns:
JSON string with appointment confirmation
"""
logger.info(f"📅 [book_appointment] Called with national_id={national_id}, service_type={service_type}, location={location}, date={preferred_date}")
# Verify user exists
user = get_user_by_national_id(national_id)
if not user:
logger.warning(f"⚠️ [book_appointment] User not found: {national_id}")
return json.dumps({
"success": False,
"message": "المستخدم غير موجود"
}, ensure_ascii=False)
# Map service types
service_map = {
"الجوازات": "passports",
"الأحوال المدنية": "civil_affairs",
"المرور": "traffic"
}
if service_type not in service_map:
logger.warning(f"⚠️ [book_appointment] Invalid service type: {service_type}")
return json.dumps({
"success": False,
"message": f"نوع الخدمة غير صحيح. الخيارات المتاحة: {', '.join(service_map.keys())}"
}, ensure_ascii=False)
# Validate location
valid_locations = ["الرياض", "جدة", "الدمام"]
if location not in valid_locations:
logger.warning(f"⚠️ [book_appointment] Invalid location: {location}")
return json.dumps({
"success": False,
"message": f"الموقع غير صحيح. الخيارات المتاحة: {', '.join(valid_locations)}"
}, ensure_ascii=False)
# Get available slots
service_key = service_map[service_type]
available_slots = AVAILABLE_APPOINTMENT_SLOTS.get(service_key, {}).get(location, [])
if not available_slots:
logger.warning(f"⚠️ [book_appointment] No slots available for {service_type} in {location}")
return json.dumps({
"success": False,
"message": f"لا توجد مواعيد متاحة في {location} لخدمة {service_type}"
}, ensure_ascii=False)
# Find matching slot (simplified - just pick first available on that date)
matching_slot = None
for slot in available_slots:
if slot.startswith(preferred_date):
matching_slot = slot
break
if not matching_slot:
logger.warning(f"⚠️ [book_appointment] No slots on date {preferred_date}. Available: {available_slots}")
return json.dumps({
"success": False,
"message": f"لا توجد مواعيد متاحة في التاريخ {preferred_date}",
"available_slots": available_slots
}, ensure_ascii=False)
# Generate appointment ID
appointment_id = f"APT{datetime.now().strftime('%Y%m%d%H%M%S')}"
logger.info(f"✅ [book_appointment] Created appointment {appointment_id} for: {user.personal_data.first_name} {user.personal_data.last_name} at {matching_slot}")
return json.dumps({
"success": True,
"appointment_id": appointment_id,
"message": f"تم حجز الموعد بنجاح. رقم الحجز: {appointment_id}",
"details": {
"national_id": national_id,
"name": f"{user.personal_data.first_name} {user.personal_data.last_name}",
"service_type": service_type,
"location": location,
"appointment_time": matching_slot,
"confirmation_code": appointment_id
}
}, ensure_ascii=False)
# ============================================================================
# Run Server
# ============================================================================
if __name__ == "__main__":
import sys
import os
# Show startup message to stderr (won't interfere with MCP protocol on stdout)
print("🚀 Absher MCP Server starting...", file=sys.stderr)
print("✅ Server ready! Waiting for MCP client connection...", file=sys.stderr)
print(" (Press Ctrl+C to stop)", file=sys.stderr)
print("", file=sys.stderr)
# Get port from environment variable (used by hosting platforms like Hostinger)
port = int(os.environ.get("PORT", 8080))
# Start the MCP server with HTTP transport for remote access
# - transport="streamable-http": Uses HTTP for communication with MCP clients
# - host="0.0.0.0": Accepts connections from any IP (needed for remote deployment)
# - port: The port to listen on (from environment or default 8080)
# - log_level="debug": Enables detailed logging for development and troubleshooting
mcp.run(transport="streamable-http", host="0.0.0.0", port=port, log_level="debug")
