"""
Main Server Entry Point
Clean, modular server implementation that uses the registry pattern
to automatically discover and register all components.
"""
import asyncio
import logging
from fastmcp import FastMCP
from .config.settings import get_settings
from .config.logging import setup_logging, get_logger
from .tools.registry import register_all_tools
from .resources.registry import register_all_resources
from .prompts.registry import register_all_prompts
logger = get_logger(__name__)
def create_server() -> FastMCP:
"""
Factory function to create a configured MCP server.
This function:
1. Loads configuration
2. Sets up logging
3. Creates FastMCP instance
4. Registers all components using registry pattern
Returns:
Configured FastMCP server instance
"""
# Load configuration
settings = get_settings()
# Setup logging
setup_logging(settings.log_level)
logger.info(f"π Creating MCP server: {settings.server_name}")
logger.info(f"Environment: {settings.environment}")
# Create FastMCP instance
mcp = FastMCP(settings.server_name)
# Register all components using registry pattern
logger.info("Registering components...")
try:
register_all_tools(mcp)
register_all_resources(mcp)
register_all_prompts(mcp)
logger.info("β
All components registered successfully")
except Exception as e:
logger.error(f"β Failed to register components: {e}")
raise
return mcp
async def main():
"""
Main server execution function.
Creates the server and runs it with the configured transport and settings.
"""
settings = get_settings()
logger.info("=" * 50)
logger.info(f"π Starting MCP Server: {settings.server_name}")
logger.info(f"Version: {settings.version}")
logger.info(f"Environment: {settings.environment}")
logger.info(f"Transport: {settings.transport}")
logger.info(f"Host: {settings.host}")
logger.info(f"Port: {settings.port}")
logger.info("=" * 50)
# Create the server
mcp = create_server()
try:
# Run the server
await mcp.run_async(
transport=settings.transport,
host=settings.host,
port=settings.port
)
except KeyboardInterrupt:
logger.info("π Server stopped by user")
except Exception as e:
logger.error(f"β Server error: {e}")
raise
finally:
logger.info("π Server shutdown complete")
def run_server():
"""
Synchronous entry point for running the server.
This function can be used for programmatic server startup.
"""
asyncio.run(main())
if __name__ == "__main__":
run_server()