main.py•4.71 kB
"""Main entry point for KYC MCP Server."""
import asyncio
import signal
from pathlib import Path
from config.settings import get_settings
from src.auth.jwt_manager import JWTManager
from src.cache.redis_cache import CacheManager
from src.clients.kyc_api_client import KYCAPIClient
from src.registry.tool_registry import ToolRegistry
from src.server.mcp_server import KYCMCPServer
from src.tools.pan_aadhaar_link import PANAadhaarLinkTool
from src.tools.pan_verification import PANVerificationTool
from src.utils.logger import configure_logging, get_logger
from src.utils.rate_limiter import RateLimiter
logger = get_logger(__name__)
class Application:
"""Main application class."""
def __init__(self):
"""Initialize application."""
self.settings = get_settings()
self.cache: CacheManager = None
self.api_client: KYCAPIClient = None
self.server: KYCMCPServer = None
async def initialize(self) -> None:
"""Initialize all components."""
logger.info("initializing_application", version="1.0.0")
# Configure logging
configure_logging(self.settings.LOG_LEVEL)
# Initialize cache manager
if self.settings.CACHE_ENABLED:
self.cache = CacheManager(
host=self.settings.REDIS_HOST,
port=self.settings.REDIS_PORT,
db=self.settings.REDIS_DB,
password=self.settings.REDIS_PASSWORD,
ssl=self.settings.REDIS_SSL,
)
await self.cache.connect()
logger.info("cache_manager_initialized")
else:
logger.info("cache_disabled")
# Initialize JWT manager
jwt_manager = JWTManager(
secret_key=self.settings.KYC_JWT_SECRET,
algorithm=self.settings.KYC_JWT_ALGORITHM,
)
# Generate JWT token for API authentication
jwt_token = jwt_manager.generate_token(
payload={"service": "kyc-mcp-server"},
expires_in=self.settings.KYC_JWT_EXPIRY,
)
# Initialize API client
self.api_client = KYCAPIClient(
base_url=self.settings.KYC_API_BASE_URL,
api_key=self.settings.KYC_API_KEY,
jwt_token=jwt_token,
timeout=self.settings.REQUEST_TIMEOUT,
)
logger.info("api_client_initialized")
# Initialize rate limiter
rate_limiter = RateLimiter(
requests_per_minute=self.settings.RATE_LIMIT_PER_MINUTE,
requests_per_hour=self.settings.RATE_LIMIT_PER_HOUR,
)
logger.info("rate_limiter_initialized")
# Initialize tool registry
metadata_dir = Path(__file__).parent.parent / "metadata" / "tools"
tool_registry = ToolRegistry(metadata_dir=metadata_dir)
metadata_count = tool_registry.initialize()
logger.info("tool_registry_initialized", metadata_count=metadata_count)
# Register tools
pan_verification_tool = PANVerificationTool(api_client=self.api_client)
tool_registry.register_tool(pan_verification_tool)
pan_aadhaar_link_tool = PANAadhaarLinkTool(api_client=self.api_client)
tool_registry.register_tool(pan_aadhaar_link_tool)
logger.info("tools_registered", count=len(tool_registry.tools))
# Initialize MCP server
self.server = KYCMCPServer(
tool_registry=tool_registry,
cache_manager=self.cache,
rate_limiter=rate_limiter,
)
logger.info("mcp_server_initialized")
async def run(self) -> None:
"""Run the MCP server."""
logger.info("starting_mcp_server")
await self.server.run()
async def shutdown(self) -> None:
"""Graceful shutdown."""
logger.info("shutting_down_application")
if self.cache:
await self.cache.close()
if self.api_client:
await self.api_client.close()
logger.info("application_shutdown_complete")
async def main():
"""Main entry point."""
app = Application()
# Setup signal handlers for graceful shutdown
loop = asyncio.get_event_loop()
def signal_handler(sig):
logger.info("received_signal", signal=sig)
asyncio.create_task(app.shutdown())
loop.stop()
for sig in (signal.SIGTERM, signal.SIGINT):
loop.add_signal_handler(sig, lambda s=sig: signal_handler(s))
try:
await app.initialize()
await app.run()
except Exception as e:
logger.error("application_error", error=str(e), exc_info=True)
raise
finally:
await app.shutdown()
if __name__ == "__main__":
asyncio.run(main())