"""
Bible MCP Server
This demonstrates how to create a complete MCP server by extending
BaseMCPServer from the mcp-weather core infrastructure.
"""
import logging
import sys
from typing import List, Optional, Any
from fastapi import FastAPI, APIRouter, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from core.server import BaseMCPServer
from core.auth_mcp import create_auth_provider
from core.cache import RedisCacheClient
from mcp_bible.config import AppConfig, get_config
from mcp_bible.bible_service import BibleService
from mcp_bible.service import BibleMCPService
logger = logging.getLogger(__name__)
class BibleMCPServer(BaseMCPServer):
"""
Bible MCP Server implementation
Extends BaseMCPServer to provide Bible functionality via MCP protocol.
This shows how to implement all required abstract methods.
"""
@property
def service_title(self) -> str:
"""Title shown in API documentation"""
return "Bible MCP Server"
@property
def service_description(self) -> str:
"""Description shown in API documentation"""
return (
"Provides Bible passage retrieval capabilities via Model Context Protocol (MCP). "
"Supports multiple Bible translations with automatic parsing and cleaning. "
"Built using mcp-weather core infrastructure."
)
@property
def service_version(self) -> str:
"""Service version"""
return "1.0.0"
@property
def allowed_cors_origins(self) -> List[str]:
"""CORS origins for web clients"""
# Use CORS origins from configuration
if hasattr(self.config, "server") and hasattr(self.config.server, "cors_origins"):
return self.config.server.cors_origins
# Fallback to default values if not configured
return [
"http://localhost:3000",
"http://localhost:8080"
]
def create_auth_provider(self) -> Optional[Any]:
"""
Create authentication provider for this server
Returns None to disable auth, or create an Authentik provider
"""
# Check if authentication is enabled using flattened config
auth_enabled = self.get_config("server.auth_enabled", self.get_config("auth_enabled", True))
if not auth_enabled:
logger.info("Authentication is disabled")
return None
if not self.config.authentik:
logger.warning("Authentication enabled but no Authentik config found")
return None
logger.info("Creating Authentik authentication provider")
return create_auth_provider("bible")
def create_router(self) -> APIRouter:
"""
Create REST API router with endpoints
Uses feature-based organization - each feature has its own routes.py
This demonstrates automatic feature route discovery and composition.
"""
import importlib
import pkgutil
from pathlib import Path
# Create main router
router = APIRouter()
# Access the bible service through the MCP service wrapper
bible_service = self.service.bible_service
# Add base routes (health, info)
@router.get("/health")
async def health():
"""Health check endpoint"""
return {
"status": "healthy",
"service": "bible",
"version": self.service_version
}
@router.get("/info")
async def info():
"""Service information endpoint"""
return {
"title": self.service_title,
"description": self.service_description,
"version": self.service_version,
"supported_versions": self.config.bible_api.supported_versions
}
# Discover and include feature routes
features_package = "mcp_bible.features"
features_path = Path(__file__).parent / "features"
if features_path.exists():
for _, feature_name, is_pkg in pkgutil.iter_modules([str(features_path)]):
if not is_pkg:
continue
try:
# Import the routes module from the feature
routes_module_name = f"{features_package}.{feature_name}.routes"
routes_module = importlib.import_module(routes_module_name)
# Check if the module has a create_router function
if hasattr(routes_module, "create_router"):
# Create and include the feature router
feature_router = routes_module.create_router(bible_service)
router.include_router(feature_router)
logger.info(f"✓ Included routes from feature: {feature_name}")
except ModuleNotFoundError:
# Feature doesn't have routes.py, skip it
logger.debug(f"Feature '{feature_name}' has no routes.py, skipping")
except Exception as e:
logger.error(f"Error including routes from feature '{feature_name}': {e}")
return router
def register_exception_handlers(self, app: FastAPI) -> None:
"""
Register custom exception handlers
This demonstrates how to add service-specific error handling.
"""
from fastapi import Request
from fastapi.responses import JSONResponse
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
"""Handle validation errors"""
return JSONResponse(
status_code=400,
content={
"error": "Validation Error",
"detail": str(exc)
}
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""Handle unexpected errors"""
logger.error(f"Unhandled exception: {exc}", exc_info=True)
return JSONResponse(
status_code=500,
content={
"error": "Internal Server Error",
"detail": "An unexpected error occurred"
}
)
def main():
"""
Main entry point for the Bible server
This shows the complete initialization flow:
1. Load configuration
2. Create services (cache, bible, MCP wrapper)
3. Create server
4. Run server
"""
try:
# Load configuration from environment
logger.info("Loading configuration...")
config = get_config()
# Create Redis cache client
logger.info("Initializing Redis cache...")
cache_client = RedisCacheClient(config.redis_cache)
# Create bible service (business logic)
logger.info("Initializing Bible service...")
bible_service = BibleService(
timeout=config.bible_api.timeout
)
# Create MCP service wrapper
logger.info("Creating MCP service wrapper...")
mcp_service = BibleMCPService(bible_service)
# Create and run server
logger.info("Creating server...")
server = BibleMCPServer(config, mcp_service)
logger.info("Starting server...")
server.run()
except KeyboardInterrupt:
logger.info("\nShutdown requested by user")
sys.exit(0)
except Exception as e:
logger.error(f"Fatal error: {e}", exc_info=True)
sys.exit(1)
if __name__ == "__main__":
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
main()
