scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-14 09:03:16 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 214

fix(mcp): scope /v1/mcp/access_groups to caller accessible servers (#46912)

Browse source
This commit is contained in:
Pyhroff 2026-11-07 12:51:58 +05:30 • committed by Pyhroff
parent ef1cb19351
commit f889b7e40c
3 changed files with 97 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/proxy/_lazy_openapi_snapshot.json
View file

@ -42714,7 +42714,7 @@
"paths": {
"/v1/mcp/access_groups": {
"get": {
"description": "Get all available MCP access groups from the database AND config",
"description": "Get available MCP access groups from the database AND config.\nAdmins receive the full catalog; non-admin callers are scoped to servers they have access to.",
"operationId": "get_mcp_access_groups_v1_mcp_access_groups_get",
"responses": {
"200": {

46
litellm/proxy/management_endpoints/mcp_management_endpoints.py
View file

@ -1112,34 +1112,44 @@ if MCP_AVAILABLE:
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Get all available MCP access groups from the database AND config
Get available MCP access groups from the database AND config.
Admins receive the full catalog; non-admin callers are scoped to servers they have access to.
"""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
from litellm.proxy.proxy_server import prisma_client
access_groups: Final = set()
access_groups: Final[set[str]] = set()
# Get from config-loaded servers
for server in global_mcp_server_manager.config_mcp_servers.values():
if server.access_groups:
access_groups.update(server.access_groups)
def _collect_groups(server_obj: Any) -> None:
for attr in ("mcp_access_groups", "access_groups"):
raw_groups = getattr(server_obj, attr, None)
if raw_groups and isinstance(raw_groups, (list, set, tuple)):
for g in raw_groups:
if g:
access_groups.add(str(g))
# Get from DB
if prisma_client is not None:
try:
mcp_servers: Final[Sequence[prisma_models.LiteLLM_MCPServerTable]] = await MCPServerRepository(
prisma_client
).table.find_many()
for server in mcp_servers:
if hasattr(server, "mcp_access_groups") and server.mcp_access_groups:
access_groups.update(server.mcp_access_groups)
except Exception as e:
verbose_proxy_logger.debug("Error getting MCP access groups: %s", e)
if user_api_key_has_admin_view(user_api_key_dict):
for cfg_server in global_mcp_server_manager.config_mcp_servers.values():
_collect_groups(cfg_server)
if prisma_client is not None:
try:
db_servers: Final[Sequence[prisma_models.LiteLLM_MCPServerTable]] = await MCPServerRepository(
prisma_client
).table.find_many()
for db_server in db_servers:
_collect_groups(db_server)
except Exception as e:
verbose_proxy_logger.debug("Error getting MCP access groups: %s", e)
else:
accessible_servers: Final = await _resolve_accessible_mcp_servers(user_api_key_dict)
for acc_server in accessible_servers:
_collect_groups(acc_server)
# Convert to sorted list
access_groups_list: Final = sorted(list(access_groups))
access_groups_list: Final[list[str]] = sorted(list(access_groups))
return {"access_groups": access_groups_list}
@router.get(

68
tests/unit/proxy/management_endpoints/test_mcp_management_endpoints.py
View file

@ -11704,3 +11704,71 @@ async def test_mcp_register_application_type_reaches_upstream_or_is_rejected(
assert posted["application_type"] == application_type
finally:
mgmt_endpoints._temporary_mcp_servers.pop(server.server_id, None)
@pytest.mark.asyncio
async def test_get_mcp_access_groups_admin_sees_all_groups() -> None:
admin_auth = generate_mock_user_api_key_auth(user_role=LitellmUserRoles.PROXY_ADMIN)
config_server = MagicMock()
config_server.access_groups = ["config-group-1"]
db_server = MagicMock()
db_server.mcp_access_groups = ["db-group-1"]
mock_manager = MagicMock()
mock_manager.config_mcp_servers = {"server-1": config_server}
mock_repo = MagicMock()
mock_repo.table.find_many = AsyncMock(return_value=[db_server])
with (
patch("litellm.proxy.management_endpoints.mcp_management_endpoints.user_api_key_has_admin_view", return_value=True),
patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager", mock_manager),
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.MCPServerRepository",
return_value=mock_repo,
),
):
result = await mgmt_endpoints.get_mcp_access_groups(user_api_key_dict=admin_auth)
assert result == {"access_groups": ["config-group-1", "db-group-1"]}
@pytest.mark.asyncio
async def test_get_mcp_access_groups_scoped_for_non_admin() -> None:
non_admin_auth = generate_mock_user_api_key_auth(user_role=LitellmUserRoles.INTERNAL_USER)
accessible_server = MagicMock()
accessible_server.mcp_access_groups = ["allowed-group"]
accessible_server.access_groups = None
with (
patch("litellm.proxy.management_endpoints.mcp_management_endpoints.user_api_key_has_admin_view", return_value=False),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._resolve_accessible_mcp_servers",
AsyncMock(return_value=[accessible_server]),
) as mock_resolve,
):
result = await mgmt_endpoints.get_mcp_access_groups(user_api_key_dict=non_admin_auth)
mock_resolve.assert_awaited_once_with(non_admin_auth)
assert result == {"access_groups": ["allowed-group"]}
@pytest.mark.asyncio
async def test_get_mcp_access_groups_non_admin_empty_when_no_accessible_servers() -> None:
non_admin_auth = generate_mock_user_api_key_auth(user_role=LitellmUserRoles.INTERNAL_USER)
with (
patch("litellm.proxy.management_endpoints.mcp_management_endpoints.user_api_key_has_admin_view", return_value=False),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._resolve_accessible_mcp_servers",
AsyncMock(return_value=[]),
),
):
result = await mgmt_endpoints.get_mcp_access_groups(user_api_key_dict=non_admin_auth)
assert result == {"access_groups": []}