fix(mcp): add public_mcp_hub_strict_whitelist flag (default True) for migration

This commit is contained in:
michelligabriele 2026-05-12 22:05:15 +02:00
parent 6969fa5e98
commit 02817bc92e
No known key found for this signature in database
4 changed files with 73 additions and 12 deletions

View file

@ -426,6 +426,7 @@ disable_copilot_system_to_assistant: bool = (
False # If false (default), converts all 'system' role messages to 'assistant' for GitHub Copilot compatibility. Set to true to disable this behavior.
)
public_mcp_servers: Optional[List[str]] = None
public_mcp_hub_strict_whitelist: bool = True
public_model_groups: Optional[List[str]] = None
public_agent_groups: Optional[List[str]] = None
# Supports both old format (Dict[str, str]) and new format (Dict[str, Dict[str, Any]])

View file

@ -3162,19 +3162,34 @@ class MCPServerManager:
"""
Return the MCP servers published to the AI Hub via /v1/mcp/make_public.
Mirrors /public/model_hub and /public/agent_hub: gates strictly on the
Default (litellm.public_mcp_hub_strict_whitelist=True): mirrors
/public/model_hub and /public/agent_hub — gates strictly on the
litellm.public_mcp_servers whitelist. Returns an empty list when no
servers have been published. The per-server available_on_public_internet
flag is unrelated — it governs IP-based access in
_is_server_accessible_from_ip, not hub visibility.
Legacy (litellm.public_mcp_hub_strict_whitelist=False): preserves the
pre-fix behavior where any server with available_on_public_internet=True
is also included. Intended as a one-release migration window for
deployments that relied on the OR-with-default semantics; will be
removed in a future release.
"""
if litellm.public_mcp_servers is None:
return []
public_ids = set(litellm.public_mcp_servers)
if litellm.public_mcp_hub_strict_whitelist:
if litellm.public_mcp_servers is None:
return []
public_ids = set(litellm.public_mcp_servers)
return [
server
for server in self.get_registry().values()
if server.server_id in public_ids
]
public_ids = set(litellm.public_mcp_servers or [])
return [
server
for server in self.get_registry().values()
if server.server_id in public_ids
if server.available_on_public_internet or server.server_id in public_ids
]
def expand_permission_list(self, identifiers: List[str]) -> List[str]:

View file

@ -3519,5 +3519,57 @@ class TestGetPublicMCPServers:
assert manager.get_public_mcp_servers() == []
class TestGetPublicMCPServersLegacyMode:
"""
Legacy migration knob: litellm.public_mcp_hub_strict_whitelist=False
preserves the pre-fix OR-with-default semantics for one release so
operators that relied on the old behavior have a window to call
/v1/mcp/make_public before /public/mcp_hub goes empty.
"""
def _make_server(self, server_id, available_on_public_internet=True):
return MCPServer(
server_id=server_id,
name=server_id,
server_name=server_id,
transport=MCPTransport.http,
available_on_public_internet=available_on_public_internet,
)
def _make_manager(self, servers):
manager = MCPServerManager()
for s in servers:
manager.config_mcp_servers[s.server_id] = s
return manager
@patch("litellm.public_mcp_hub_strict_whitelist", False)
@patch("litellm.public_mcp_servers", None)
def test_legacy_returns_default_flag_servers_when_whitelist_is_none(self):
"""Legacy mode + no whitelist → every server with the default
available_on_public_internet=True appears (old behavior)."""
manager = self._make_manager(
[
self._make_server("a", available_on_public_internet=True),
self._make_server("b", available_on_public_internet=False),
]
)
result = manager.get_public_mcp_servers()
assert [s.server_id for s in result] == ["a"]
@patch("litellm.public_mcp_hub_strict_whitelist", False)
@patch("litellm.public_mcp_servers", ["b"])
def test_legacy_unions_whitelist_and_default_flag(self):
"""Legacy mode unions the whitelist with any
available_on_public_internet=True server."""
manager = self._make_manager(
[
self._make_server("a", available_on_public_internet=True),
self._make_server("b", available_on_public_internet=False),
]
)
result = manager.get_public_mcp_servers()
assert sorted(s.server_id for s in result) == ["a", "b"]
if __name__ == "__main__":
pytest.main([__file__])

View file

@ -661,13 +661,6 @@ def test_public_mcp_hub_returns_only_whitelisted_servers():
transport=MCPTransport.http,
available_on_public_internet=True,
)
hidden = MCPServer(
server_id="hidden",
name="hidden",
server_name="hidden",
transport=MCPTransport.http,
available_on_public_internet=True,
)
mock_manager = MagicMock()
mock_manager.get_public_mcp_servers.return_value = [listed]