MCP Server Template

""" Resource Registration System Provides centralized registration and management of MCP resources using a registry pattern for dynamic component discovery. """ import logging from typing import List, Type, Dict, Any from fastmcp import FastMCP from .base import BaseResource from .system import SystemResources from ..config.logging import get_logger logger = get_logger(__name__) class ResourceRegistry: """ Central registry for managing MCP resources. Provides functionality to register resource classes and automatically register all resources with a FastMCP instance. """ def __init__(self): """Initialize the resource registry.""" self._resource_classes: List[Type[BaseResource]] = [] self._registered_resources: Dict[str, BaseResource] = {} def register(self, resource_class: Type[BaseResource]) -> None: """ Register a resource class with the registry. Args: resource_class: The resource class to register """ if not issubclass(resource_class, BaseResource): raise TypeError(f"Resource class {resource_class.__name__} must inherit from BaseResource") self._resource_classes.append(resource_class) logger.info(f"Registered resource class: {resource_class.__name__}") def register_all(self, mcp: FastMCP) -> None: """ Register all resources with the FastMCP instance. Args: mcp: The FastMCP instance to register resources with """ logger.info(f"Registering {len(self._resource_classes)} resource classes") for resource_class in self._resource_classes: try: # Instantiate the resource resource_instance = resource_class() # Register with MCP resource_instance.register_with_mcp(mcp) # Store the instance for potential future use self._registered_resources[resource_instance.name] = resource_instance logger.info(f"Successfully registered resource: {resource_instance.name}") except Exception as e: logger.error(f"Failed to register resource {resource_class.__name__}: {e}") # Continue with other resources even if one fails def get_registered_resources(self) -> Dict[str, BaseResource]: """ Get all registered resource instances. Returns: Dictionary mapping resource names to instances """ return self._registered_resources.copy() def get_resource_info(self) -> List[Dict[str, Any]]: """ Get information about all registered resources. Returns: List of dictionaries containing resource metadata """ return [resource.get_info() for resource in self._registered_resources.values()] # Global registry instance registry = ResourceRegistry() # Register built-in resources registry.register(SystemResources) def register_all_resources(mcp: FastMCP) -> None: """ Register all resources with the MCP server. This is the main entry point for resource registration that should be called from the server initialization. Args: mcp: The FastMCP instance to register resources with """ registry.register_all(mcp) def get_resource_registry() -> ResourceRegistry: """ Get the global resource registry instance. Returns: The global resource registry """ return registry def register_resource_class(resource_class: Type[BaseResource]) -> None: """ Convenience function to register a new resource class. Args: resource_class: The resource class to register """ registry.register(resource_class)