app.py•4.23 kB
"""
Main application that initializes and runs the MCP server.
"""
from typing import Optional
from .core.config_loader import Settings, load_config
from .core.logger import configure_logging, get_logger
from .core.server import MCPServer
from .prompts.example import EXAMPLE_PROMPT_SCHEMA, example_prompt_handler
from .registry.manager import RegistryManager
from .resources.config import CONFIG_RESOURCE_SCHEMA, config_resource_handler
from .resources.status import STATUS_RESOURCE_SCHEMA, status_resource_handler
from .tools.calculator import CALCULATOR_SCHEMA, calculator_handler
from .tools.search import SEARCH_SCHEMA, search_handler
class MCPTemplateApp:
"""
Main application class that orchestrates the MCP server lifecycle.
"""
def __init__(self, settings: Settings):
"""
Initialize the application.
Args:
settings: Application settings
"""
self.settings = settings
# Configure logging
configure_logging(
level=settings.logging.level,
format=settings.logging.format,
log_file=settings.logging.file,
)
self.logger = get_logger(__name__)
# Initialize components
self.server = MCPServer(settings)
self.registry_manager: Optional[RegistryManager] = None
if settings.registry.enabled:
self.registry_manager = RegistryManager(settings)
# Register tools, resources, and prompts
self._register_components()
def _register_components(self) -> None:
"""Register all tools, resources, and prompts with the server."""
self.logger.info("Registering tools, resources, and prompts")
# Register tools
self.server.register_tool(
"example_calculator",
calculator_handler,
CALCULATOR_SCHEMA,
)
self.server.register_tool(
"example_search",
search_handler,
SEARCH_SCHEMA,
)
# Register resources (wrap handlers to pass settings)
async def config_handler(uri: str) -> str:
return await config_resource_handler(uri, self.settings)
self.server.register_resource(
"example_config",
config_handler,
CONFIG_RESOURCE_SCHEMA,
)
self.server.register_resource(
"example_status",
status_resource_handler,
STATUS_RESOURCE_SCHEMA,
)
# Register prompts
self.server.register_prompt(
"example_prompt",
example_prompt_handler,
EXAMPLE_PROMPT_SCHEMA,
)
self.logger.info("All components registered successfully")
async def run(self) -> None:
"""
Run the application.
This method starts the registry manager (if enabled) and the MCP server.
"""
self.logger.info(
"Starting MCP Template application",
version=self.settings.server.version,
)
try:
# Register with registry if enabled
if self.registry_manager:
async with self.registry_manager:
# Run the server
await self.server.run()
else:
# Run without registry
await self.server.run()
except KeyboardInterrupt:
self.logger.info("Received keyboard interrupt")
except Exception as e:
self.logger.error("Application error", error=str(e), exc_info=True)
raise
finally:
await self.cleanup()
async def cleanup(self) -> None:
"""Cleanup resources on shutdown."""
self.logger.info("Cleaning up resources")
# Additional cleanup can be added here
self.logger.info("Cleanup complete")
def create_app(settings: Optional[Settings] = None) -> MCPTemplateApp:
"""
Factory function to create the application.
Args:
settings: Optional settings. If not provided, will load from config.
Returns:
MCPTemplateApp instance
"""
if settings is None:
settings = load_config()
return MCPTemplateApp(settings)