feat: Update registry main application and server service

Author: dheerajoruganty

registry/main.py

@@ -9,10 +9,12 @@
 
 import logging
 from contextlib import asynccontextmanager
+from typing import Annotated
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Cookie
 from fastapi.staticfiles import StaticFiles
 from fastapi.templating import Jinja2Templates
+from fastapi.responses import RedirectResponse
 
 # Import domain routers
 from registry.auth.routes import router as auth_router
@@ -145,6 +147,9 @@ async def health_check():
     return {"status": "healthy", "service": "mcp-gateway-registry"}
 
 
+
+
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run(

registry/services/server_service.py

@@ -221,6 +221,79 @@ def get_all_servers(self) -> Dict[str, Dict[str, Any]]:
         """Get all registered servers."""
         return self.registered_servers.copy()
 
+    def get_filtered_servers(self, accessible_servers: List[str]) -> Dict[str, Dict[str, Any]]:
+        """
+        Get servers filtered by user's accessible servers list.
+        
+        Args:
+            accessible_servers: List of server names the user can access
+            
+        Returns:
+            Dict of servers the user is authorized to see
+        """
+        if not accessible_servers:
+            logger.debug("User has no accessible servers, returning empty dict")
+            return {}
+        
+        logger.info(f"DEBUG: get_filtered_servers called with accessible_servers: {accessible_servers}")
+        logger.info(f"DEBUG: Available registered servers paths: {list(self.registered_servers.keys())}")
+        
+        filtered_servers = {}
+        for path, server_info in self.registered_servers.items():
+            server_name = server_info.get("server_name", "")
+            # Extract technical name from path (remove leading slash)
+            technical_name = path.lstrip('/')
+            logger.info(f"DEBUG: Checking server path='{path}', server_name='{server_name}', technical_name='{technical_name}' against accessible_servers")
+            
+            # Check if user has access to this server using technical name
+            if technical_name in accessible_servers:
+                filtered_servers[path] = server_info
+                logger.info(f"DEBUG: ✓ User has access to server: {technical_name} ({server_name})")
+            else:
+                logger.info(f"DEBUG: ✗ User does not have access to server: {technical_name} ({server_name})")
+        
+        logger.info(f"Filtered {len(filtered_servers)} servers from {len(self.registered_servers)} total servers")
+        return filtered_servers
+
+    def get_all_servers_with_permissions(self, accessible_servers: Optional[List[str]] = None) -> Dict[str, Dict[str, Any]]:
+        """
+        Get servers with optional filtering based on user permissions.
+        
+        Args:
+            accessible_servers: Optional list of server names the user can access.
+                               If None, returns all servers (admin access).
+                               
+        Returns:
+            Dict of servers the user is authorized to see
+        """
+        if accessible_servers is None:
+            # Admin access - return all servers
+            logger.debug("Admin access - returning all servers")
+            return self.get_all_servers()
+        else:
+            # Filtered access - return only accessible servers
+            logger.debug(f"Filtered access - returning servers accessible to user: {accessible_servers}")
+            return self.get_filtered_servers(accessible_servers)
+
+    def user_can_access_server_path(self, path: str, accessible_servers: List[str]) -> bool:
+        """
+        Check if user can access a specific server by path.
+        
+        Args:
+            path: Server path to check
+            accessible_servers: List of server names the user can access
+            
+        Returns:
+            True if user can access the server, False otherwise
+        """
+        server_info = self.get_server_info(path)
+        if not server_info:
+            return False
+            
+        # Extract technical name from path (remove leading slash)
+        technical_name = path.lstrip('/')
+        return technical_name in accessible_servers
+
     def is_service_enabled(self, path: str) -> bool:
         """Check if a service is enabled."""
         result = self.service_state.get(path, False)