Class MCPServerManager

optillm/plugins/mcp_plugin.py:501–625

Manages MCP servers and capabilities

            return False

class MCPServerManager:
    """Manages MCP servers and capabilities"""
    
    def __init__(self, config_manager: MCPConfigManager):
        self.config_manager = config_manager
        self.servers: Dict[str, MCPServer] = {}
        self.initialized = False
        
        # Cache of capabilities
        self.all_tools = []
        self.all_resources = []
        self.all_prompts = []
    
    async def initialize(self) -> bool:
        """Initialize and cache all server capabilities"""
        if self.initialized:
            return True
            
        # Create servers
        for server_name, server_config in self.config_manager.servers.items():
            self.servers[server_name] = MCPServer(server_name, server_config)
        
        # Connect to all servers and discover capabilities
        connected_servers = 0
        for server_name, server in self.servers.items():
            success = await server.connect_and_discover()
            if success:
                connected_servers += 1
                # Cache server capabilities
                for tool in server.tools:
                    tool_info = {
                        "server": server_name,
                        "name": tool.name,
                        "description": tool.description,
                        "input_schema": tool.inputSchema
                    }
                    self.all_tools.append(tool_info)
                    logger.debug(f"Cached tool: {tool_info}")
                
                for resource in server.resources:
                    resource_info = {
                        "server": server_name,
                        "uri": resource.uri,
                        "name": resource.name,
                        "description": resource.description
                    }
                    self.all_resources.append(resource_info)
                    logger.debug(f"Cached resource: {resource_info}")
                
                for prompt in server.prompts:
                    prompt_info = {
                        "server": server_name,
                        "name": prompt.name,
                        "description": prompt.description,
                        "arguments": prompt.arguments
                    }
                    self.all_prompts.append(prompt_info)
                    logger.debug(f"Cached prompt: {prompt_info}")