[Feat] MCP Gateway - Allow setting MCP Servers as Private/Public available on Internet

CLAUDE.md

@@ -90,6 +90,7 @@ LiteLLM is a unified interface for 100+ LLM providers with two main components:
 - Pydantic v2 for data validation
 - Async/await patterns throughout
 - Type hints required for all public APIs
+- **Avoid imports within methods** — place all imports at the top of the file (module-level). Inline imports inside functions/methods make dependencies harder to trace and hurt readability. The only exception is avoiding circular imports where absolutely necessary.
 
 ### Testing Strategy
 - Unit tests in `tests/test_litellm/`

litellm/proxy/_experimental/mcp_server/auth/litellm_auth_handler.py

@@ -27,6 +27,7 @@ def __init__(
         oauth2_headers: Optional[Dict[str, str]] = None,
         mcp_protocol_version: Optional[str] = None,
         raw_headers: Optional[Dict[str, str]] = None,
+        client_ip: Optional[str] = None,
     ):
         self.user_api_key_auth = user_api_key_auth
         self.mcp_auth_header = mcp_auth_header
@@ -35,3 +36,4 @@ def __init__(
         self.mcp_protocol_version = mcp_protocol_version
         self.oauth2_headers = oauth2_headers
         self.raw_headers = raw_headers
+        self.client_ip = client_ip

litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py

@@ -9,13 +9,14 @@
     get_async_httpx_client,
     httpxSpecialProvider,
 )
+from litellm.proxy.auth.ip_address_utils import IPAddressUtils
 from litellm.proxy.common_utils.encrypt_decrypt_utils import (
     decrypt_value_helper,
     encrypt_value_helper,
 )
 from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
-from litellm.types.mcp_server.mcp_server_manager import MCPServer
 from litellm.proxy.utils import get_server_root_path
+from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
 router = APIRouter(
     tags=["mcp"],
@@ -300,7 +301,10 @@ async def authorize(
     )
 
     lookup_name = mcp_server_name or client_id
-    mcp_server = global_mcp_server_manager.get_mcp_server_by_name(lookup_name)
+    client_ip = IPAddressUtils.get_mcp_client_ip(request)
+    mcp_server = global_mcp_server_manager.get_mcp_server_by_name(
+        lookup_name, client_ip=client_ip
+    )
     if mcp_server is None:
         raise HTTPException(status_code=404, detail="MCP server not found")
     return await authorize_with_server(
@@ -342,7 +346,10 @@ async def token_endpoint(
     )
 
     lookup_name = mcp_server_name or client_id
-    mcp_server = global_mcp_server_manager.get_mcp_server_by_name(lookup_name)
+    client_ip = IPAddressUtils.get_mcp_client_ip(request)
+    mcp_server = global_mcp_server_manager.get_mcp_server_by_name(
+        lookup_name, client_ip=client_ip
+    )
     if mcp_server is None:
         raise HTTPException(status_code=404, detail="MCP server not found")
     return await exchange_token_with_server(
@@ -425,7 +432,10 @@ def _build_oauth_protected_resource_response(
     request_base_url = get_request_base_url(request)
     mcp_server: Optional[MCPServer] = None
     if mcp_server_name:
-        mcp_server = global_mcp_server_manager.get_mcp_server_by_name(mcp_server_name)
+        client_ip = IPAddressUtils.get_mcp_client_ip(request)
+        mcp_server = global_mcp_server_manager.get_mcp_server_by_name(
+            mcp_server_name, client_ip=client_ip
+        )
 
     # Build resource URL based on the pattern
     if mcp_server_name:
@@ -538,7 +548,10 @@ def _build_oauth_authorization_server_response(
 
     mcp_server: Optional[MCPServer] = None
     if mcp_server_name:
-        mcp_server = global_mcp_server_manager.get_mcp_server_by_name(mcp_server_name)
+        client_ip = IPAddressUtils.get_mcp_client_ip(request)
+        mcp_server = global_mcp_server_manager.get_mcp_server_by_name(
+            mcp_server_name, client_ip=client_ip
+        )
 
     return {
         "issuer": request_base_url,  # point to your proxy
@@ -629,7 +642,10 @@ async def register_client(request: Request, mcp_server_name: Optional[str] = Non
     if not mcp_server_name:
         return dummy_return
 
-    mcp_server = global_mcp_server_manager.get_mcp_server_by_name(mcp_server_name)
+    client_ip = IPAddressUtils.get_mcp_client_ip(request)
+    mcp_server = global_mcp_server_manager.get_mcp_server_by_name(
+        mcp_server_name, client_ip=client_ip
+    )
     if mcp_server is None:
         return dummy_return
     return await register_client_with_server(

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -53,6 +53,7 @@
     MCPTransportType,
     UserAPIKeyAuth,
 )
+from litellm.proxy.auth.ip_address_utils import IPAddressUtils
 from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_value_helper
 from litellm.proxy.utils import ProxyLogging
 from litellm.types.llms.custom_http import httpxSpecialProvider
@@ -65,20 +66,22 @@
 from litellm.types.utils import CallTypes
 
 try:
-    from mcp.shared.tool_name_validation import (  # type: ignore
+    from mcp.shared.tool_name_validation import (
+        validate_tool_name,  # pyright: ignore[reportAssignmentType]
+    )
+    from mcp.shared.tool_name_validation import (
         SEP_986_URL,
-        validate_tool_name,
     )
 except ImportError:
     from pydantic import BaseModel
     SEP_986_URL = "https://github.com/modelcontextprotocol/protocol/blob/main/proposals/0001-tool-name-validation.md"
 
-    class ToolNameValidationResult(BaseModel):
+    class _ToolNameValidationResult(BaseModel):
         is_valid: bool = True
         warnings: list = []
 
-    def validate_tool_name(name: str) -> ToolNameValidationResult:  # type: ignore[misc]
-        return ToolNameValidationResult()
+    def validate_tool_name(name: str) -> _ToolNameValidationResult:
+        return _ToolNameValidationResult()
 
 
 # Probe includes characters on both sides of the separator to mimic real prefixed tool names.
@@ -324,6 +327,9 @@ async def load_servers_from_config(
                 access_groups=server_config.get("access_groups", None),
                 static_headers=server_config.get("static_headers", None),
                 allow_all_keys=bool(server_config.get("allow_all_keys", False)),
+                available_on_public_internet=bool(
+                    server_config.get("available_on_public_internet", False)
+                ),
             )
             self.config_mcp_servers[server_id] = new_server
 
@@ -622,6 +628,9 @@ async def build_mcp_server_from_table(
             allowed_tools=getattr(mcp_server, "allowed_tools", None),
             disallowed_tools=getattr(mcp_server, "disallowed_tools", None),
             allow_all_keys=mcp_server.allow_all_keys,
+            available_on_public_internet=bool(
+                getattr(mcp_server, "available_on_public_internet", False)
+            ),
             updated_at=getattr(mcp_server, "updated_at", None),
         )
         return new_server
@@ -696,6 +705,23 @@ async def get_allowed_mcp_servers(
             verbose_logger.warning(f"Failed to get allowed MCP servers: {str(e)}.")
             return allow_all_server_ids
 
+    def filter_server_ids_by_ip(
+        self, server_ids: List[str], client_ip: Optional[str]
+    ) -> List[str]:
+        """
+        Filter server IDs by client IP — external callers only see public servers.
+
+        Returns server_ids unchanged when client_ip is None (no filtering).
+        """
+        if client_ip is None:
+            return server_ids
+        return [
+            sid
+            for sid in server_ids
+            if (s := self.get_mcp_server_by_id(sid)) is not None
+            and self._is_server_accessible_from_ip(s, client_ip)
+        ]
+
     async def get_tools_for_server(self, server_id: str) -> List[MCPTool]:
         """
         Get the tools for a given server
@@ -2200,6 +2226,42 @@ def get_mcp_servers_from_ids(self, server_ids: List[str]) -> List[MCPServer]:
                 servers.append(server)
         return servers
 
+    def _get_general_settings(self) -> Dict[str, Any]:
+        """Get general_settings, importing lazily to avoid circular imports."""
+        try:
+            from litellm.proxy.proxy_server import (
+                general_settings as proxy_general_settings,
+            )
+            return proxy_general_settings
+        except ImportError:
+            # Fallback if proxy_server not available
+            return {}
+
+    def _is_server_accessible_from_ip(
+        self, server: MCPServer, client_ip: Optional[str]
+    ) -> bool:
+        """
+        Check if a server is accessible from the given client IP.
+
+        - If client_ip is None, no IP filtering is applied (internal callers).
+        - If the server has available_on_public_internet=True, it's always accessible.
+        - Otherwise, only internal/private IPs can access it.
+        """
+        if client_ip is None:
+            return True
+        if server.available_on_public_internet:
+            return True
+        # Check backwards compat: litellm.public_mcp_servers
+        public_ids = set(litellm.public_mcp_servers or [])
+        if server.server_id in public_ids:
+            return True
+        # Non-public server: only accessible from internal IPs
+        general_settings = self._get_general_settings()
+        internal_networks = IPAddressUtils.parse_internal_networks(
+            general_settings.get("mcp_internal_ip_ranges")
+        )
+        return IPAddressUtils.is_internal_ip(client_ip, internal_networks)
+
     def get_mcp_server_by_id(self, server_id: str) -> Optional[MCPServer]:
         """
         Get the MCP Server from the server id
@@ -2212,27 +2274,76 @@ def get_mcp_server_by_id(self, server_id: str) -> Optional[MCPServer]:
 
     def get_public_mcp_servers(self) -> List[MCPServer]:
         """
-        Get the public MCP servers
+        Get the public MCP servers (available_on_public_internet=True flag on server).
+        Also includes servers from litellm.public_mcp_servers for backwards compat.
         """
         servers: List[MCPServer] = []
-        if litellm.public_mcp_servers is None:
-            return servers
-        for server_id in litellm.public_mcp_servers:
-            server = self.get_mcp_server_by_id(server_id)
-            if server:
+        public_ids = set(litellm.public_mcp_servers or [])
+        for server in self.get_registry().values():
+            if server.available_on_public_internet or server.server_id in public_ids:
                 servers.append(server)
         return servers
 
-    def get_mcp_server_by_name(self, server_name: str) -> Optional[MCPServer]:
+    def get_mcp_server_by_name(
+        self, server_name: str, client_ip: Optional[str] = None
+    ) -> Optional[MCPServer]:
         """
-        Get the MCP Server from the server name
+        Get the MCP Server from the server name.
+
+        Uses priority-based matching to avoid collisions:
+        1. First pass: exact alias match (highest priority)
+        2. Second pass: exact server_name match
+        3. Third pass: exact name match (lowest priority)
+
+        Args:
+            server_name: The server name to look up.
+            client_ip: Optional client IP for access control. When provided,
+                       non-public servers are hidden from external IPs.
         """
         registry = self.get_registry()
+
+        # Pass 1: Match by alias (highest priority)
+        for server in registry.values():
+            if server.alias == server_name:
+                if not self._is_server_accessible_from_ip(server, client_ip):
+                    return None
+                return server
+
+        # Pass 2: Match by server_name
         for server in registry.values():
             if server.server_name == server_name:
+                if not self._is_server_accessible_from_ip(server, client_ip):
+                    return None
+                return server
+
+        # Pass 3: Match by name (lowest priority)
+        for server in registry.values():
+            if server.name == server_name:
+                if not self._is_server_accessible_from_ip(server, client_ip):
+                    return None
                 return server
+
         return None
 
+    def get_filtered_registry(
+        self, client_ip: Optional[str] = None
+    ) -> Dict[str, MCPServer]:
+        """
+        Get registry filtered by client IP access control.
+
+        Args:
+            client_ip: Optional client IP. When provided, non-public servers
+                       are hidden from external IPs. When None, returns all servers.
+        """
+        registry = self.get_registry()
+        if client_ip is None:
+            return registry
+        return {
+            k: v
+            for k, v in registry.items()
+            if self._is_server_accessible_from_ip(v, client_ip)
+        }
+
     def _generate_stable_server_id(
         self,
         server_name: str,