feat(mcp/auth): additive key access-group grants + opt-in member assignment

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

@@ -566,7 +566,7 @@ async def get_allowed_mcp_servers(
                 )
             )
 
-            key_access_group_extras = (
+            key_access_group_grants = (
                 await MCPRequestHandler._get_key_access_group_mcp_server_extras(
                     user_api_key_auth
                 )
@@ -577,21 +577,22 @@ async def get_allowed_mcp_servers(
             #########################################################
             key_set = set(allowed_mcp_servers_for_key)
             team_set = set(allowed_mcp_servers_for_team)
-            extras_set = set(key_access_group_extras)
+            grants_set = set(key_access_group_grants)
 
-            has_lower_level_mcp_restrictions = bool(key_set or team_set or extras_set)
+            has_lower_level_mcp_restrictions = bool(key_set or team_set or grants_set)
 
-            # 1. Team-gated base scope.
+            # 1. Key/team ceiling. An empty set means "this level does not restrict".
             if not team_set:
                 base = key_set  # no team restriction
             elif not key_set:
                 base = team_set  # key has no own perms → inherits team
             else:
                 base = key_set & team_set  # both restrict → intersect
 
-            # 2. Extend with access-group extras (LIT-3189 — bypasses team
-            # ceiling, gated by group's assigned_team_ids / assigned_key_ids).
-            allowed_mcp_servers: List[str] = list(base | extras_set)
+            # 2. Add the key's access-group grants on top. These are additive:
+            # attaching a group to the key grants its servers regardless of the
+            # team ceiling.
+            allowed_mcp_servers: List[str] = list(base | grants_set)
 
             #########################################################
             # Check end_user permissions if end_user_id is set
@@ -890,11 +891,12 @@ async def _get_key_access_group_mcp_server_extras(
     ) -> List[str]:
         """
         Resolve the key's unified `access_group_ids` (LiteLLM_AccessGroupTable) to
-        MCP server IDs, gated by the access group's `assigned_team_ids` /
-        `assigned_key_ids`. These servers extend the team's MCP scope rather
-        than being capped by it. Tag-style `mcp_access_groups` (per-server tags)
-        are intentionally not handled here — they have no assignment fields and
-        remain subject to the team ceiling.
+        MCP server IDs as additive grants: a group attached to the key extends the
+        key's allowed servers on top of the key/team ceiling rather than being
+        capped by the team. Attaching the group to the key is itself the grant —
+        no `assigned_key_ids` / `assigned_team_ids` re-check. Tag-style
+        `mcp_access_groups` (per-server tags) live in the key's object_permission
+        scope, not here.
         """
         if user_api_key_auth is None:
             return []
@@ -903,21 +905,27 @@ async def _get_key_access_group_mcp_server_extras(
                 global_mcp_server_manager,
             )
             from litellm.proxy.auth.auth_checks import (
-                get_authorized_resources_from_key_access_groups,
+                _get_mcp_server_ids_from_access_groups,
+            )
+            from litellm.proxy.proxy_server import (
+                prisma_client,
+                proxy_logging_obj,
+                user_api_key_cache,
             )
 
-            raw_server_ids = await get_authorized_resources_from_key_access_groups(
-                valid_token=user_api_key_auth,
-                team_object=None,
-                resource_field="access_mcp_server_ids",
+            raw_server_ids = await _get_mcp_server_ids_from_access_groups(
+                access_group_ids=user_api_key_auth.access_group_ids or [],
+                prisma_client=prisma_client,
+                user_api_key_cache=user_api_key_cache,
+                proxy_logging_obj=proxy_logging_obj,
             )
             if not raw_server_ids:
                 return []
             # Permission entries may be server_ids OR names/aliases — expand to ids.
             return global_mcp_server_manager.expand_permission_list(raw_server_ids)
         except Exception as e:
             verbose_logger.warning(
-                f"Failed to get key access group MCP server extras: {str(e)}"
+                f"Failed to get key access group MCP server grants: {str(e)}"
             )
             return []
 
@@ -926,16 +934,13 @@ async def _get_allowed_mcp_servers_for_key(
         user_api_key_auth: Optional[UserAPIKeyAuth] = None,
     ) -> List[str]:
         """
-        Get allowed MCP servers for a key (the key's own scope).
+        Get the key's own MCP ceiling from its object_permission
+        (mcp_servers, tag-style mcp_access_groups, mcp_tool_permissions).
 
-        Unions two sources:
-        - Legacy key.object_permission (mcp_servers, mcp_access_groups,
-          mcp_tool_permissions).
-        - Unified key.access_group_ids → access_group.access_mcp_server_ids.
-          Mirrors the ungated fallback in can_key_call_model — the group is
-          attached to the key itself, so it grants the key's own scope (no
-          assigned_key_ids re-check). The gated, team-ceiling-busting override
-          lives in _get_key_access_group_mcp_server_extras.
+        Unified key.access_group_ids are NOT resolved here — they are additive
+        grants handled by _get_key_access_group_mcp_server_extras and unioned on
+        top of the key/team ceiling, so they must not enter this scope (which is
+        intersected against the team).
         """
         if user_api_key_auth is None:
             return []
@@ -944,7 +949,6 @@ async def _get_allowed_mcp_servers_for_key(
                 global_mcp_server_manager,
             )
             from litellm.proxy.auth.auth_checks import (
-                _get_mcp_server_ids_from_access_groups,
                 get_object_permission,
             )
             from litellm.proxy.proxy_server import (
@@ -953,19 +957,6 @@ async def _get_allowed_mcp_servers_for_key(
                 user_api_key_cache,
             )
 
-            # Unified key.access_group_ids → MCP servers (ungated: the group is
-            # attached to the key, so it grants the key's own scope). Entries in
-            # access_mcp_server_ids may be server_ids OR names/aliases, so expand
-            # to ids here — matching the legacy object_permission path below.
-            key_access_group_servers = global_mcp_server_manager.expand_permission_list(
-                await _get_mcp_server_ids_from_access_groups(
-                    access_group_ids=user_api_key_auth.access_group_ids or [],
-                    prisma_client=prisma_client,
-                    user_api_key_cache=user_api_key_cache,
-                    proxy_logging_obj=proxy_logging_obj,
-                )
-            )
-
             # Get key object permission (already loaded in main auth flow, or fetch from DB)
             key_object_permission = MCPRequestHandler._get_key_object_permission(
                 user_api_key_auth
@@ -983,7 +974,7 @@ async def _get_allowed_mcp_servers_for_key(
                     proxy_logging_obj=proxy_logging_obj,
                 )
             if key_object_permission is None:
-                return list(set(key_access_group_servers))
+                return []
 
             # Permission entries may be server_ids OR names/aliases — expand to ids.
             direct_mcp_servers = global_mcp_server_manager.expand_permission_list(
@@ -1005,12 +996,7 @@ async def _get_allowed_mcp_servers_for_key(
             )
 
             # Combine all lists
-            all_servers = (
-                direct_mcp_servers
-                + access_group_servers
-                + tool_perm_servers
-                + key_access_group_servers
-            )
+            all_servers = direct_mcp_servers + access_group_servers + tool_perm_servers
             return list(set(all_servers))
         except Exception as e:
             verbose_logger.warning(

litellm/proxy/_types.py

@@ -242,6 +242,11 @@ class KeyManagementRoutes(str, enum.Enum):
     TEAM_KEY_BULK_UPDATE = "/team/key/bulk_update"
     KEY_RESET_SPEND = "/key/{key_id}/reset_spend"
 
+    # Field-level opt-in permission (not a real HTTP route). When present in a
+    # team's `team_member_permissions`, non-admin members of that team may set
+    # `access_group_ids` on keys they create/update. Default-deny.
+    KEY_ACCESS_GROUP_ASSIGNMENT = "/key/access_group_assignment"
+
     # info and health routes
     KEY_INFO = "/key/info"
     KEY_HEALTH = "/key/health"
@@ -552,6 +557,7 @@ class LiteLLMRoutes(enum.Enum):
         KeyManagementRoutes.SPEND_LOGS_V2.value,
         KeyManagementRoutes.KEY_RESET_SPEND.value,
         KeyManagementRoutes.KEY_ALIASES.value,
+        KeyManagementRoutes.KEY_ACCESS_GROUP_ASSIGNMENT.value,
     ]
 
     management_routes = (

litellm/proxy/management_endpoints/key_management_endpoints.py

@@ -325,6 +325,14 @@ def _team_key_generation_check(
         _team_key_generation.get("required_params"),
     )
 
+    # Field-level opt-in: non-admin members may only assign access groups when
+    # the team has enabled KEY_ACCESS_GROUP_ASSIGNMENT.
+    TeamMemberPermissionChecks.enforce_member_can_assign_access_groups(
+        user_api_key_dict=user_api_key_dict,
+        team_table=team_table,
+        access_group_ids=data.access_group_ids,
+    )
+
     return True
 
 
@@ -2267,6 +2275,14 @@ async def _validate_update_key_data(
                 detail=f"Team not found for team_id={data.team_id}. Non-admin users cannot set keys to non-existent teams.",
             )
 
+        # Field-level opt-in: non-admin members may only assign access groups when
+        # the team has enabled KEY_ACCESS_GROUP_ASSIGNMENT.
+        TeamMemberPermissionChecks.enforce_member_can_assign_access_groups(
+            user_api_key_dict=user_api_key_dict,
+            team_table=team_obj,
+            access_group_ids=data.access_group_ids,
+        )
+
         if team_obj is not None:
             await _check_team_key_limits(
                 team_table=team_obj,
@@ -4511,6 +4527,23 @@ async def regenerate_key_fn(  # noqa: PLR0915
                 detail={"error": "You are not authorized to regenerate this key"},
             )
 
+        # Gate access_group_ids on regenerate, same as /key/generate and
+        # /key/update. Use the existing key's team since the body may omit it.
+        if data is not None and data.access_group_ids:
+            regenerate_team_table: Optional[LiteLLM_TeamTableCachedObj] = None
+            if _key_in_db.team_id is not None:
+                regenerate_team_table = await get_team_object(
+                    team_id=_key_in_db.team_id,
+                    prisma_client=prisma_client,
+                    user_api_key_cache=user_api_key_cache,
+                    check_db_only=True,
+                )
+            TeamMemberPermissionChecks.enforce_member_can_assign_access_groups(
+                user_api_key_dict=user_api_key_dict,
+                team_table=regenerate_team_table,
+                access_group_ids=data.access_group_ids,
+            )
+
         verbose_proxy_logger.info(
             "Key regeneration requested: key_alias=%s",
             getattr(_key_in_db, "key_alias", None),

litellm/proxy/management_helpers/team_member_permission_checks.py

@@ -154,6 +154,70 @@ def does_team_member_have_permissions_for_endpoint(
 
         return True
 
+    @staticmethod
+    def enforce_member_can_assign_access_groups(
+        user_api_key_dict: UserAPIKeyAuth,
+        team_table: Optional[LiteLLM_TeamTableCachedObj],
+        access_group_ids: Optional[List[str]],
+    ) -> None:
+        """
+        Field-level opt-in gate: a non-admin team member may only set
+        `access_group_ids` on a (team) key if their team has opted in by adding
+        `KEY_ACCESS_GROUP_ASSIGNMENT` to `team_member_permissions`.
+
+        Bypassed for proxy admins, team admins, and personal (non-team) keys.
+        Default-deny: members cannot self-assign access groups until enabled.
+
+        Raises HTTPException(403) when a gated member attempts the assignment.
+        """
+        from fastapi import HTTPException
+
+        from litellm.proxy.management_endpoints.key_management_endpoints import (
+            _get_user_in_team,
+        )
+
+        # No-op when the request does not assign any access groups.
+        if not access_group_ids:
+            return
+
+        # Proxy admins always bypass.
+        if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value:
+            return
+
+        # Personal (non-team) keys are out of scope for team-member gating.
+        if team_table is None:
+            return
+
+        team_member_object = _get_user_in_team(
+            team_table=team_table, user_id=user_api_key_dict.user_id
+        )
+
+        # Team admins always bypass (consistent with other member-permission checks).
+        if team_member_object is not None and team_member_object.role == "admin":
+            return
+
+        permissions = (
+            TeamMemberPermissionChecks._get_list_of_route_enum_as_str(
+                TeamMemberPermissionChecks.get_permissions_for_team_member(
+                    team_member_object=team_member_object,
+                    team_table=team_table,
+                )
+            )
+            if team_member_object is not None
+            else []
+        )
+
+        if KeyManagementRoutes.KEY_ACCESS_GROUP_ASSIGNMENT.value not in permissions:
+            raise HTTPException(
+                status_code=403,
+                detail=(
+                    "Team members cannot assign access groups to keys for team "
+                    f"{team_table.team_id}. Ask a team or proxy admin to enable the "
+                    f"'{KeyManagementRoutes.KEY_ACCESS_GROUP_ASSIGNMENT.value}' team "
+                    "member permission to allow this."
+                ),
+            )
+
     @staticmethod
     async def user_belongs_to_keys_team(
         user_api_key_dict: UserAPIKeyAuth,