The MCP plugin system allows developers to extend MCP server functionality without modifying core code. Here are detailed plugin architecture and implementation methods:
Plugin Architecture Design
MCP plugin system should consider following aspects:
- Plugin Discovery: Automatically discover and load plugins
- Plugin Lifecycle: Manage plugin loading, initialization, and unloading
- Plugin Isolation: Ensure plugins are isolated from each other
- Plugin Communication: Provide communication mechanisms between plugins
- Plugin Security: Restrict plugin permissions and resource access
1. Plugin Interface Definition
from abc import ABC, abstractmethod from typing import Dict, Any, List class MCPPlugin(ABC): """MCP plugin base class""" def __init__(self, config: Dict[str, Any] = None): self.config = config or {} self.name = self.__class__.__name__ self.version = getattr(self, 'VERSION', '1.0.0') @abstractmethod async def initialize(self, server): """Initialize plugin""" pass @abstractmethod async def shutdown(self): """Shutdown plugin""" pass @abstractmethod def get_tools(self) -> List[Dict[str, Any]]: """Get tools provided by plugin""" return [] @abstractmethod def get_resources(self) -> List[Dict[str, Any]]: """Get resources provided by plugin""" return [] @abstractmethod def get_prompts(self) -> List[Dict[str, Any]]: """Get prompts provided by plugin""" return [] def get_metadata(self) -> Dict[str, Any]: """Get plugin metadata""" return { "name": self.name, "version": self.version, "description": getattr(self, 'DESCRIPTION', ''), "author": getattr(self, 'AUTHOR', ''), "dependencies": getattr(self, 'DEPENDENCIES', []) }
2. Plugin Manager
import importlib import inspect import os from pathlib import Path from typing import Dict, List, Type class PluginManager: def __init__(self, server): self.server = server self.plugins: Dict[str, MCPPlugin] = {} self.plugin_directories: List[str] = [] def add_plugin_directory(self, directory: str): """Add plugin directory""" if directory not in self.plugin_directories: self.plugin_directories.append(directory) async def discover_plugins(self): """Discover plugins""" discovered_plugins = [] for directory in self.plugin_directories: plugin_path = Path(directory) if not plugin_path.exists(): continue # Traverse plugin directory for item in plugin_path.iterdir(): if item.is_file() and item.suffix == '.py': # Single file plugin discovered_plugins.append(str(item)) elif item.is_dir() and (item / '__init__.py').exists(): # Package plugin discovered_plugins.append(str(item)) return discovered_plugins async def load_plugin(self, plugin_path: str) -> bool: """Load plugin""" try: # Dynamically import plugin module spec = importlib.util.spec_from_file_location( "plugin_module", plugin_path ) if spec is None or spec.loader is None: return False module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) # Find plugin class plugin_class = None for name, obj in inspect.getmembers(module): if (inspect.isclass(obj) and issubclass(obj, MCPPlugin) and obj is not MCPPlugin): plugin_class = obj break if plugin_class is None: return False # Create plugin instance plugin = plugin_class() # Initialize plugin await plugin.initialize(self.server) # Register plugin self.plugins[plugin.name] = plugin # Register tools, resources, prompts provided by plugin self._register_plugin_tools(plugin) self._register_plugin_resources(plugin) self._register_plugin_prompts(plugin) return True except Exception as e: print(f"Failed to load plugin {plugin_path}: {e}") return False def _register_plugin_tools(self, plugin: MCPPlugin): """Register plugin tools""" tools = plugin.get_tools() for tool_info in tools: @self.server.tool( name=tool_info["name"], description=tool_info["description"] ) async def tool_wrapper(**kwargs): return await tool_info["function"](**kwargs) def _register_plugin_resources(self, plugin: MCPPlugin): """Register plugin resources""" resources = plugin.get_resources() for resource_info in resources: @self.server.resource( uri=resource_info["uri"], name=resource_info["name"], description=resource_info["description"] ) async def resource_wrapper(): return await resource_info["function"]() def _register_plugin_prompts(self, plugin: MCPPlugin): """Register plugin prompts""" prompts = plugin.get_prompts() for prompt_info in prompts: @self.server.prompt( name=prompt_info["name"], description=prompt_info["description"] ) async def prompt_wrapper(**kwargs): return await prompt_info["function"](**kwargs) async def unload_plugin(self, plugin_name: str) -> bool: """Unload plugin""" if plugin_name not in self.plugins: return False plugin = self.plugins[plugin_name] try: # Shutdown plugin await plugin.shutdown() # Remove from plugin list del self.plugins[plugin_name] return True except Exception as e: print(f"Failed to unload plugin {plugin_name}: {e}") return False async def reload_plugin(self, plugin_name: str) -> bool: """Reload plugin""" if plugin_name not in self.plugins: return False # Unload first await self.unload_plugin(plugin_name) # Reload (need to save plugin path) # Simplified here, actual implementation needs to save plugin path return True def get_plugin_info(self, plugin_name: str) -> Dict[str, Any]: """Get plugin information""" if plugin_name not in self.plugins: return {} plugin = self.plugins[plugin_name] return plugin.get_metadata() def list_plugins(self) -> List[Dict[str, Any]]: """List all plugins""" return [ plugin.get_metadata() for plugin in self.plugins.values() ]
3. Example Plugin Implementation
# plugins/database_plugin.py class DatabasePlugin(MCPPlugin): """Database plugin""" VERSION = "1.0.0" DESCRIPTION = "Provides database query and management functionality" AUTHOR = "Your Name" DEPENDENCIES = ["sqlalchemy"] def __init__(self, config: Dict[str, Any] = None): super().__init__(config) self.db_connection = None async def initialize(self, server): """Initialize database connection""" from sqlalchemy import create_engine db_url = self.config.get("database_url", "sqlite:///mcp.db") self.db_connection = create_engine(db_url) print(f"Database plugin {self.name} initialized") async def shutdown(self): """Close database connection""" if self.db_connection: self.db_connection.dispose() print(f"Database plugin {self.name} shutdown") def get_tools(self) -> List[Dict[str, Any]]: """Get database tools""" return [ { "name": "query_database", "description": "Execute database query", "function": self._query_database }, { "name": "execute_sql", "description": "Execute SQL statement", "function": self._execute_sql } ] def get_resources(self) -> List[Dict[str, Any]]: """Get database resources""" return [ { "uri": "db://schema", "name": "Database Schema", "description": "Database table structure", "function": self._get_schema } ] def get_prompts(self) -> List[Dict[str, Any]]: """Get database prompts""" return [ { "name": "generate_query", "description": "Generate SQL query prompt", "function": self._generate_query_prompt } ] async def _query_database(self, query: str) -> str: """Execute database query""" from sqlalchemy import text with self.db_connection.connect() as conn: result = conn.execute(text(query)) rows = result.fetchall() return f"Query results: {len(rows)} rows" async def _execute_sql(self, sql: str) -> str: """Execute SQL statement""" from sqlalchemy import text with self.db_connection.connect() as conn: conn.execute(text(sql)) conn.commit() return "SQL executed successfully" async def _get_schema(self) -> str: """Get database schema""" from sqlalchemy import inspect inspector = inspect(self.db_connection) tables = inspector.get_table_names() return f"Database tables: {', '.join(tables)}" async def _generate_query_prompt(self, table_name: str) -> str: """Generate query prompt""" return f""" Please generate SQL query for table {table_name}. Requirements: 1. Only use SELECT queries 2. Include appropriate WHERE conditions 3. Add LIMIT to restrict result count """
4. Plugin Configuration
import json from pathlib import Path class PluginConfig: def __init__(self, config_dir: str = "config/plugins"): self.config_dir = Path(config_dir) self.config_dir.mkdir(parents=True, exist_ok=True) def load_config(self, plugin_name: str) -> Dict[str, Any]: """Load plugin configuration""" config_file = self.config_dir / f"{plugin_name}.json" if not config_file.exists(): return {} with open(config_file, 'r') as f: return json.load(f) def save_config( self, plugin_name: str, config: Dict[str, Any] ): """Save plugin configuration""" config_file = self.config_dir / f"{plugin_name}.json" with open(config_file, 'w') as f: json.dump(config, f, indent=2) def get_all_configs(self) -> Dict[str, Dict[str, Any]]: """Get all plugin configurations""" configs = {} for config_file in self.config_dir.glob("*.json"): plugin_name = config_file.stem with open(config_file, 'r') as f: configs[plugin_name] = json.load(f) return configs
5. Plugin Dependency Management
from typing import Dict, List, Set class PluginDependencyManager: def __init__(self): self.dependencies: Dict[str, Set[str]] = {} self.dependents: Dict[str, Set[str]] = {} def add_dependency(self, plugin_name: str, dependency: str): """Add dependency relationship""" if plugin_name not in self.dependencies: self.dependencies[plugin_name] = set() self.dependencies[plugin_name].add(dependency) if dependency not in self.dependents: self.dependents[dependency] = set() self.dependents[dependency].add(plugin_name) def get_load_order(self, plugins: List[str]) -> List[str]: """Get plugin load order (topological sort)""" visited = set() result = [] def visit(plugin_name: str): if plugin_name in visited: return visited.add(plugin_name) # Load dependencies first for dep in self.dependencies.get(plugin_name, []): visit(dep) result.append(plugin_name) for plugin_name in plugins: visit(plugin_name) return result def check_circular_dependency(self) -> bool: """Check circular dependency""" visited = set() recursion_stack = set() def has_cycle(plugin_name: str) -> bool: visited.add(plugin_name) recursion_stack.add(plugin_name) for dep in self.dependencies.get(plugin_name, []): if dep not in visited: if has_cycle(dep): return True elif dep in recursion_stack: return True recursion_stack.remove(plugin_name) return False for plugin_name in self.dependencies: if plugin_name not in visited: if has_cycle(plugin_name): return True return False
6. Plugin Sandbox
import sys from typing import Any class PluginSandbox: def __init__(self): self.restricted_modules = { 'os', 'sys', 'subprocess', 'socket', 'pickle', 'shelve', 'marshal' } def create_sandbox(self, plugin_name: str) -> dict: """Create plugin sandbox environment""" safe_globals = { '__builtins__': self._create_safe_builtins(), '__name__': f'plugin_{plugin_name}', } return safe_globals def _create_safe_builtins(self) -> dict: """Create safe built-in functions""" safe_builtins = {} # Allowed safe built-in functions allowed_builtins = [ 'abs', 'all', 'any', 'bool', 'dict', 'enumerate', 'filter', 'float', 'int', 'len', 'list', 'map', 'max', 'min', 'range', 'set', 'sorted', 'str', 'sum', 'tuple', 'type', 'zip' ] for name in allowed_builtins: if hasattr(__builtins__, name): safe_builtins[name] = getattr(__builtins__, name) return safe_builtins def execute_in_sandbox( self, code: str, sandbox_env: dict ) -> Any: """Execute code in sandbox""" try: exec(code, sandbox_env) return True except Exception as e: print(f"Sandbox execution failed: {e}") return False
Best Practices:
- Plugin Isolation: Ensure plugins are isolated from each other to avoid conflicts
- Dependency Management: Properly handle plugin dependency relationships
- Error Handling: Plugin errors should not affect the core system
- Version Compatibility: Support plugin version management and compatibility checking
- Security Restrictions: Restrict plugin permissions and resource access
- Complete Documentation: Provide clear documentation for each plugin
Through a comprehensive plugin system, you can flexibly extend MCP server functionality to meet various customization needs.